Coverage for app/backend/src/tests/test_strong_verification.py: 98%

1import json 

2from datetime import date, timedelta 

3from unittest.mock import ANY, patch 

4from urllib.parse import urlencode 

5 

6import grpc 

7import pytest 

8from google.protobuf import empty_pb2 

9from sqlalchemy import select, update 

10from sqlalchemy.sql import or_ 

11 

12import couchers.jobs.handlers 

13import couchers.servicers.account 

14from couchers.config import config 

15from couchers.crypto import asym_decrypt, b64encode_unpadded 

16from couchers.db import session_scope 

17from couchers.jobs.handlers import update_badges 

18from couchers.jobs.worker import process_job 

19from couchers.materialized_views import refresh_materialized_views_rapid 

20from couchers.models import ( 

21 PassportSex, 

22 StrongVerificationAttempt, 

23 StrongVerificationAttemptStatus, 

24 StrongVerificationCallbackEvent, 

25 User, 

26) 

27from couchers.proto import account_pb2, admin_pb2, api_pb2 

28from couchers.proto.google.api import httpbody_pb2 

29from tests.fixtures.db import generate_user 

30from tests.fixtures.misc import PushCollector 

31from tests.fixtures.sessions import account_session, api_session, real_admin_session, real_iris_session 

32 

33 

34@pytest.fixture(autouse=True) 

35def _(testconfig): 

36 pass 

37 

38 

39class MockSVFlow: 

40 """Simulates the strong verification flow, including all IRIS callbacks.""" 

41 

42 def __init__(self, user: User, token: str, verification_id: int = 5731012934821983) -> None: 

43 self.user = user 

44 self.token = token 

45 self.verification_id = verification_id 

46 

47 iris_token_data = { 

48 "merchant_id": 5731012934821982, 

49 "session_id": self.verification_id, 

50 "seed": 1674246339, 

51 "face_verification": False, 

52 "host": "https://passportreader.app", 

53 } 

54 self.iris_token = b64encode_unpadded(json.dumps(iris_token_data).encode("utf8")) 

55 

56 with account_session(self.token) as account: 

57 # start by initiation 

58 with patch("couchers.servicers.account.requests.post") as mock: 

59 json_resp1 = { 

60 "id": self.verification_id, 

61 "token": self.iris_token, 

62 } 

63 mock.return_value = type( 

64 "__MockResponse", 

65 (), 

66 { 

67 "status_code": 200, 

68 "text": json.dumps(json_resp1), 

69 "json": lambda: json_resp1, 

70 }, 

71 ) 

72 res = account.InitiateStrongVerification(empty_pb2.Empty()) 

73 

74 mock.assert_called_once_with( 

75 "https://passportreader.app/api/v1/session.create", 

76 auth=("dummy_pubkey", "dummy_secret"), 

77 json={ 

78 "callback_url": "http://localhost:8888/iris/webhook", 

79 "face_verification": False, 

80 "passport_only": True, 

81 "reference": ANY, 

82 }, 

83 timeout=10, 

84 verify="/etc/ssl/certs/ca-certificates.crt", 

85 ) 

86 self.reference_data = mock.call_args.kwargs["json"]["reference"] 

87 self.verification_attempt_token = res.verification_attempt_token 

88 return_url = f"http://localhost:3000/complete-strong-verification?verification_attempt_token={self.verification_attempt_token}" 

89 assert res.redirect_url == "https://passportreader.app/open?" + urlencode( 

90 {"token": self.iris_token, "redirect_url": return_url} 

91 ) 

92 

93 assert ( 

94 account.GetStrongVerificationAttemptStatus( 

95 account_pb2.GetStrongVerificationAttemptStatusReq( 

96 verification_attempt_token=self.verification_attempt_token 

97 ) 

98 ).status 

99 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_TO_OPEN_APP 

100 ) 

101 

102 def process_iris_callbacks( 

103 self, 

104 *, 

105 nationality: str = "US", 

106 sex: PassportSex | None = None, 

107 date_of_birth: date | None = None, 

108 document_type: str = "PASSPORT", 

109 document_number: str = "31195855", 

110 document_expiry: date | None = None, 

111 expected_status: StrongVerificationAttemptStatus = StrongVerificationAttemptStatus.succeeded, 

112 ) -> None: 

113 self.process_iris_initiated_callback() 

