Coverage for app/backend/src/tests/test_verification.py: 99%

1from unittest.mock import Mock, patch 

2 

3import grpc 

4import pytest 

5from google.protobuf import empty_pb2 

6from sqlalchemy import select, 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.utils import now 

15from tests.fixtures.db import generate_user 

16from tests.fixtures.misc import PushCollector, process_jobs 

17from tests.fixtures.sessions import account_session, api_session 

18 

19 

20def test_ChangePhone(db, monkeypatch, push_collector: PushCollector): 

21 user, token = generate_user() 

22 user_id = user.id 

23 

24 with account_session(token) as account: 

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

26 assert res.phone == "" 

27 

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

29 

30 # Try with a too long number 

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

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

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

34 

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

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

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

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

39 

40 # Test with operator not supported by SMS backend 

41 def deny_operator(phone, message): 

42 assert phone == "+46701740605" 

43 return "unsupported operator" 

44 

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

46 

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

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

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

50 

51 # Test with successfully sent SMS 

52 def succeed(phone, message): 

53 assert phone == "+46701740605" 

54 return "success" 

55 

56 assert push_collector.count_for_user(user_id) == 0 

57 

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

59 

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

61 

62 with session_scope() as session: 

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

64 assert user.phone == "+46701740605" 

65 assert user.phone_verification_token 

66 assert len(user.phone_verification_token) == 6 

67 

68 process_jobs() 

69 push = push_collector.pop_for_user(user_id, last=True) 

70 assert push.content.title == "Phone verification started" 

71 assert push.content.body == "You started phone number verification with the number +46 70 174 06 05." 

72 

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

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

75 assert res.phone == "+46701740605" 

76 assert not res.phone_verified 

77 

78 # Remove phone number 

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

80 

81 with session_scope() as session: 

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

83 assert user.phone is None 

84 assert user.phone_verification_token is None 

85 

86 

87def test_ChangePhone_ratelimit(db, monkeypatch): 

88 user, token = generate_user() 

89 user_id = user.id 

90 with account_session(token) as account: 

91 

92 def succeed(phone, message): 

93 return "success" 

94 

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

96 

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

98 

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

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

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

102 

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

104 with session_scope() as session: 

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

106 assert user.phone == "+46701740605" 

107 assert user.phone_verification_token 

108 assert len(user.phone_verification_token) == 6 

109 

110 

111def test_VerifyPhone(push_collector: PushCollector): 

112 user, token = generate_user() 

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

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

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

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

117 

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

119 assert res.verification == 0.0 

120 

121 with session_scope() as session: 

122 session.execute( 

123 update(User) 

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

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

126 ) 

127 

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

129 

130 process_jobs() 

131 push = push_collector.pop_for_user(user.id, last=True) 

132 assert push.content.title == "Phone verification completed" 

133 assert push.content.body == "Your phone number was successfully verified as +46 70 174 06 05." 

134 

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

136 assert res.verification == 1.0 

137 

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

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

140 assert res.phone == "+46701740605" 

141 

142 

143def test_VerifyPhone_antibrute(): 

144 user, token = generate_user( 

145 phone_verification_token="111112", 

146 phone_verification_sent=now(), 

147 phone="+46701740605", 

148 ) 

149 

150 with account_session(token) as account: 

151 for _ in range(10): 151 ↛ 156line 151 didn't jump to line 156 because the loop on line 151 didn't complete

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

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

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

155 break 

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

157 

158 

159def test_phone_uniqueness(monkeypatch): 

160 user1, token1 = generate_user() 

161 user2, token2 = generate_user() 

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

163 

164 def succeed(phone, message): 

165 return "success" 

166 

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

168 

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

170 with session_scope() as session: 

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

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

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

174 assert res.phone == "+46701740605" 

175 assert res.phone_verified 

176 

177 # Let user2 steal user1:s phone number 

178 

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

180 

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

182 assert res.phone == "+46701740605" 

183 assert res.phone_verified 

184 

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

186 assert res.phone == "+46701740605" 

187 assert not res.phone_verified 

188 

189 with session_scope() as session: 

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

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

192 

193 # number gets wiped when it's stolen 

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

195 assert not res.phone 

196 assert not res.phone_verified 

197 

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

199 assert res.phone == "+46701740605" 

200 assert res.phone_verified 

201 

202 

203def test_send_sms(db, monkeypatch): 

204 new_config = config.copy() 

205 new_config.SMS_SENDER_ID = "CouchersOrg" 

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

207 

208 msg_id = random_hex() 

209 

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

211 sns = Mock() 

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

213 mock.client.return_value = sns 

214 

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

216 

217 mock.client.assert_called_once_with("sns") 

218 sns.publish.assert_called_once_with( 

219 PhoneNumber="+46701740605", 

220 Message="Testing SMS message", 

221 MessageAttributes={ 

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

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

224 }, 

225 ) 

226 

227 with session_scope() as session: 

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

229 assert sms.message_id == msg_id 

230 assert sms.sms_sender_id == "CouchersOrg" 

231 assert sms.number == "+46701740605" 

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

233 

234 

235def test_ChangePhone_sms_disabled(db, feature_flags): 

236 feature_flags.set("sms_enabled", False) 

237 user, token = generate_user() 

238 with account_session(token) as account: 

239 # Setting a new number requires sending an SMS, which is gated. 

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

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

242 assert e.value.code() == grpc.StatusCode.UNAVAILABLE 

243 

244 # Removing a number doesn't send an SMS, so it's still allowed. 

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

246 

247 

248def test_sms_verification_no_donation(): 

249 user, token = generate_user(last_donated=None) 

250 with account_session(token) as account: 

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

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

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

254 assert e.value.details() == "You need to donate to Couchers.org before you can verify your phone number."