Coverage for 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 

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 

39def _emulate_iris_callback(session_id, session_state, reference): 

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

41 with real_iris_session() as iris: 

42 data = json.dumps( 

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

44 ).encode("ascii") 

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

46 

47 

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

49 

50 

51def do_and_check_sv( 

52 user, 

53 token, 

54 verification_id, 

55 sex, 

56 dob, 

57 document_type, 

58 document_number, 

59 document_expiry, 

60 nationality, 

61 return_after=None, 

62): 

63 iris_token_data = { 

64 "merchant_id": 5731012934821982, 

65 "session_id": verification_id, 

66 "seed": 1674246339, 

67 "face_verification": False, 

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

69 } 

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

71 

72 with account_session(token) as account: 

73 # start by initiation 

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

75 json_resp1 = { 

76 "id": verification_id, 

77 "token": iris_token, 

78 } 

79 mock.return_value = type( 

80 "__MockResponse", 

81 (), 

82 { 

83 "status_code": 200, 

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

85 "json": lambda: json_resp1, 

86 }, 

87 ) 

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

89 mock.assert_called_once_with( 

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

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

92 json={ 

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

94 "face_verification": False, 

95 "passport_only": True, 

96 "reference": ANY, 

97 }, 

98 timeout=10, 

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

100 ) 

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

102 verification_attempt_token = res.verification_attempt_token 

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

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

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

106 ) 

107 

108 assert ( 

109 account.GetStrongVerificationAttemptStatus( 

110 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

111 ).status 

112 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_TO_OPEN_APP 

113 ) 

114 

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

116 _emulate_iris_callback(verification_id, "INITIATED", reference_data) 

117 

118 with account_session(token) as account: 

119 assert ( 

120 account.GetStrongVerificationAttemptStatus( 

121 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

122 ).status 

123 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_IN_APP 

124 ) 

125 

126 if return_after == "INITIATED": 

127 return reference_data 

128 

129 _emulate_iris_callback(verification_id, "COMPLETED", reference_data) 

130 

131 with account_session(token) as account: 

132 assert ( 

133 account.GetStrongVerificationAttemptStatus( 

134 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

135 ).status 

136 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND 

137 ) 

138 

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

140 return reference_data 

141 

142 _emulate_iris_callback(verification_id, "APPROVED", reference_data) 

143 

144 with account_session(token) as account: 

145 assert ( 

146 account.GetStrongVerificationAttemptStatus( 

147 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

148 ).status 

149 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND 

150 ) 

151 

152 if return_after == "APPROVED": 

153 return reference_data 

154 

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

156 json_resp2 = { 

157 "id": verification_id, 

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

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

160 "state": "APPROVED", 

161 "reference": reference_data, 

162 "user_ip": "10.123.123.123", 

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

164 "given_names": "John Wayne", 

165 "surname": "Doe", 

166 "nationality": nationality, 

167 "sex": sex, 

168 "date_of_birth": dob, 

169 "document_type": document_type, 

170 "document_number": document_number, 

171 "expiry_date": document_expiry.isoformat(), 

172 "issuing_country": nationality, 

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

174 "portrait": "dGVzdHRlc3R0ZXN0...", 

175 } 

176 mock.return_value = type( 

177 "__MockResponse", 

178 (), 

179 { 

180 "status_code": 200, 

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

182 "json": lambda: json_resp2, 

183 }, 

184 ) 

185 while process_job(): 

186 pass 

187 

188 mock.assert_called_once_with( 

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

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

191 json={"id": verification_id}, 

192 timeout=10, 

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

194 ) 

195 

196 with account_session(token) as account: 

197 assert ( 

198 account.GetStrongVerificationAttemptStatus( 

199 account_pb2.GetStrongVerificationAttemptStatusReq(verification_attempt_token=verification_attempt_token) 

200 ).status 

201 == account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_SUCCEEDED 

202 ) 

203 

204 with session_scope() as session: 

205 verification_attempt = session.execute( 

206 select(StrongVerificationAttempt).where( 

207 StrongVerificationAttempt.verification_attempt_token == verification_attempt_token 

208 ) 

209 ).scalar_one() 

210 assert verification_attempt.user_id == user.id 

211 assert verification_attempt.status == StrongVerificationAttemptStatus.succeeded 

212 assert verification_attempt.has_full_data 

213 assert verification_attempt.passport_encrypted_data 

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

215 # assert verification_attempt.passport_sex == PassportSex.male 

216 assert verification_attempt.has_minimal_data 

217 assert verification_attempt.passport_expiry_date == document_expiry 

218 assert verification_attempt.passport_nationality == nationality 

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

220 assert verification_attempt.iris_token == iris_token 

221 assert verification_attempt.iris_session_id == verification_id 

222 

223 private_key = bytes.fromhex("e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272") 

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

