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 

6 

7import couchers.phone.sms 

8from couchers.config import config 

9from couchers.crypto import random_hex 

10from couchers.db import session_scope 

11from couchers.models import SMS, User 

12from couchers.proto import account_pb2, api_pb2 

13from couchers.sql import couchers_select as select 

14from couchers.utils import now 

15from tests.test_fixtures import ( # noqa 

16 account_session, 

17 api_session, 

18 db, 

19 generate_user, 

20 notifications_session, 

21 process_jobs, 

22 push_collector, 

23 testconfig, 

24) 

25 

26 

27@pytest.fixture(autouse=True) 

28def _(testconfig): 

29 pass 

30 

31 

32def test_ChangePhone(db, monkeypatch, push_collector): 

33 user, token = generate_user() 

34 user_id = user.id 

35 

36 with account_session(token) as account: 

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

38 assert res.phone == "" 

39 

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

41 

42 # Try with a too long number 

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

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

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

46 

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

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

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

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

51 

52 # Test with operator not supported by SMS backend 

53 def deny_operator(phone, message): 

54 assert phone == "+46701740605" 

55 return "unsupported operator" 

56 

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

58 

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

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

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

62 

63 # Test with successfully sent SMS 

64 def succeed(phone, message): 

65 assert phone == "+46701740605" 

66 return "success" 

67 

68 push_collector.assert_user_has_count(user_id, 0) 

69 

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

71 

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

73 

74 with session_scope() as session: 

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

76 assert user.phone == "+46701740605" 

77 assert len(user.phone_verification_token) == 6 

78 

79 process_jobs() 

80 push_collector.assert_user_has_single_matching( 

81 user_id, 

82 title="Phone verification started", 

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

84 ) 

85 

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

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

88 assert res.phone == "+46701740605" 

89 assert not res.phone_verified 

90 

91 # Remove phone number 

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

93 

94 with session_scope() as session: 

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

96 assert user.phone is None 

97 assert user.phone_verification_token is None 

98 

99 

100def test_ChangePhone_ratelimit(db, monkeypatch): 

101 user, token = generate_user() 

102 user_id = user.id 

103 with account_session(token) as account: 

104 

105 def succeed(phone, message): 

106 return "success" 

107 

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

109 

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

111 

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

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

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

115 

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

117 with session_scope() as session: 

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

119 assert user.phone == "+46701740605" 

120 assert len(user.phone_verification_token) == 6 

121 

122 

123def test_VerifyPhone(push_collector): 

124 user, token = generate_user() 

125 user_id = user.id 

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 user = session.execute(select(User).where(User.id == user_id)).scalar_one() 

136 user.phone = "+46701740605" 

137 user.phone_verification_token = "111112" 

138 user.phone_verification_sent = now() 

139 

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

141 

142 process_jobs() 

143 push_collector.assert_user_has_single_matching( 

144 user_id, 

145 title="Phone successfully verified", 

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

147 ) 

148 

149 res = api.GetUser(api_pb2.GetUserReq(user=str(user_id))) 

150 assert res.verification == 1.0 

151 

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

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

154 assert res.phone == "+46701740605" 

155 

156 

157def test_VerifyPhone_antibrute(): 

158 user, token = generate_user() 

159 user_id = user.id 

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

161 with session_scope() as session: 

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

163 user.phone_verification_token = "111112" 

164 user.phone_verification_sent = now() 

165 user.phone = "+46701740605" 

166 

167 for _ in range(10): 

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

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

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

171 break 

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

173 

174 

175def test_phone_uniqueness(monkeypatch): 

176 user1, token1 = generate_user() 

177 user2, token2 = generate_user() 

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

179 

180 def succeed(phone, message): 

181 return "success" 

182 

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

184 

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

186 with session_scope() as session: 

187 user = session.execute(select(User).where(User.id == user1.id)).scalar_one() 

188 token = user.phone_verification_token 

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 user = session.execute(select(User).where(User.id == user2.id)).scalar_one() 

208 token = user.phone_verification_token 

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

210 

211 # number gets wiped when it's stolen 

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

213 assert not res.phone 

214 assert not res.phone_verified 

215 

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

217 assert res.phone == "+46701740605" 

218 assert res.phone_verified 

219 

220 

221def test_send_sms(db, monkeypatch): 

222 new_config = config.copy() 

223 new_config["ENABLE_SMS"] = True 

224 new_config["SMS_SENDER_ID"] = "CouchersOrg" 

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

226 

227 msg_id = random_hex() 

228 

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

230 sns = Mock() 

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

232 mock.client.return_value = sns 

233 

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

235 

236 mock.client.assert_called_once_with("sns") 

237 sns.publish.assert_called_once_with( 

238 PhoneNumber="+46701740605", 

239 Message="Testing SMS message", 

240 MessageAttributes={ 

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

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

243 }, 

244 ) 

245 

246 with session_scope() as session: 

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

248 assert sms.message_id == msg_id 

249 assert sms.sms_sender_id == "CouchersOrg" 

250 assert sms.number == "+46701740605" 

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

252 

253 

254def test_send_sms_disabled(db): 

255 assert not config["ENABLE_SMS"] 

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

257 

258 

259def test_sms_verification_no_donation(): 

260 user, token = generate_user(last_donated=None) 

261 with account_session(token) as account: 

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

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

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

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