114 self.process_iris_completed_callback() 

115 self.process_iris_approved_callback( 

116 nationality=nationality, 

117 sex=sex, 

118 date_of_birth=date_of_birth, 

119 document_type=document_type, 

120 document_number=document_number, 

121 document_expiry=document_expiry, 

122 expected_status=expected_status, 

123 ) 

124 

125 def process_iris_initiated_callback(self) -> None: 

126 # ok, now the user downloads the app, scans their id, and Iris ID sends callbacks to the server 

127 self._emulate_iris_callback("INITIATED") 

128 

129 with account_session(self.token) as account: 

130 assert ( 

131 account.GetStrongVerificationAttemptStatus( 

132 account_pb2.GetStrongVerificationAttemptStatusReq( 

133 verification_attempt_token=self.verification_attempt_token 

134 ) 

135 ).status 

136 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_IN_APP 

137 ) 

138 

139 def process_iris_completed_callback(self) -> None: 

140 self._emulate_iris_callback("COMPLETED") 

141 

142 with account_session(self.token) as account: 

143 assert ( 

144 account.GetStrongVerificationAttemptStatus( 

145 account_pb2.GetStrongVerificationAttemptStatusReq( 

146 verification_attempt_token=self.verification_attempt_token 

147 ) 

148 ).status 

149 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND 

150 ) 

151 

152 def process_iris_approved_callback( 

153 self, 

154 *, 

155 nationality: str = "US", 

156 sex: PassportSex | None = None, 

157 date_of_birth: date | None = None, 

158 document_type: str = "PASSPORT", 

159 document_number: str = "31195855", 

160 document_expiry: date | None = None, 

161 expected_status: StrongVerificationAttemptStatus = StrongVerificationAttemptStatus.succeeded, 

162 ) -> None: 

163 if sex is None: 

164 match self.user.gender: 

165 case "Man": 165 ↛ 167line 165 didn't jump to line 167 because the pattern on line 165 always matched

166 sex = PassportSex.male 

167 case "Woman": 

168 sex = PassportSex.female 

169 case _: 

170 sex = PassportSex.unspecified 

171 if date_of_birth is None: 

172 date_of_birth = self.user.birthdate 

173 if document_expiry is None: 

174 document_expiry = date.today() + timedelta(days=5 * 365) 

175 

176 self._emulate_iris_callback("APPROVED") 

177 

178 with account_session(self.token) as account: 

179 assert ( 

180 account.GetStrongVerificationAttemptStatus( 

181 account_pb2.GetStrongVerificationAttemptStatusReq( 

182 verification_attempt_token=self.verification_attempt_token 

183 ) 

184 ).status 

185 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND 

186 ) 

187 

188 with patch("couchers.jobs.handlers.requests.post") as mock: 

189 json_resp = { 

190 "id": self.verification_id, 

191 "created": "2024-05-11T15:46:46Z", 

192 "expires": "2024-05-11T16:17:26Z", 

193 "state": "APPROVED", 

194 "reference": self.reference_data, 

195 "user_ip": "10.123.123.123", 

196 "user_agent": "Iris%20ID/168357896 CFNetwork/1494.0.7 Darwin/23.4.0", 

197 "given_names": "John Wayne", 

198 "surname": "Doe", 

199 "nationality": nationality, 

200 "sex": sex.name.upper(), 

201 "date_of_birth": date_of_birth.isoformat(), 

202 "document_type": document_type, 

203 "document_number": document_number, 

204 "expiry_date": document_expiry.isoformat(), 

205 "issuing_country": nationality, 

206 "issuer": "Department of State, U.S. Government", 

207 "portrait": "dGVzdHRlc3R0ZXN0...", 

208 } 

209 mock.return_value = type( 

210 "__MockResponse", 

211 (), 

212 { 

213 "status_code": 200, 

214 "text": json.dumps(json_resp), 

215 "json": lambda: json_resp, 

216 }, 

217 ) 

218 while process_job(): 

219 pass 

220 

221 mock.assert_called_once_with( 

222 "https://passportreader.app/api/v1/session.get", 

223 auth=("dummy_pubkey", "dummy_secret"), 

224 json={"id": self.verification_id}, 

225 timeout=10, 

226 verify="/etc/ssl/certs/ca-certificates.crt", 

227 ) 

228 

229 # The API should now report success or failure 

230 with account_session(self.token) as account: 

