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

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 

12from couchers.config import config 

13from couchers.crypto import asym_decrypt, b64encode_unpadded 

14from couchers.db import session_scope 

15from couchers.jobs.handlers import update_badges 

16from couchers.jobs.worker import process_job 

17from couchers.materialized_views import refresh_materialized_views_rapid 

18from couchers.models import ( 

19 PassportSex, 

20 StrongVerificationAttempt, 

21 StrongVerificationAttemptStatus, 

22 StrongVerificationCallbackEvent, 

23 User, 

24) 

25from couchers.proto import account_pb2, admin_pb2, api_pb2 

26from couchers.proto.google.api import httpbody_pb2 

27from tests.fixtures.db import generate_user 

28from tests.fixtures.misc import PushCollector 

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

30 

31 

32def _emulate_iris_callback(session_id, session_state, reference): 

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

34 with real_iris_session() as iris: 

35 data = json.dumps( 

36 {"session_id": session_id, "session_state": session_state, "session_reference": reference} 

37 ).encode("ascii") 

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

39 

40 

41default_expiry = date.today() + timedelta(days=5 * 365) 

42 

43 

44def do_and_check_sv( 

45 user, 

46 token, 

47 verification_id, 

48 sex, 

49 dob, 

50 document_type, 

51 document_number, 

52 document_expiry, 

53 nationality, 

54 return_after=None, 

55): 

56 iris_token_data = { 

57 "merchant_id": 5731012934821982, 

58 "session_id": verification_id, 

59 "seed": 1674246339, 

60 "face_verification": False, 

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

62 } 

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

64 

65 with account_session(token) as account: 

66 # start by initiation 

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

68 json_resp1 = { 

69 "id": verification_id, 

70 "token": iris_token, 

71 } 

72 mock.return_value = type( 

73 "__MockResponse", 

74 (), 

75 { 

76 "status_code": 200, 

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

78 "json": lambda: json_resp1, 

79 }, 

80 ) 

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

82 mock.assert_called_once_with( 

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

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

85 json={ 

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

87 "face_verification": False, 

88 "passport_only": True, 

89 "reference": ANY, 

90 }, 

91 timeout=10, 

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

93 ) 

94 reference_data = mock.call_args.kwargs["json"]["reference"] 

95 verification_attempt_token = res.verification_attempt_token 

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

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

98 {"token": iris_token, "redirect_url": return_url} 

99 ) 

100 

101 assert ( 

102 account.GetStrongVerificationAttemptStatus( 

103 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

104 ).status 

105 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_TO_OPEN_APP 

106 ) 

107 

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

109 _emulate_iris_callback(verification_id, "INITIATED", reference_data) 

110 

111 with account_session(token) as account: 

112 assert ( 

113 account.GetStrongVerificationAttemptStatus( 

114 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

115 ).status 

116 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_IN_APP 

117 ) 

118 

119 if return_after == "INITIATED": 

120 return reference_data 

121 

122 _emulate_iris_callback(verification_id, "COMPLETED", reference_data) 

123 

124 with account_session(token) as account: 

125 assert ( 

126 account.GetStrongVerificationAttemptStatus( 

127 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

128 ).status 

129 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND 

130 ) 

131 

132 if return_after == "COMPLETED": 132 ↛ 133line 132 didn't jump to line 133 because the condition on line 132 was never true

133 return reference_data 

134 

135 _emulate_iris_callback(verification_id, "APPROVED", reference_data) 

136 

137 with account_session(token) as account: 

138 assert ( 

139 account.GetStrongVerificationAttemptStatus( 

140 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

141 ).status 

142 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND 

143 ) 

144 

145 if return_after == "APPROVED": 

146 return reference_data 

147 

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

149 json_resp2 = { 

150 "id": verification_id, 

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

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

153 "state": "APPROVED", 

154 "reference": reference_data, 

155 "user_ip": "10.123.123.123", 

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

157 "given_names": "John Wayne", 

158 "surname": "Doe", 

159 "nationality": nationality, 

160 "sex": sex, 

161 "date_of_birth": dob, 

162 "document_type": document_type, 

163 "document_number": document_number, 

164 "expiry_date": document_expiry.isoformat(), 

165 "issuing_country": nationality, 

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

167 "portrait": "dGVzdHRlc3R0ZXN0...", 

168 } 

