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 import errors 

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.sql import couchers_select as select 

14from couchers.utils import now 

15from proto import account_pb2, api_pb2 

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 user_id = user.id 

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

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

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

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

131 

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

133 assert res.verification == 0.0 

134 

135 with session_scope() as session: 

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

137 user.phone = "+46701740605" 

138 user.phone_verification_token = "111112" 

139 user.phone_verification_sent = now() 

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 user_id = user.id 

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

162 with session_scope() as session: 

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

164 user.phone_verification_token = "111112" 

165 user.phone_verification_sent = now() 

166 user.phone = "+46701740605" 

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

189 token = user.phone_verification_token 

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

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

192 assert res.phone == "+46701740605" 

193 assert res.phone_verified 

194 

195 # Let user2 steal user1:s phone number 

196 

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

198 

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

200 assert res.phone == "+46701740605" 

201 assert res.phone_verified 

202 

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

204 assert res.phone == "+46701740605" 

205 assert not res.phone_verified 

206 

207 with session_scope() as session: 

208 user = session.execute(select(User).where(User.id == user2.id)).scalar_one() 

209 token = user.phone_verification_token 

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

211 

212 # number gets wiped when it's stolen 

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

214 assert not res.phone 

215 assert not res.phone_verified 

216 

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

218 assert res.phone == "+46701740605" 

219 assert res.phone_verified 

220 

221 

222def test_send_sms(db, monkeypatch): 

223 new_config = config.copy() 

224 new_config["ENABLE_SMS"] = True 

225 new_config["SMS_SENDER_ID"] = "CouchersOrg" 

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

227 

228 msg_id = random_hex() 

229 

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

231 sns = Mock() 

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

233 mock.client.return_value = sns 

234 

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

236 

237 mock.client.assert_called_once_with("sns") 

238 sns.publish.assert_called_once_with( 

239 PhoneNumber="+46701740605", 

240 Message="Testing SMS message", 

241 MessageAttributes={ 

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

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

244 }, 

245 ) 

246 

247 with session_scope() as session: 

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

249 assert sms.message_id == msg_id 

250 assert sms.sms_sender_id == "CouchersOrg" 

251 assert sms.number == "+46701740605" 

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

253 

254 

255def test_send_sms_disabled(db): 

256 assert not config["ENABLE_SMS"] 

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

258 

259 

260def test_sms_verification_no_donation(): 

261 user, token = generate_user(has_donated=False) 

262 with account_session(token) as account: 

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

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

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

266 assert e.value.details() == errors.NOT_DONATED