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 

20@pytest.fixture(autouse=True) 

21def _(testconfig): 

22 pass 

23 

24 

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

26 user, token = generate_user() 

27 user_id = user.id 

28 

29 with account_session(token) as account: 

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

31 assert res.phone == "" 

32 

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

34 

35 # Try with a too long number 

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

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

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

39 

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

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

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

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

44 

45 # Test with operator not supported by SMS backend 

46 def deny_operator(phone, message): 

47 assert phone == "+46701740605" 

48 return "unsupported operator" 

49 

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

51 

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

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

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

55 

56 # Test with successfully sent SMS 

57 def succeed(phone, message): 

58 assert phone == "+46701740605" 

59 return "success" 

60 

61 assert push_collector.count_for_user(user_id) == 0 

62 

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

64 

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

66 

67 with session_scope() as session: 

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

69 assert user.phone == "+46701740605" 

70 assert user.phone_verification_token 

71 assert len(user.phone_verification_token) == 6 

72 

73 process_jobs() 

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

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

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

77 

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

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

80 assert res.phone == "+46701740605" 

81 assert not res.phone_verified 

82 

83 # Remove phone number 

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

85 

86 with session_scope() as session: 

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

88 assert user.phone is None 

89 assert user.phone_verification_token is None 

90 

91 

92def test_ChangePhone_ratelimit(db, monkeypatch): 

93 user, token = generate_user() 

94 user_id = user.id 

95 with account_session(token) as account: 

96 

97 def succeed(phone, message): 

98 return "success" 

99 

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

101 

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

103 

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

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

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

107 

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

109 with session_scope() as session: 

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

111 assert user.phone == "+46701740605" 

112 assert user.phone_verification_token 

113 assert len(user.phone_verification_token) == 6 

114 

115 

116def test_VerifyPhone(push_collector: PushCollector): 

117 user, token = generate_user() 

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

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

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

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

122 

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

124 assert res.verification == 0.0 

125 

126 with session_scope() as session: 

127 session.execute( 

128 update(User) 

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

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

131 ) 

132 

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

134 

135 process_jobs() 

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

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

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

139 

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

141 assert res.verification == 1.0 

142 

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

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

145 assert res.phone == "+46701740605" 

146 

147 

148def test_VerifyPhone_antibrute(): 

149 user, token = generate_user( 

150 phone_verification_token="111112", 

151 phone_verification_sent=now(), 

152 phone="+46701740605", 

153 ) 

154 

155 with account_session(token) as account: 

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

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

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

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

160 break 

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

162 

163 

164def test_phone_uniqueness(monkeypatch): 

165 user1, token1 = generate_user() 

166 user2, token2 = generate_user() 

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

168 

169 def succeed(phone, message): 

170 return "success" 

171 

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

173 

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

175 with session_scope() as session: 

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

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

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

179 assert res.phone == "+46701740605" 

180 assert res.phone_verified 

181 

182 # Let user2 steal user1:s phone number 

183 

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

185 

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

187 assert res.phone == "+46701740605" 

188 assert res.phone_verified 

189 

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

191 assert res.phone == "+46701740605" 

192 assert not res.phone_verified 

193 

194 with session_scope() as session: 

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

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

197 

198 # number gets wiped when it's stolen 

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

200 assert not res.phone 

201 assert not res.phone_verified 

202 

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

204 assert res.phone == "+46701740605" 

205 assert res.phone_verified 

206 

207 

208def test_send_sms(db, monkeypatch): 

209 new_config = config.copy() 

210 new_config["ENABLE_SMS"] = True 

211 new_config["SMS_SENDER_ID"] = "CouchersOrg" 

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

213 

214 msg_id = random_hex() 

215 

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

217 sns = Mock() 

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

219 mock.client.return_value = sns 

220 

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

222 

223 mock.client.assert_called_once_with("sns") 

224 sns.publish.assert_called_once_with( 

225 PhoneNumber="+46701740605", 

226 Message="Testing SMS message", 

227 MessageAttributes={ 

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

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

230 }, 

231 ) 

232 

233 with session_scope() as session: 

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

235 assert sms.message_id == msg_id 

236 assert sms.sms_sender_id == "CouchersOrg" 

237 assert sms.number == "+46701740605" 

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

239 

240 

241def test_send_sms_disabled(db): 

242 assert not config["ENABLE_SMS"] 

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

244 

245 

246def test_sms_verification_no_donation(): 

247 user, token = generate_user(last_donated=None) 

248 with account_session(token) as account: 

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

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

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

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