169 mock.return_value = type( 

170 "__MockResponse", 

171 (), 

172 { 

173 "status_code": 200, 

174 "text": json.dumps(json_resp2), 

175 "json": lambda: json_resp2, 

176 }, 

177 ) 

178 while process_job(): 

179 pass 

180 

181 mock.assert_called_once_with( 

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

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

184 json={"id": verification_id}, 

185 timeout=10, 

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

187 ) 

188 

189 with account_session(token) as account: 

190 assert ( 

191 account.GetStrongVerificationAttemptStatus( 

192 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

193 ).status 

194 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_SUCCEEDED 

195 ) 

196 

197 with session_scope() as session: 

198 verification_attempt = session.execute( 

199 select(StrongVerificationAttempt).where( 

200 StrongVerificationAttempt.verification_attempt_token == verification_attempt_token 

201 ) 

202 ).scalar_one() 

203 assert verification_attempt.user_id == user.id 

204 assert verification_attempt.status == StrongVerificationAttemptStatus.succeeded 

205 assert verification_attempt.has_full_data 

206 assert verification_attempt.passport_encrypted_data 

207 # assert verification_attempt.passport_date_of_birth == date(1988, 1, 1) 

208 # assert verification_attempt.passport_sex == PassportSex.male 

209 assert verification_attempt.has_minimal_data 

210 assert verification_attempt.passport_expiry_date == document_expiry 

211 assert verification_attempt.passport_nationality == nationality 

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

213 assert verification_attempt.iris_token == iris_token 

214 assert verification_attempt.iris_session_id == verification_id 

215 

216 private_key = bytes.fromhex("e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272") 

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

218 assert decrypted_data == json_resp2 

219 

220 callbacks = ( 

221 session.execute( 

222 select(StrongVerificationCallbackEvent.iris_status) 

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

224 .order_by(StrongVerificationCallbackEvent.created.asc()) 

225 ) 

226 .scalars() 

227 .all() 

228 ) 

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

230 

231 

232@pytest.fixture 

233def sv_config() -> None: 

234 # An autouse fixture runs before non-autouse fixtures of the same scope. 

235 config.IRIS_ID_PUBKEY = "dummy_pubkey" 

236 config.IRIS_ID_SECRET = "dummy_secret" 

237 config.VERIFICATION_DATA_PUBLIC_KEY = bytes.fromhex( 

238 "dd740a2b2a35bf05041a28257ea439b30f76f056f3698000b71e6470cd82275f" 

239 ) 

240 

241 

242def test_strong_verification_happy_path(db, sv_config): 

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

244 _, superuser_token = generate_user(is_superuser=True) 

245 

246 update_badges(empty_pb2.Empty()) 

247 refresh_materialized_views_rapid(empty_pb2.Empty()) 

248 

249 with api_session(token) as api: 

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

251 assert "strong_verification" not in res.badges 

252 assert not res.has_strong_verification 

253 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED 

254 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED 

255 assert ( 

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

257 == res.has_strong_verification 

258 ) 

259 

260 do_and_check_sv( 

261 user, 

262 token, 

263 verification_id=5731012934821983, 

264 sex="MALE", 

265 dob="1988-01-01", 

266 document_type="PASSPORT", 

267 document_number="31195855", 

268 document_expiry=default_expiry, 

269 nationality="US", 

270 ) 

271 

272 with session_scope() as session: 

273 verification_attempt = session.execute( 

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

275 ).scalar_one() 

276 assert verification_attempt.status == StrongVerificationAttemptStatus.succeeded 

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

278 assert verification_attempt.passport_sex == PassportSex.male 

279 assert verification_attempt.passport_expiry_date == default_expiry 

280 assert verification_attempt.passport_nationality == "US" 