225 assert decrypted_data == json_resp2 

226 

227 callbacks = ( 

228 session.execute( 

229 select(StrongVerificationCallbackEvent.iris_status) 

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

231 .order_by(StrongVerificationCallbackEvent.created.asc()) 

232 ) 

233 .scalars() 

234 .all() 

235 ) 

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

237 

238 

239def monkeypatch_sv_config(monkeypatch): 

240 new_config = config.copy() 

241 new_config["ENABLE_STRONG_VERIFICATION"] = True 

242 new_config["IRIS_ID_PUBKEY"] = "dummy_pubkey" 

243 new_config["IRIS_ID_SECRET"] = "dummy_secret" 

244 new_config["VERIFICATION_DATA_PUBLIC_KEY"] = bytes.fromhex( 

245 "dd740a2b2a35bf05041a28257ea439b30f76f056f3698000b71e6470cd82275f" 

246 ) 

247 

248 private_key = bytes.fromhex("e6c2fbf3756b387bc09a458a7b85935718ef3eb1c2777ef41d335c9f6c0ab272") 

249 

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

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

252 

253 

254def test_strong_verification_happy_path(db, monkeypatch): 

255 monkeypatch_sv_config(monkeypatch) 

256 

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

258 _, superuser_token = generate_user(is_superuser=True) 

259 

260 update_badges(empty_pb2.Empty()) 

261 refresh_materialized_views_rapid(empty_pb2.Empty()) 

262 

263 with api_session(token) as api: 

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

265 assert "strong_verification" not in res.badges 

266 assert not res.has_strong_verification 

267 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED 

268 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED 

269 assert ( 

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

271 == res.has_strong_verification 

272 ) 

273 

274 do_and_check_sv( 

275 user, 

276 token, 

277 verification_id=5731012934821983, 

278 sex="MALE", 

279 dob="1988-01-01", 

280 document_type="PASSPORT", 

281 document_number="31195855", 

282 document_expiry=default_expiry, 

283 nationality="US", 

284 ) 

285 

286 with session_scope() as session: 

287 verification_attempt = session.execute( 

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

289 ).scalar_one() 

290 assert verification_attempt.status == StrongVerificationAttemptStatus.succeeded 

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

292 assert verification_attempt.passport_sex == PassportSex.male 

293 assert verification_attempt.passport_expiry_date == default_expiry 

294 assert verification_attempt.passport_nationality == "US" 

295 assert verification_attempt.passport_last_three_document_chars == "855" 

296 

297 update_badges(empty_pb2.Empty()) 

298 refresh_materialized_views_rapid(empty_pb2.Empty()) 

299 

300 # the user should now have strong verification 

301 with api_session(token) as api: 

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

303 assert "strong_verification" in res.badges 

304 assert res.has_strong_verification 

305 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

306 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

307 assert ( 

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

309 == res.has_strong_verification 

310 ) 

311 

312 # wrong dob = no badge 

313 with session_scope() as session: 

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

315 

316 update_badges(empty_pb2.Empty()) 

317 refresh_materialized_views_rapid(empty_pb2.Empty()) 

318 

319 with api_session(token) as api: 

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

321 assert "strong_verification" not in res.badges 

322 assert not res.has_strong_verification 

323 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_MISMATCH 

324 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

325 assert ( 

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

327 == res.has_strong_verification 

328 ) 

329 

330 # bad gender-sex correspondence = no badge 

331 with session_scope() as session: 

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

333 

334 update_badges(empty_pb2.Empty()) 

335 refresh_materialized_views_rapid(empty_pb2.Empty()) 

336 

337 with api_session(token) as api: 

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

339 assert "strong_verification" not in res.badges 

340 assert not res.has_strong_verification 

341 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

342 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

343 assert ( 

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

345 == res.has_strong_verification 

346 ) 

347 

348 with account_session(token) as account: 

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

350 assert not 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_MISMATCH 

353 

354 # back to should have a badge 

355 with session_scope() as session: 

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

357 

358 update_badges(empty_pb2.Empty()) 

359 refresh_materialized_views_rapid(empty_pb2.Empty()) 

360 

361 with api_session(token) as api: 

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

363 assert "strong_verification" in res.badges 

364 assert res.has_strong_verification 

365 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

366 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

367 assert ( 

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

369 == res.has_strong_verification 

370 ) 

371 with account_session(token) as account: 

372 assert not any( 

373 reminder.HasField("complete_verification_reminder") 

374 for reminder in account.GetReminders(empty_pb2.Empty()).reminders 

375 ) 

376 

377 # check has_passport_sex_gender_exception 

378 with real_admin_session(superuser_token) as admin: 

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

380 assert "strong_verification" in res.badges 

381 assert res.has_strong_verification 

382 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

383 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

384 

385 admin.SetPassportSexGenderException( 

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

387 ) 

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