231 expected_pb_status = ( 

232 account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_SUCCEEDED 

233 if expected_status == StrongVerificationAttemptStatus.succeeded 

234 else account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_FAILED 

235 ) 

236 assert ( 

237 account.GetStrongVerificationAttemptStatus( 

238 account_pb2.GetStrongVerificationAttemptStatusReq( 

239 verification_attempt_token=self.verification_attempt_token 

240 ) 

241 ).status 

242 == expected_pb_status 

243 ) 

244 

245 # The StrongVerificationAttempt row should now reflect data from the IRIS callback 

246 with session_scope() as session: 

247 verification_attempt = session.execute( 

248 select(StrongVerificationAttempt).where( 

249 StrongVerificationAttempt.verification_attempt_token == self.verification_attempt_token 

250 ) 

251 ).scalar_one() 

252 

253 assert verification_attempt.user_id == self.user.id 

254 assert verification_attempt.iris_token == self.iris_token 

255 assert verification_attempt.iris_session_id == self.verification_id 

256 assert verification_attempt.status == expected_status 

257 

258 # Check minimal data 

259 if verification_attempt.status != StrongVerificationAttemptStatus.failed: 

260 assert verification_attempt.has_minimal_data 

261 assert verification_attempt.passport_expiry_date == document_expiry 

262 assert verification_attempt.passport_nationality == nationality 

263 assert verification_attempt.passport_last_three_document_chars == document_number[-3:] 

264 else: 

265 assert not verification_attempt.has_minimal_data 

266 

267 # Check full data 

268 if verification_attempt.status == StrongVerificationAttemptStatus.succeeded: 

269 assert verification_attempt.has_full_data 

270 assert verification_attempt.passport_encrypted_data 

271 assert verification_attempt.passport_date_of_birth == date_of_birth 

272 assert verification_attempt.passport_sex == sex 

273 

274 private_key = bytes.fromhex("e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272") 

275 decrypted_data = json.loads(asym_decrypt(private_key, verification_attempt.passport_encrypted_data)) 

276 assert decrypted_data == json_resp 

277 else: 

278 assert not verification_attempt.has_full_data 

279 

280 # We should have gone through all IRIS callbacks 

281 callbacks = ( 

282 session.execute( 

283 select(StrongVerificationCallbackEvent.iris_status) 

284 .where(StrongVerificationCallbackEvent.verification_attempt_id == verification_attempt.id) 

285 .order_by(StrongVerificationCallbackEvent.created.asc()) 

286 ) 

287 .scalars() 

288 .all() 

289 ) 

290 assert callbacks == ["INITIATED", "COMPLETED", "APPROVED"] 

291 

292 def _emulate_iris_callback(self, session_state: str): 

293 assert session_state in ["CREATED", "INITIATED", "FAILED", "ABORTED", "COMPLETED", "REJECTED", "APPROVED"] 

294 with real_iris_session() as iris: 

295 data = json.dumps( 

296 { 

297 "session_id": self.verification_id, 

298 "session_state": session_state, 

299 "session_reference": self.reference_data, 

300 } 

301 ).encode("ascii") 

302 iris.Webhook(httpbody_pb2.HttpBody(content_type="application/json", data=data)) 

303 

304 

305def monkeypatch_sv_config(monkeypatch): 

306 new_config = config.copy() 

307 new_config.IRIS_ID_PUBKEY = "dummy_pubkey" 

308 new_config.IRIS_ID_SECRET = "dummy_secret" 

309 new_config.VERIFICATION_DATA_PUBLIC_KEY = bytes.fromhex( 

310 "dd740a2b2a35bf05041a28257ea439b30f76f056f3698000b71e6470cd82275f" 

311 ) 

312 

313 private_key = bytes.fromhex("e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272") 

314 

315 monkeypatch.setattr(couchers.servicers.account, "config", new_config) 

316 monkeypatch.setattr(couchers.jobs.handlers, "config", new_config) 

317 

318 

319def test_strong_verification_happy_path(db, monkeypatch): 

320 monkeypatch_sv_config(monkeypatch) 

321 

322 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

323 _, superuser_token = generate_user(is_superuser=True) 

324 

325 update_badges(empty_pb2.Empty()) 

326 refresh_materialized_views_rapid(empty_pb2.Empty()) 

327 

328 with api_session(token) as api: 

329 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