281 assert verification_attempt.passport_last_three_document_chars == "855" 

282 

283 update_badges(empty_pb2.Empty()) 

284 refresh_materialized_views_rapid(empty_pb2.Empty()) 

285 

286 # the user should now have strong verification 

287 with api_session(token) as api: 

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

289 assert "strong_verification" in res.badges 

290 assert res.has_strong_verification 

291 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

292 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

293 assert ( 

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

295 == res.has_strong_verification 

296 ) 

297 

298 # wrong dob = no badge 

299 with session_scope() as session: 

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

301 

302 update_badges(empty_pb2.Empty()) 

303 refresh_materialized_views_rapid(empty_pb2.Empty()) 

304 

305 with api_session(token) as api: 

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

307 assert "strong_verification" not in res.badges 

308 assert not res.has_strong_verification 

309 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_MISMATCH 

310 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

311 assert ( 

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

313 == res.has_strong_verification 

314 ) 

315 

316 # bad gender-sex correspondence = no badge 

317 with session_scope() as session: 

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

319 

320 update_badges(empty_pb2.Empty()) 

321 refresh_materialized_views_rapid(empty_pb2.Empty()) 

322 

323 with api_session(token) as api: 

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

325 assert "strong_verification" not in res.badges 

326 assert not res.has_strong_verification 

327 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

328 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

329 assert ( 

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

331 == res.has_strong_verification 

332 ) 

333 

334 with account_session(token) as account: 

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

336 assert not res.has_strong_verification 

337 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

338 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

339 

340 # back to should have a badge 

341 with session_scope() as session: 

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

343 

344 update_badges(empty_pb2.Empty()) 

345 refresh_materialized_views_rapid(empty_pb2.Empty()) 

346 

347 with api_session(token) as api: 

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

349 assert "strong_verification" in res.badges 

350 assert res.has_strong_verification 

351 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

352 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

353 assert ( 

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

355 == res.has_strong_verification 

356 ) 

357 

358 # check has_passport_sex_gender_exception 

359 with real_admin_session(superuser_token) as admin: 

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

361 assert "strong_verification" in res.badges 

362 assert res.has_strong_verification 

363 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

364 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

365 

366 admin.SetPassportSexGenderException( 

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

368 ) 

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

370 

371 update_badges(empty_pb2.Empty()) 

372 refresh_materialized_views_rapid(empty_pb2.Empty()) 

373 

374 with api_session(token) as api: 

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

376 assert "strong_verification" in res.badges 

377 assert res.has_strong_verification 

378 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

379 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

380 assert ( 

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

382 == res.has_strong_verification 

383 ) 

384 

385 with real_admin_session(superuser_token) as admin: 

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

387 assert "strong_verification" in res.badges 

388 assert res.has_strong_verification 

389 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

390 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

391 

392 # now turn exception off 

393 admin.SetPassportSexGenderException( 

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

395 ) 

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 real_admin_session(superuser_token) as admin: 

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

413 assert "strong_verification" not in res.badges 

414 assert not res.has_strong_verification 

415 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

416 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

417 

418 

419def test_strong_verification_delete_data(db, sv_config): 

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

421 _, superuser_token = generate_user(is_superuser=True) 

422 

423 refresh_materialized_views_rapid(empty_pb2.Empty()) 

424 

425 with api_session(token) as api: 

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

427 assert ( 

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

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

430 ) 

431 

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

433 with account_session(token) as account: 

434 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

435 

436 do_and_check_sv( 

437 user, 

438 token, 

439 verification_id=5731012934821983, 

440 sex="MALE", 

441 dob="1988-01-01", 

442 document_type="PASSPORT", 

443 document_number="31195855", 

444 document_expiry=default_expiry, 

445 nationality="US", 

446 ) 

447 

448 refresh_materialized_views_rapid(empty_pb2.Empty()) 

449 

450 # the user should now have strong verification 

451 with api_session(token) as api: 

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

453 assert ( 

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

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

456 ) 

457 

458 # check removing SV data 