389 

390 update_badges(empty_pb2.Empty()) 

391 refresh_materialized_views_rapid(empty_pb2.Empty()) 

392 

393 with api_session(token) as api: 

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

395 assert "strong_verification" in res.badges 

396 assert res.has_strong_verification 

397 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

398 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

399 assert ( 

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

401 == res.has_strong_verification 

402 ) 

403 with account_session(token) as account: 

404 assert not any( 

405 reminder.HasField("complete_verification_reminder") 

406 for reminder in account.GetReminders(empty_pb2.Empty()).reminders 

407 ) 

408 

409 with real_admin_session(superuser_token) as admin: 

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

411 assert "strong_verification" in res.badges 

412 assert res.has_strong_verification 

413 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

414 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

415 

416 # now turn exception off 

417 admin.SetPassportSexGenderException( 

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

419 ) 

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" not in res.badges 

427 assert not 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_MISMATCH 

430 assert ( 

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

432 == res.has_strong_verification 

433 ) 

434 with account_session(token) as account: 

435 assert any( 

436 reminder.HasField("complete_verification_reminder") 

437 for reminder in account.GetReminders(empty_pb2.Empty()).reminders 

438 ) 

439 

440 with real_admin_session(superuser_token) as admin: 

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

442 assert "strong_verification" not in res.badges 

443 assert not res.has_strong_verification 

444 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

445 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_MISMATCH 

446 

447 

448def test_strong_verification_delete_data(db, monkeypatch): 

449 monkeypatch_sv_config(monkeypatch) 

450 

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

452 _, superuser_token = generate_user(is_superuser=True) 

453 

454 refresh_materialized_views_rapid(empty_pb2.Empty()) 

455 

456 with api_session(token) as api: 

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

458 assert ( 

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

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

461 ) 

462 

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

464 with account_session(token) as account: 

465 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

466 

467 do_and_check_sv( 

468 user, 

469 token, 

470 verification_id=5731012934821983, 

471 sex="MALE", 

472 dob="1988-01-01", 

473 document_type="PASSPORT", 

474 document_number="31195855", 

475 document_expiry=default_expiry, 

476 nationality="US", 

477 ) 

478 

479 refresh_materialized_views_rapid(empty_pb2.Empty()) 

480 

481 # the user should now have strong verification 

482 with api_session(token) as api: 

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

484 assert ( 

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

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

487 ) 

488 

489 # check removing SV data 

490 with account_session(token) as account: 

491 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

492 

493 refresh_materialized_views_rapid(empty_pb2.Empty()) 

494 

495 with api_session(token) as api: 

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

497 assert ( 

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

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

500 ) 

501 

502 with session_scope() as session: 

503 assert ( 

504 len( 

505 session.execute( 

506 select(StrongVerificationAttempt).where( 

507 or_( 

508 StrongVerificationAttempt.passport_encrypted_data != None, 

509 StrongVerificationAttempt.passport_date_of_birth != None, 

510 StrongVerificationAttempt.passport_sex != None, 

511 ) 

512 ) 

513 ) 

514 .scalars() 

515 .all() 

516 ) 

517 == 0 

518 ) 

519 

520 

521def test_strong_verification_expiry(db, monkeypatch): 

522 monkeypatch_sv_config(monkeypatch) 

523 

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

525 _, superuser_token = generate_user(is_superuser=True) 

526 

527 refresh_materialized_views_rapid(empty_pb2.Empty()) 

528 

529 with api_session(token) as api: 

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

531 assert ( 

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

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

534 ) 

535 

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

537 

538 do_and_check_sv( 

539 user, 

540 token, 

541 verification_id=5731012934821983, 

542 sex="MALE", 

543 dob="1988-01-01", 

544 document_type="PASSPORT", 

545 document_number="31195855", 

546 document_expiry=expiry, 

547 nationality="US", 

548 ) 

549 

550 # the user should now have strong verification 

551 with api_session(token) as api: 

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

553 assert res.has_strong_verification 

554 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_VERIFIED 

555 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_VERIFIED 

556 

557 with session_scope() as session: 

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

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

560 

561 with api_session(token) as api: 

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

563 assert not res.has_strong_verification 

564 assert res.birthdate_verification_status == api_pb2.BIRTHDATE_VERIFICATION_STATUS_UNVERIFIED 

565 assert res.gender_verification_status == api_pb2.GENDER_VERIFICATION_STATUS_UNVERIFIED 

566 

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

568 assert not res.has_strong_verification 

569 assert not res.has_strong_verification 

570 

571 do_and_check_sv( 

572 user, 

573 token, 

574 verification_id=5731012934821985, 

575 sex="MALE", 

576 dob="1988-01-01", 

577 document_type="PASSPORT", 

578 document_number="PA41323412", 

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

580 nationality="AU", 

581 ) 