330 assert "strong_verification" not in res.badges 

331 assert not res.has_strong_verification 

332 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED 

333 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED 

334 assert ( 

335 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

336 == res.has_strong_verification 

337 ) 

338 

339 expiry = date.today() + timedelta(days=5 * 365) 

340 

341 MockSVFlow(user=user, token=token).process_iris_callbacks( 

342 nationality="US", 

343 sex=PassportSex.male, 

344 date_of_birth=date.fromisoformat("1988-01-01"), 

345 document_number="31195855", 

346 document_expiry=expiry, 

347 ) 

348 

349 with session_scope() as session: 

350 verification_attempt = session.execute( 

351 select(StrongVerificationAttempt).where(StrongVerificationAttempt.user_id == user.id) 

352 ).scalar_one() 

353 assert verification_attempt.status == StrongVerificationAttemptStatus.succeeded 

354 assert verification_attempt.passport_date_of_birth == date(1988, 1, 1) 

355 assert verification_attempt.passport_sex == PassportSex.male 

356 assert verification_attempt.passport_expiry_date == expiry 

357 assert verification_attempt.passport_nationality == "US" 

358 assert verification_attempt.passport_last_three_document_chars == "855" 

359 

360 update_badges(empty_pb2.Empty()) 

361 refresh_materialized_views_rapid(empty_pb2.Empty()) 

362 

363 # the user should now have strong verification 

364 with api_session(token) as api: 

365 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

366 assert "strong_verification" in res.badges 

367 assert res.has_strong_verification 

368 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

369 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

370 assert ( 

371 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

372 == res.has_strong_verification 

373 ) 

374 

375 # wrong dob = no badge 

376 with session_scope() as session: 

377 session.execute(update(User).where(User.id == user.id).values(birthdate=date(1988, 1, 2))) 

378 

379 update_badges(empty_pb2.Empty()) 

380 refresh_materialized_views_rapid(empty_pb2.Empty()) 

381 

382 with api_session(token) as api: 

383 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

384 assert "strong_verification" not in res.badges 

385 assert not res.has_strong_verification 

386 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_MISMATCH 

387 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

388 assert ( 

389 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

390 == res.has_strong_verification 

391 ) 

392 

393 # bad gender-sex correspondence = no badge 

394 with session_scope() as session: 

395 session.execute(update(User).where(User.id == user.id).values(birthdate=date(1988, 1, 1), gender="Woman")) 

396 

397 update_badges(empty_pb2.Empty()) 

398 refresh_materialized_views_rapid(empty_pb2.Empty()) 

399 

400 with api_session(token) as api: 

401 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

402 assert "strong_verification" not in res.badges 

403 assert not res.has_strong_verification 

404 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

405 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

406 assert ( 

407 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

408 == res.has_strong_verification 

409 ) 

410 

411 with account_session(token) as account: 

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

413 assert not res.has_strong_verification 

414 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

415 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

416 

417 # back to should have a badge 

418 with session_scope() as session: 

419 session.execute(update(User).where(User.id == user.id).values(gender="Man")) 

420 

421 update_badges(empty_pb2.Empty()) 

422 refresh_materialized_views_rapid(empty_pb2.Empty()) 

423 

424 with api_session(token) as api: 

425 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

426 assert "strong_verification" in res.badges 

427 assert res.has_strong_verification 

428 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

429 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

430 assert ( 

431 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

432 == res.has_strong_verification 

433 ) 

434 

435 # check has_passport_sex_gender_exception 

436 with real_admin_session(superuser_token) as admin: 

437 res = admin.GetUserDetails(admin_pb2.GetUserDetailsReq(user=user.username)) 

438 assert "strong_verification" in res.badges 

439 assert res.has_strong_verification 

440 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

441 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

442 

443 admin.SetPassportSexGenderException( 

444 admin_pb2.SetPassportSexGenderExceptionReq(user=user.username, passport_sex_gender_exception=True) 

445 ) 

446 admin.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=user.username, gender="Woman")) 

447 

448 update_badges(empty_pb2.Empty()) 

449 refresh_materialized_views_rapid(empty_pb2.Empty()) 

450 

451 with api_session(token) as api: 

452 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

453 assert "strong_verification" in res.badges 

454 assert res.has_strong_verification 

455 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

456 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

457 assert ( 

458 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

459 == res.has_strong_verification 

460 ) 