459 with account_session(token) as account: 

460 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

461 

462 refresh_materialized_views_rapid(empty_pb2.Empty()) 

463 

464 with api_session(token) as api: 

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

466 assert ( 

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

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

469 ) 

470 

471 with session_scope() as session: 

472 assert ( 

473 len( 

474 session.execute( 

475 select(StrongVerificationAttempt).where( 

476 or_( 

477 StrongVerificationAttempt.passport_encrypted_data != None, 

478 StrongVerificationAttempt.passport_date_of_birth != None, 

479 StrongVerificationAttempt.passport_sex != None, 

480 ) 

481 ) 

482 ) 

483 .scalars() 

484 .all() 

485 ) 

486 == 0 

487 ) 

488 

489 

490def test_strong_verification_expiry(db, sv_config): 

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

492 _, superuser_token = generate_user(is_superuser=True) 

493 

494 refresh_materialized_views_rapid(empty_pb2.Empty()) 

495 

496 with api_session(token) as api: 

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

498 assert ( 

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

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

501 ) 

502 

503 expiry = date.today() + timedelta(days=10) 

504 

505 do_and_check_sv( 

506 user, 

507 token, 

508 verification_id=5731012934821983, 

509 sex="MALE", 

510 dob="1988-01-01", 

511 document_type="PASSPORT", 

512 document_number="31195855", 

513 document_expiry=expiry, 

514 nationality="US", 

515 ) 

516 

517 # the user should now have strong verification 

518 with api_session(token) as api: 

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

520 assert res.has_strong_verification 

521 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

522 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

523 

524 with session_scope() as session: 

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

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

527 

528 with api_session(token) as api: 

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

530 assert not res.has_strong_verification 

531 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED 

532 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED 

533 

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

535 assert not res.has_strong_verification 

536 assert not res.has_strong_verification 

537 

538 do_and_check_sv( 

539 user, 

540 token, 

541 verification_id=5731012934821985, 

542 sex="MALE", 

543 dob="1988-01-01", 

544 document_type="PASSPORT", 

545 document_number="PA41323412", 

546 document_expiry=date.today() + timedelta(days=365), 

547 nationality="AU", 

548 ) 

549 

550 refresh_materialized_views_rapid(empty_pb2.Empty()) 

551 

552 with api_session(token) as api: 

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

554 assert ( 

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

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

557 ) 

558 

559 

560def test_strong_verification_regression(db, sv_config): 

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

562 

563 do_and_check_sv( 

564 user, 

565 token, 

566 verification_id=5731012934821983, 

567 sex="MALE", 

568 dob="1988-01-01", 

569 document_type="PASSPORT", 

570 document_number="31195855", 

571 document_expiry=default_expiry, 

572 nationality="US", 

573 return_after="INITIATED", 

574 ) 

575 

576 with api_session(token) as api: 

577 api.Ping(api_pb2.PingReq()) 

578 

579 

580def test_strong_verification_regression2(db, sv_config): 

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

582 

583 do_and_check_sv( 

584 user, 

585 token, 

586 verification_id=5731012934821983, 

587 sex="MALE", 

588 dob="1988-01-01", 

589 document_type="PASSPORT", 

590 document_number="31195855", 

591 document_expiry=default_expiry, 

592 nationality="US", 

593 return_after="INITIATED", 

594 ) 

595 

596 do_and_check_sv( 

597 user, 

598 token, 

599 verification_id=5731012934821985, 

600 sex="MALE", 

601 dob="1988-01-01", 

602 document_type="PASSPORT", 

603 document_number="PA41323412", 

604 document_expiry=default_expiry, 

605 nationality="AU", 

606 ) 

607 

608 refresh_materialized_views_rapid(empty_pb2.Empty()) 

609 

610 with api_session(token) as api: 

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

612 assert ( 

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

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

615 ) 

616 

617 

618def test_strong_verification_disabled(db, feature_flags): 

619 feature_flags.set("strong_verification_enabled", False) 

620 user, token = generate_user() 

621 