582 

583 refresh_materialized_views_rapid(empty_pb2.Empty()) 

584 

585 with api_session(token) as api: 

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

587 assert ( 

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

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

590 ) 

591 

592 

593def test_strong_verification_regression(db, monkeypatch): 

594 monkeypatch_sv_config(monkeypatch) 

595 

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

597 

598 do_and_check_sv( 

599 user, 

600 token, 

601 verification_id=5731012934821983, 

602 sex="MALE", 

603 dob="1988-01-01", 

604 document_type="PASSPORT", 

605 document_number="31195855", 

606 document_expiry=default_expiry, 

607 nationality="US", 

608 return_after="INITIATED", 

609 ) 

610 

611 with api_session(token) as api: 

612 api.Ping(api_pb2.PingReq()) 

613 

614 

615def test_strong_verification_regression2(db, monkeypatch): 

616 monkeypatch_sv_config(monkeypatch) 

617 

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

619 

620 do_and_check_sv( 

621 user, 

622 token, 

623 verification_id=5731012934821983, 

624 sex="MALE", 

625 dob="1988-01-01", 

626 document_type="PASSPORT", 

627 document_number="31195855", 

628 document_expiry=default_expiry, 

629 nationality="US", 

630 return_after="INITIATED", 

631 ) 

632 

633 do_and_check_sv( 

634 user, 

635 token, 

636 verification_id=5731012934821985, 

637 sex="MALE", 

638 dob="1988-01-01", 

639 document_type="PASSPORT", 

640 document_number="PA41323412", 

641 document_expiry=default_expiry, 

642 nationality="AU", 

643 ) 

644 

645 refresh_materialized_views_rapid(empty_pb2.Empty()) 

646 

647 with api_session(token) as api: 

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

649 assert ( 

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

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

652 ) 

653 

654 

655def test_strong_verification_disabled(db): 

656 user, token = generate_user() 

657 

658 with account_session(token) as account: 

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

660 account.InitiateStrongVerification(empty_pb2.Empty()) 

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

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

663 

664 

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

666 monkeypatch_sv_config(monkeypatch) 

667 

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

669 _, superuser_token = generate_user(is_superuser=True) 

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 do_and_check_sv( 

681 user, 

682 token, 

683 verification_id=5731012934821983, 

684 sex="MALE", 

685 dob="1988-01-01", 

686 document_type="PASSPORT", 

687 document_number="31195855", 

688 document_expiry=default_expiry, 

689 nationality="US", 

690 ) 

691 

692 refresh_materialized_views_rapid(empty_pb2.Empty()) 

693 

694 # the user should now have strong verification 

695 with api_session(token) as api: 

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

697 assert ( 

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

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

700 ) 

701 

702 # check removing SV data 

703 with account_session(token) as account: 

704 account.DeleteStrongVerificationData(empty_pb2.Empty()) 

705 

706 refresh_materialized_views_rapid(empty_pb2.Empty()) 

707 

708 with api_session(token) as api: 

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

710 assert ( 

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

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

713 ) 

714 

715 with session_scope() as session: 

716 assert ( 

717 len( 

718 session.execute( 

719 select(StrongVerificationAttempt).where( 

720 or_( 

721 StrongVerificationAttempt.passport_encrypted_data != None, 

722 StrongVerificationAttempt.passport_date_of_birth != None, 

723 StrongVerificationAttempt.passport_sex != None, 

724 ) 

725 ) 

726 ) 

727 .scalars() 

728 .all() 

729 ) 

730 == 0 

731 ) 

732 

733 reference_data = do_and_check_sv( 

734 user, 

735 token, 

736 verification_id=5731012934821984, 

737 sex="MALE", 

738 dob="1988-01-01", 

739 document_type="PASSPORT", 

740 document_number="31195855", 

741 document_expiry=default_expiry, 

742 nationality="US", 

743 return_after="APPROVED", 

744 ) 

745 

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

747 json_resp2 = { 

748 "id": 5731012934821984, 

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

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

751 "state": "APPROVED", 

752 "reference": reference_data, 

753 "user_ip": "10.123.123.123", 

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

755 "given_names": "John Wayne", 

756 "surname": "Doe", 

757 "nationality": "US", 

758 "sex": "MALE", 

759 "date_of_birth": "1988-01-01", 

760 "document_type": "PASSPORT", 

761 "document_number": "31195855", 

762 "expiry_date": default_expiry.isoformat(), 

763 "issuing_country": "US", 

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

765 "portrait": "dGVzdHRlc3R0ZXN0...", 

766 } 

767 mock.return_value = type( 

768 "__MockResponse", 

769 (), 

770 { 

771 "status_code": 200, 

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

773 "json": lambda: json_resp2, 

774 }, 

775 ) 

776 while process_job(): 

777 pass 

778