461 

462 with real_admin_session(superuser_token) as admin: 

463 res = admin.GetUserDetails(admin_pb2.GetUserDetailsReq(user=user.username)) 

464 assert "strong_verification" in res.badges 

465 assert res.has_strong_verification 

466 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

467 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

468 

469 # now turn exception off 

470 admin.SetPassportSexGenderException( 

471 admin_pb2.SetPassportSexGenderExceptionReq(user=user.username, passport_sex_gender_exception=False) 

472 ) 

473 

474 update_badges(empty_pb2.Empty()) 

475 refresh_materialized_views_rapid(empty_pb2.Empty()) 

476 

477 with api_session(token) as api: 

478 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

479 assert "strong_verification" not in res.badges 

480 assert not res.has_strong_verification 

481 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

482 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

483 assert ( 

484 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

485 == res.has_strong_verification 

486 ) 

487 

488 with real_admin_session(superuser_token) as admin: 

489 res = admin.GetUserDetails(admin_pb2.GetUserDetailsReq(user=user.username)) 

490 assert "strong_verification" not in res.badges 

491 assert not res.has_strong_verification 

492 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

493 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

494 

495 

496def test_strong_verification_delete_data(db, monkeypatch): 

497 monkeypatch_sv_config(monkeypatch) 

498 

499 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

500 _, superuser_token = generate_user(is_superuser=True) 

501 

502 refresh_materialized_views_rapid(empty_pb2.Empty()) 

503 

504 with api_session(token) as api: 

505 assert not api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

506 assert ( 

507 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

508 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

509 ) 

510 

511 # can remove SV data even if there is none, should do nothing 

512 with account_session(token) as account: 

513 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

514 

515 MockSVFlow(user=user, token=token).process_iris_callbacks() 

516 

517 refresh_materialized_views_rapid(empty_pb2.Empty()) 

518 

519 # the user should now have strong verification 

520 with api_session(token) as api: 

521 assert api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

522 assert ( 

523 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

524 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

525 ) 

526 

527 # check removing SV data 

528 with account_session(token) as account: 

529 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

530 

531 refresh_materialized_views_rapid(empty_pb2.Empty()) 

532 

533 with api_session(token) as api: 

534 assert not api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

535 assert ( 

536 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

537 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

538 ) 

539 

540 with session_scope() as session: 

541 assert ( 

542 len( 

543 session.execute( 

544 select(StrongVerificationAttempt).where( 

545 or_( 

546 StrongVerificationAttempt.passport_encrypted_data != None, 

547 StrongVerificationAttempt.passport_date_of_birth != None, 

548 StrongVerificationAttempt.passport_sex != None, 

549 ) 

550 ) 

551 ) 

552 .scalars() 

553 .all() 

554 ) 

555 == 0 

556 ) 

557 

558 

559def test_strong_verification_expiry(db, monkeypatch): 

560 monkeypatch_sv_config(monkeypatch) 

561 

562 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

563 _, superuser_token = generate_user(is_superuser=True) 

564 

565 refresh_materialized_views_rapid(empty_pb2.Empty()) 

566 

567 with api_session(token) as api: 

568 assert not api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

569 assert ( 

570 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

571 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

572 ) 

573 

574 MockSVFlow(user=user, token=token).process_iris_callbacks(document_expiry=date.today() + timedelta(days=10)) 

575 

576 # the user should now have strong verification 

577 with api_session(token) as api: 

578 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

579 assert res.has_strong_verification 

580 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

581 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

582 

583 with session_scope() as session: 

584 attempt = session.execute(select(StrongVerificationAttempt)).scalars().one() 

585 attempt.passport_expiry_date = date.today() - timedelta(days=2) 

586 

587 with api_session(token) as api: 

588 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

589 assert not res.has_strong_verification 

590 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED 

591 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED 

592 

593 res = api.GetUser(api_pb2.GetUserReq(user=user.username)) 

594 assert not res.has_strong_verification 

595 assert not res.has_strong_verification 

596 

597 MockSVFlow(user=user, token=token, verification_id=5731012934821985).process_iris_callbacks( 

598 nationality="AU", document_number="PA41323412", document_expiry=date.today() + timedelta(days=365) 

599 ) 

600 

601 refresh_materialized_views_rapid(empty_pb2.Empty()) 

602 

603 with api_session(token) as api: 