622 with account_session(token) as account: 

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

624 account.InitiateStrongVerification(empty_pb2.Empty()) 

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

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

627 

628 

629def test_strong_verification_delete_data_cant_reverify(db, sv_config, push_collector: PushCollector): 

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

631 _, superuser_token = generate_user(is_superuser=True) 

632 

633 refresh_materialized_views_rapid(empty_pb2.Empty()) 

634 

635 with api_session(token) as api: 

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

637 assert ( 

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

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

640 ) 

641 

642 do_and_check_sv( 

643 user, 

644 token, 

645 verification_id=5731012934821983, 

646 sex="MALE", 

647 dob="1988-01-01", 

648 document_type="PASSPORT", 

649 document_number="31195855", 

650 document_expiry=default_expiry, 

651 nationality="US", 

652 ) 

653 

654 refresh_materialized_views_rapid(empty_pb2.Empty()) 

655 

656 # the user should now have strong verification 

657 with api_session(token) as api: 

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

659 assert ( 

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

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

662 ) 

663 

664 # There should be a notification confirming it 

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

666 

667 # check removing SV data 

668 with account_session(token) as account: 

669 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

670 

671 refresh_materialized_views_rapid(empty_pb2.Empty()) 

672 

673 with api_session(token) as api: 

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

675 assert ( 

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

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

678 ) 

679 

680 with session_scope() as session: 

681 assert ( 

682 len( 

683 session.execute( 

684 select(StrongVerificationAttempt).where( 

685 or_( 

686 StrongVerificationAttempt.passport_encrypted_data != None, 

687 StrongVerificationAttempt.passport_date_of_birth != None, 

688 StrongVerificationAttempt.passport_sex != None, 

689 ) 

690 ) 

691 ) 

692 .scalars() 

693 .all() 

694 ) 

695 == 0 

696 ) 

697 

698 reference_data = do_and_check_sv( 

699 user, 

700 token, 

701 verification_id=5731012934821984, 

702 sex="MALE", 

703 dob="1988-01-01", 

704 document_type="PASSPORT", 

705 document_number="31195855", 

706 document_expiry=default_expiry, 

707 nationality="US", 

708 return_after="APPROVED", 

709 ) 

710 

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

712 json_resp2 = { 

713 "id": 5731012934821984, 

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

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

716 "state": "APPROVED", 

717 "reference": reference_data, 

718 "user_ip": "10.123.123.123", 

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

720 "given_names": "John Wayne", 

721 "surname": "Doe", 

722 "nationality": "US", 

723 "sex": "MALE", 

724 "date_of_birth": "1988-01-01", 

725 "document_type": "PASSPORT", 

726 "document_number": "31195855", 

727 "expiry_date": default_expiry.isoformat(), 

728 "issuing_country": "US", 

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

730 "portrait": "dGVzdHRlc3R0ZXN0...", 

731 } 

732 mock.return_value = type( 

733 "__MockResponse", 

734 (), 

735 { 

736 "status_code": 200, 

737 "text": json.dumps(json_resp2), 

738 "json": lambda: json_resp2, 

739 }, 

740 ) 

741 while process_job(): 

742 pass 

743 

744 mock.assert_called_once_with( 

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

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

747 json={"id": 5731012934821984}, 

748 timeout=10, 

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

750 ) 

751 

752 with session_scope() as session: 

753 verification_attempt = session.execute( 

754 select(StrongVerificationAttempt).where(StrongVerificationAttempt.iris_session_id == 5731012934821984) 

755 ).scalar_one() 

756 assert verification_attempt.user_id == user.id 

757 assert verification_attempt.status == StrongVerificationAttemptStatus.duplicate 

758 

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

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

761 assert ( 

762 push.content.body 

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

764 ) 

765 

766 refresh_materialized_views_rapid(empty_pb2.Empty()) 

767 

768 with api_session(token) as api: 

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

770 assert ( 

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

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

773 ) 

774 

775 

776def test_strong_verification_duplicate_other_user(db, sv_config, push_collector: PushCollector):