Coverage for src/tests/test_verification.py: 100%

1from unittest.mock import Mock, patch 

2 

3import grpc 

4import pytest 

5from google.protobuf import empty_pb2 

6from sqlalchemy import update 

7 

8import couchers.phone.sms 

9from couchers.config import config 

10from couchers.crypto import random_hex 

11from couchers.db import session_scope 

12from couchers.models import SMS, User 

13from couchers.proto import account_pb2, api_pb2 

14from couchers.sql import couchers_select as select 

15from couchers.utils import now 

16from tests.test_fixtures import ( # noqa 

17 account_session, 

18 api_session, 

19 db, 

20 generate_user, 

21 notifications_session, 

22 process_jobs, 

23 push_collector, 

24 testconfig, 

25) 

26 

27 

28@pytest.fixture(autouse=True) 

29def _(testconfig): 

30 pass 

31 

32 

33def test_ChangePhone(db, monkeypatch, push_collector): 

34 user, token = generate_user() 

35 user_id = user.id 

36 

37 with account_session(token) as account: 

38 res = account.GetAccountInfo(empty_pb2.Empty()) 

39 assert res.phone == "" 

40 

41 monkeypatch.setattr(couchers.phone.sms, "send_sms", pytest.fail) 

42 

43 # Try with a too long number 

44 with pytest.raises(grpc.RpcError) as e: 

45 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+4670174060666666")) 

46 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT 

47 

48 # try to see if one digit too much is caught before attempting to send sms 

49 with pytest.raises(grpc.RpcError) as e: 

50 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+467017406066")) 

51 assert e.value.code() == grpc.StatusCode.UNIMPLEMENTED 

52 

53 # Test with operator not supported by SMS backend 

54 def deny_operator(phone, message): 

55 assert phone == "+46701740605" 

56 return "unsupported operator" 

57 

58 monkeypatch.setattr(couchers.phone.sms, "send_sms", deny_operator) 

59 

60 with pytest.raises(grpc.RpcError) as e: 

61 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

62 assert e.value.code() == grpc.StatusCode.UNIMPLEMENTED 

63 

64 # Test with successfully sent SMS 

65 def succeed(phone, message): 

66 assert phone == "+46701740605" 

67 return "success" 

68 

69 push_collector.assert_user_has_count(user_id, 0) 

70 

71 monkeypatch.setattr(couchers.phone.sms, "send_sms", succeed) 

72 

73 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

74 

75 with session_scope() as session: 

76 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

77 assert user.phone == "+46701740605" 

78 assert len(user.phone_verification_token) == 6 

79 

80 process_jobs() 

81 push_collector.assert_user_has_single_matching( 

82 user_id, 

83 title="Phone verification started", 

84 body="You started phone number verification with the number +46 70 174 06 05.", 

85 ) 

86 

87 # Phone number should show up but not be verified in your profile settings 

88 res = account.GetAccountInfo(empty_pb2.Empty()) 

89 assert res.phone == "+46701740605" 

90 assert not res.phone_verified 

91 

92 # Remove phone number 

93 account.ChangePhone(account_pb2.ChangePhoneReq(phone="")) 

94 

95 with session_scope() as session: 

96 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

97 assert user.phone is None 

98 assert user.phone_verification_token is None 

99 

100 

101def test_ChangePhone_ratelimit(db, monkeypatch): 

102 user, token = generate_user() 

103 user_id = user.id 

104 with account_session(token) as account: 

105 

106 def succeed(phone, message): 

107 return "success" 

108 

109 monkeypatch.setattr(couchers.phone.sms, "send_sms", succeed) 

110 

111 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

112 

113 with pytest.raises(grpc.RpcError) as e: 

114 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740606")) 

115 assert e.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED 

116 

117 # Check that an earlier phone number/verification status is still saved 

118 with session_scope() as session: 

119 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

120 assert user.phone == "+46701740605" 

121 assert len(user.phone_verification_token) == 6 

122 

123 

124def test_VerifyPhone(push_collector): 

125 user, token = generate_user() 

126 with account_session(token) as account, api_session(token) as api: 

127 with pytest.raises(grpc.RpcError) as e: 

128 account.VerifyPhone(account_pb2.VerifyPhoneReq(token="123455")) 

129 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

130 

131 res = api.GetUser(api_pb2.GetUserReq(user=str(user.id))) 

132 assert res.verification == 0.0 

133 

134 with session_scope() as session: 

135 session.execute( 

136 update(User) 

137 .where(User.id == user.id) 

138 .values(phone_verification_token="111112", phone_verification_sent=now(), phone="+46701740605") 

139 ) 

140 

141 account.VerifyPhone(account_pb2.VerifyPhoneReq(token="111112")) 

142 

143 process_jobs() 

144 push_collector.assert_user_has_single_matching( 

145 user.id, 

146 title="Phone successfully verified", 

147 body="Your phone was successfully verified as +46 70 174 06 05 on Couchers.org.", 

148 ) 