604 assert api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

605 assert ( 

606 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

607 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

608 ) 

609 

610 

611def test_strong_verification_regression(db, monkeypatch): 

612 monkeypatch_sv_config(monkeypatch) 

613 

614 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

615 

616 sv_flow = MockSVFlow(user=user, token=token) 

617 sv_flow.process_iris_initiated_callback() 

618 

619 with api_session(token) as api: 

620 api.Ping(api_pb2.PingReq()) 

621 

622 

623def test_strong_verification_regression2(db, monkeypatch): 

624 monkeypatch_sv_config(monkeypatch) 

625 

626 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

627 

628 sv_flow = MockSVFlow(user=user, token=token, verification_id=5731012934821983) 

629 sv_flow.process_iris_initiated_callback() 

630 

631 sv_flow = MockSVFlow(user=user, token=token, verification_id=5731012934821985) 

632 sv_flow.process_iris_callbacks(nationality="AU", document_number="PA41323412") 

633 

634 refresh_materialized_views_rapid(empty_pb2.Empty()) 

635 

636 with api_session(token) as api: 

637 assert api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

638 assert ( 

639 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

640 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

641 ) 

642 

643 

644def test_strong_verification_disabled(db, feature_flags): 

645 feature_flags.set("strong_verification_enabled", False) 

646 user, token = generate_user() 

647 

648 with account_session(token) as account: 

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

650 account.InitiateStrongVerification(empty_pb2.Empty()) 

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

652 assert e.value.details() == "Strong verification is currently disabled." 

653 

654 

655def test_strong_verification_delete_data_cant_reverify(db, monkeypatch, push_collector: PushCollector): 

656 monkeypatch_sv_config(monkeypatch) 

657 

658 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

659 _, superuser_token = generate_user(is_superuser=True) 

660 

661 refresh_materialized_views_rapid(empty_pb2.Empty()) 

662 

663 with api_session(token) as api: 

664 assert not api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

665 assert ( 

666 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

667 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

668 ) 

669 

670 MockSVFlow(user=user, token=token).process_iris_callbacks() 

671 

672 refresh_materialized_views_rapid(empty_pb2.Empty()) 

673 

674 # the user should now have strong verification 

675 with api_session(token) as api: 

676 assert api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

677 assert ( 

678 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

679 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

680 ) 

681 

682 # There should be a notification confirming it 

683 push_collector.pop_for_user(user.id, last=True) 

684 

685 # check removing SV data 

686 with account_session(token) as account: 

687 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

688 

689 refresh_materialized_views_rapid(empty_pb2.Empty()) 

690 

691 with api_session(token) as api: 

692 assert not api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

693 assert ( 

694 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

695 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

696 ) 

697 

698 with session_scope() as session: 

699 assert ( 

700 len( 

701 session.execute( 

702 select(StrongVerificationAttempt).where( 

703 or_( 

704 StrongVerificationAttempt.passport_encrypted_data != None, 

705 StrongVerificationAttempt.passport_date_of_birth != None, 

706 StrongVerificationAttempt.passport_sex != None, 

707 ) 

708 ) 

709 ) 

710 .scalars() 

711 .all() 

712 ) 

713 == 0 

714 ) 

715 

716 MockSVFlow(user=user, token=token, verification_id=5731012934821984).process_iris_callbacks( 

717 expected_status=StrongVerificationAttemptStatus.duplicate 

718 ) 

719 

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

721 assert push.content.title == "Strong Verification failed" 

722 assert ( 

723 push.content.body 

724 == "You used a passport that has already been used for verification. Please use another passport." 

725 ) 

726 

727 refresh_materialized_views_rapid(empty_pb2.Empty()) 

728 

729 with api_session(token) as api: 

730 assert not api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

731 assert ( 

732 api.GetLiteUser(api_pb2.GetLiteUserReq(user=user.username)).has_strong_verification 

733 == api.GetUser(api_pb2.GetUserReq(user=user.username)).has_strong_verification 

734 ) 

735 

736 

737def test_strong_verification_duplicate_other_user(db, monkeypatch, push_collector: PushCollector): 

738 monkeypatch_sv_config(monkeypatch) 

739 

740 user, token = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

741 user2, token2 = generate_user(birthdate=date(1988, 1, 1), gender="Man") 

742 _, superuser_token = generate_user(is_superuser=True)