149 

150 res = api.GetUser(api_pb2.GetUserReq(user=str(user.id))) 

151 assert res.verification == 1.0 

152 

153 # Phone number should finally show up on in your profile settings 

154 res = account.GetAccountInfo(empty_pb2.Empty()) 

155 assert res.phone == "+46701740605" 

156 

157 

158def test_VerifyPhone_antibrute(): 

159 user, token = generate_user() 

160 with account_session(token) as account: 

161 with session_scope() as session: 

162 session.execute( 

163 update(User) 

164 .where(User.id == user.id) 

165 .values(phone_verification_token="111112", phone_verification_sent=now(), phone="+46701740605") 

166 ) 

167 

168 for _ in range(10): 

169 with pytest.raises(grpc.RpcError) as e: 

170 account.VerifyPhone(account_pb2.VerifyPhoneReq(token="123455")) 

171 if e.value.code() != grpc.StatusCode.NOT_FOUND: 

172 break 

173 assert e.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED 

174 

175 

176def test_phone_uniqueness(monkeypatch): 

177 user1, token1 = generate_user() 

178 user2, token2 = generate_user() 

179 with account_session(token1) as account1, account_session(token2) as account2: 

180 

181 def succeed(phone, message): 

182 return "success" 

183 

184 monkeypatch.setattr(couchers.phone.sms, "send_sms", succeed) 

185 

186 account1.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

187 with session_scope() as session: 

188 token = session.execute(select(User.phone_verification_token).where(User.id == user1.id)).scalar_one() 

189 account1.VerifyPhone(account_pb2.VerifyPhoneReq(token=token)) 

190 res = account1.GetAccountInfo(empty_pb2.Empty()) 

191 assert res.phone == "+46701740605" 

192 assert res.phone_verified 

193 

194 # Let user2 steal user1:s phone number 

195 

196 account2.ChangePhone(account_pb2.ChangePhoneReq(phone="+46701740605")) 

197 

198 res = account1.GetAccountInfo(empty_pb2.Empty()) 

199 assert res.phone == "+46701740605" 

200 assert res.phone_verified 

201 

202 res = account2.GetAccountInfo(empty_pb2.Empty()) 

203 assert res.phone == "+46701740605" 

204 assert not res.phone_verified 

205 

206 with session_scope() as session: 

207 token = session.execute(select(User.phone_verification_token).where(User.id == user2.id)).scalar_one() 

208 account2.VerifyPhone(account_pb2.VerifyPhoneReq(token=token)) 

209 

210 # number gets wiped when it's stolen 

211 res = account1.GetAccountInfo(empty_pb2.Empty()) 

212 assert not res.phone 

213 assert not res.phone_verified 

214 

215 res = account2.GetAccountInfo(empty_pb2.Empty()) 

216 assert res.phone == "+46701740605" 

217 assert res.phone_verified 

218 

219 

220def test_send_sms(db, monkeypatch): 

221 new_config = config.copy() 

222 new_config["ENABLE_SMS"] = True 

223 new_config["SMS_SENDER_ID"] = "CouchersOrg" 

224 monkeypatch.setattr(couchers.phone.sms, "config", new_config) 

225 

226 msg_id = random_hex() 

227 

228 with patch("couchers.phone.sms.boto3") as mock: 

229 sns = Mock() 

230 sns.publish.return_value = {"MessageId": msg_id} 

231 mock.client.return_value = sns 

232 

233 assert couchers.phone.sms.send_sms("+46701740605", "Testing SMS message") == "success" 

234 

235 mock.client.assert_called_once_with("sns") 

236 sns.publish.assert_called_once_with( 

237 PhoneNumber="+46701740605", 

238 Message="Testing SMS message", 

239 MessageAttributes={ 

240 "AWS.SNS.SMS.SMSType": {"DataType": "String", "StringValue": "Transactional"}, 

241 "AWS.SNS.SMS.SenderID": {"DataType": "String", "StringValue": "CouchersOrg"}, 

242 }, 

243 ) 

244 

245 with session_scope() as session: 

246 sms = session.execute(select(SMS)).scalar_one() 

247 assert sms.message_id == msg_id 

248 assert sms.sms_sender_id == "CouchersOrg" 

249 assert sms.number == "+46701740605" 

250 assert sms.message == "Testing SMS message" 

251 

252 

253def test_send_sms_disabled(db): 

254 assert not config["ENABLE_SMS"] 

255 assert couchers.phone.sms.send_sms("+46701740605", "Testing SMS message") == "SMS not enabled." 

256 

257 

258def test_sms_verification_no_donation(): 

259 user, token = generate_user(last_donated=None) 

260 with account_session(token) as account: 

261 with pytest.raises(grpc.RpcError) as e: 

262 account.ChangePhone(account_pb2.ChangePhoneReq(phone="+467017406066")) 

263 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION 

264 assert e.value.details() == "Please complete donation to get phone verified."