Coverage for app/backend/src/couchers/servicers/account.py: 92%

1import json 

2import logging 

3from datetime import UTC, datetime, timedelta 

4from urllib.parse import urlencode 

5 

6import grpc 

7import requests 

8from google.protobuf import empty_pb2 

9from sqlalchemy import select 

10from sqlalchemy.orm import Session 

11from sqlalchemy.sql import exists, func, update 

12from user_agents import parse as user_agents_parse 

13 

14from couchers import urls 

15from couchers.config import config 

16from couchers.constants import PHONE_REVERIFICATION_INTERVAL, SMS_CODE_ATTEMPTS, SMS_CODE_LIFETIME 

17from couchers.context import CouchersContext 

18from couchers.crypto import ( 

19 b64decode, 

20 b64encode, 

21 generate_invite_code, 

22 hash_password, 

23 simple_decrypt, 

24 simple_encrypt, 

25 urlsafe_secure_token, 

26 verify_password, 

27 verify_token, 

28) 

29from couchers.event_log import log_event 

30from couchers.helpers.completed_profile import has_completed_profile 

31from couchers.helpers.geoip import geoip_approximate_location 

32from couchers.helpers.strong_verification import get_strong_verification_fields 

33from couchers.jobs.enqueue import queue_job 

34from couchers.jobs.handlers import finalize_strong_verification 

35from couchers.materialized_views import LiteUser 

36from couchers.metrics import ( 

37 account_deletion_initiations_counter, 

38 strong_verification_data_deletions_counter, 

39 strong_verification_initiations_counter, 

40) 

41from couchers.models import ( 

42 AccountDeletionReason, 

43 AccountDeletionToken, 

44 ContributeOption, 

45 ContributorForm, 

46 HostingStatus, 

47 HostRequest, 

48 HostRequestStatus, 

49 InviteCode, 

50 Message, 

51 ModNote, 

52 ProfilePublicVisibility, 

53 StrongVerificationAttempt, 

54 StrongVerificationAttemptStatus, 

55 StrongVerificationCallbackEvent, 

56 User, 

57 UserSession, 

58 Volunteer, 

59) 

60from couchers.models.notifications import NotificationTopicAction 

61from couchers.notifications.notify import notify 

62from couchers.phone import sms 

63from couchers.phone.check import is_e164_format, is_known_operator 

64from couchers.proto import account_pb2, account_pb2_grpc, auth_pb2, iris_pb2_grpc, notification_data_pb2 

65from couchers.proto.google.api import httpbody_pb2 

66from couchers.proto.internal import internal_pb2, jobs_pb2 

67from couchers.servicers.api import lite_user_to_pb 

68from couchers.servicers.public import format_volunteer_link 

69from couchers.servicers.references import get_pending_references_to_write, reftype2api 

70from couchers.sql import where_moderated_content_visible, where_users_column_visible 

71from couchers.tasks import ( 

72 maybe_send_contributor_form_email, 

73 send_account_deletion_report_email, 

74 send_email_changed_confirmation_to_new_email, 

75) 

76from couchers.utils import ( 

77 Timestamp_from_datetime, 

78 create_lang_cookie, 

79 date_to_api, 

80 dt_from_page_token, 

81 dt_to_page_token, 

82 is_valid_email, 

83 now, 

84 to_aware_datetime, 

85) 

86 

87logger = logging.getLogger(__name__) 

88logger.setLevel(logging.DEBUG) 

89 

90contributeoption2sql = { 

91 auth_pb2.CONTRIBUTE_OPTION_UNSPECIFIED: None, 

92 auth_pb2.CONTRIBUTE_OPTION_YES: ContributeOption.yes, 

93 auth_pb2.CONTRIBUTE_OPTION_MAYBE: ContributeOption.maybe, 

94 auth_pb2.CONTRIBUTE_OPTION_NO: ContributeOption.no, 

95} 

96 

97contributeoption2api = { 

98 None: auth_pb2.CONTRIBUTE_OPTION_UNSPECIFIED, 

99 ContributeOption.yes: auth_pb2.CONTRIBUTE_OPTION_YES, 

100 ContributeOption.maybe: auth_pb2.CONTRIBUTE_OPTION_MAYBE, 

101 ContributeOption.no: auth_pb2.CONTRIBUTE_OPTION_NO, 

102} 

103 

104profilepublicitysetting2sql = { 

105 account_pb2.PROFILE_PUBLIC_VISIBILITY_UNKNOWN: None, 

106 account_pb2.PROFILE_PUBLIC_VISIBILITY_NOTHING: ProfilePublicVisibility.nothing, 

107 account_pb2.PROFILE_PUBLIC_VISIBILITY_MAP_ONLY: ProfilePublicVisibility.map_only, 

108 account_pb2.PROFILE_PUBLIC_VISIBILITY_LIMITED: ProfilePublicVisibility.limited, 

109 account_pb2.PROFILE_PUBLIC_VISIBILITY_MOST: ProfilePublicVisibility.most, 

110 account_pb2.PROFILE_PUBLIC_VISIBILITY_FULL: ProfilePublicVisibility.full, 

111} 

112 

113profilepublicitysetting2api = { 

114 None: account_pb2.PROFILE_PUBLIC_VISIBILITY_UNKNOWN, 

115 ProfilePublicVisibility.nothing: account_pb2.PROFILE_PUBLIC_VISIBILITY_NOTHING, 

116 ProfilePublicVisibility.map_only: account_pb2.PROFILE_PUBLIC_VISIBILITY_MAP_ONLY, 

117 ProfilePublicVisibility.limited: account_pb2.PROFILE_PUBLIC_VISIBILITY_LIMITED, 

118 ProfilePublicVisibility.most: account_pb2.PROFILE_PUBLIC_VISIBILITY_MOST, 

119 ProfilePublicVisibility.full: account_pb2.PROFILE_PUBLIC_VISIBILITY_FULL, 

120} 

121 

122MAX_PAGINATION_LENGTH = 50 

123 

124 

125def mod_note_to_pb(note: ModNote) -> account_pb2.ModNote: 

126 return account_pb2.ModNote( 

127 note_id=note.id, 

128 note_content=note.note_content, 

129 created=Timestamp_from_datetime(note.created), 

130 acknowledged=Timestamp_from_datetime(note.acknowledged) if note.acknowledged else None, 

131 ) 

132 

133 

134def abort_on_invalid_password(password: str, context: CouchersContext) -> None: 

135 """ 

136 Internal utility function: given a password, aborts if password is unforgivably insecure 

137 """ 

138 if len(password) < 8: 

139 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "password_too_short") 

140 

141 if len(password) > 256: 

142 # Hey, what are you trying to do? Give us a DDOS attack? 

143 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "password_too_long") 

144 

145 # check for the most common weak passwords (not meant to be an exhaustive check!) 

146 if password.lower() in ("password", "12345678", "couchers", "couchers1"): 

147 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "insecure_password") 

148 

149 

150def _volunteer_info_to_pb(volunteer: Volunteer, username: str) -> account_pb2.GetMyVolunteerInfoRes: 

151 return account_pb2.GetMyVolunteerInfoRes( 

152 display_name=volunteer.display_name, 

153 display_location=volunteer.display_location, 

154 role=volunteer.role, 

155 started_volunteering=date_to_api(volunteer.started_volunteering), 

156 stopped_volunteering=date_to_api(volunteer.stopped_volunteering) if volunteer.stopped_volunteering else None, 

157 show_on_team_page=volunteer.show_on_team_page, 

158 **format_volunteer_link(volunteer, username), 

159 ) 

160 

161 

162class Account(account_pb2_grpc.AccountServicer): 

163 def GetAccountInfo( 

164 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

165 ) -> account_pb2.GetAccountInfoRes: 

166 user, volunteer = session.execute( 

167 select(User, Volunteer).outerjoin(Volunteer, Volunteer.user_id == User.id).where(User.id == context.user_id) 

168 ).one() 

169 

170 # Test experimentation integration - check if user is in the test gate 

171 # Create 'test_growthbook_integration' in GrowthBook to test 

172 test_gate = context.get_boolean_value("test_growthbook_integration", default=False) 

173 logger.info(f"Experimentation gate 'test_growthbook_integration' for user {user.id}: {test_gate}") 

174 

175 # The donation drive (and its banner) is controlled by the donation_drive_start flag: a Unix 

176 # epoch in seconds when a drive is running, or 0/unset when there's no drive. Users who haven't 

177 # donated since the drive started see the banner. 

178 drive_start_epoch = context.get_integer_value("donation_drive_start", 0) 

179 drive_start = datetime.fromtimestamp(drive_start_epoch, tz=UTC) if drive_start_epoch else None 

180 should_show_donation_banner = drive_start is not None and ( 

181 user.last_donated is None or user.last_donated < drive_start 

182 ) 

183 

184 return account_pb2.GetAccountInfoRes( 

185 username=user.username, 

186 email=user.email, 

187 phone=user.phone if (user.phone_is_verified or not user.phone_code_expired) else None, 

188 has_donated=user.last_donated is not None, 

189 phone_verified=user.phone_is_verified, 

190 profile_complete=has_completed_profile(session, user), 

191 my_home_complete=user.has_completed_my_home, 

192 timezone=user.timezone, 

193 is_superuser=user.is_superuser, 

194 ui_language_preference=user.ui_language_preference, 

195 profile_public_visibility=profilepublicitysetting2api[user.public_visibility], 

196 is_volunteer=volunteer is not None, 

197 should_show_donation_banner=should_show_donation_banner, 

198 **get_strong_verification_fields(session, user), 

199 ) 

200 

201 def ChangePasswordV2( 

202 self, request: account_pb2.ChangePasswordV2Req, context: CouchersContext, session: Session 

203 ) -> empty_pb2.Empty: 

204 """ 

205 Changes the user's password. They have to confirm their old password just in case. 

206 

207 If they didn't have an old password previously, then we don't check that. 

208 """ 

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

210 

211 if not verify_password(user.hashed_password, request.old_password): 

212 # wrong password 

213 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_password") 

214 

215 abort_on_invalid_password(request.new_password, context) 

216 user.hashed_password = hash_password(request.new_password) 

217 

218 session.commit() 

219 

220 notify( 

221 session, 

222 user_id=user.id, 

223 topic_action=NotificationTopicAction.password__change, 

224 key="", 

225 ) 

226 log_event(context, session, "account.password_changed", {}) 

227 

228 return empty_pb2.Empty() 

229 

230 def ChangeEmailV2( 

231 self, request: account_pb2.ChangeEmailV2Req, context: CouchersContext, session: Session 

232 ) -> empty_pb2.Empty: 

233 """ 

234 Change the user's email address. 

235 

236 If the user has a password, a notification is sent to the old email, and a confirmation is sent to the new one. 

237 

238 Otherwise they need to confirm twice, via an email sent to each of their old and new emails. 

239 

240 In all confirmation emails, the user must click on the confirmation link. 

241 """ 

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

243 

244 # check password first 

245 if not verify_password(user.hashed_password, request.password): 

246 # wrong password 

247 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_password") 

248 

249 # not a valid email 

250 if not is_valid_email(request.new_email): 

251 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_email") 

252 

253 # email already in use (possibly by this user) 

254 if session.execute(select(User).where(User.email == request.new_email)).scalar_one_or_none(): 

255 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_email") 

256 

257 user.new_email = request.new_email 

258 user.new_email_token = urlsafe_secure_token() 

259 user.new_email_token_created = now() 

260 user.new_email_token_expiry = now() + timedelta(hours=2) 

261 

262 send_email_changed_confirmation_to_new_email(context, session, user) 

263 

264 # will still go into old email 

265 notify( 

266 session, 

267 user_id=user.id, 

268 topic_action=NotificationTopicAction.email_address__change, 

269 key="", 

270 data=notification_data_pb2.EmailAddressChange( 

271 new_email=request.new_email, 

272 ), 

273 ) 

274 

275 log_event(context, session, "account.email_change_initiated", {}) 

276 

277 # session autocommit 

278 return empty_pb2.Empty() 

279 

280 def ChangeLanguagePreference( 

281 self, request: account_pb2.ChangeLanguagePreferenceReq, context: CouchersContext, session: Session 

282 ) -> empty_pb2.Empty: 

283 # select the user from the db 

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

285 

286 # update the user's preference 

287 user.ui_language_preference = request.ui_language_preference 

288 context.set_cookies(create_lang_cookie(request.ui_language_preference)) 

289 

290 return empty_pb2.Empty() 

291 

292 def FillContributorForm( 

293 self, request: account_pb2.FillContributorFormReq, context: CouchersContext, session: Session 

294 ) -> empty_pb2.Empty: 

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

296 

297 form = request.contributor_form 

298 

299 form = ContributorForm( 

300 user_id=user.id, 

301 ideas=form.ideas or None, 

302 features=form.features or None, 

303 experience=form.experience or None, 

304 contribute=contributeoption2sql[form.contribute], 

305 contribute_ways=form.contribute_ways, 

306 expertise=form.expertise or None, 

307 ) 

308 

309 session.add(form) 

310 session.flush() 

311 maybe_send_contributor_form_email(session, form) 

312 

313 user.filled_contributor_form = True 

314 log_event(context, session, "contributor.form_submitted", {"is_filled": form.is_filled}) 

315 

316 return empty_pb2.Empty() 

317 

318 def GetContributorFormInfo( 

319 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

320 ) -> account_pb2.GetContributorFormInfoRes: 

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

322 

323 return account_pb2.GetContributorFormInfoRes( 

324 filled_contributor_form=user.filled_contributor_form, 

325 ) 

326 

327 def ChangePhone( 

328 self, request: account_pb2.ChangePhoneReq, context: CouchersContext, session: Session 

329 ) -> empty_pb2.Empty: 

330 phone = request.phone 

331 # early quick validation 

332 if phone and not is_e164_format(phone): 

333 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_phone") 

334 

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

336 if user.last_donated is None: 

337 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "not_donated") 

338 

339 if not phone: 

340 user.phone = None 

341 user.phone_verification_verified = None 

342 user.phone_verification_token = None 

343 user.phone_verification_attempts = 0 

344 return empty_pb2.Empty() 

345 

346 # Removing a number is always allowed; sending a verification SMS is gated. 

347 if not context.get_boolean_value("sms_enabled", default=False): 

348 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "sms_disabled") 

349 

350 if not is_known_operator(phone): 

351 context.abort_with_error_code(grpc.StatusCode.UNIMPLEMENTED, "unrecognized_phone_number") 

352 

353 if now() - user.phone_verification_sent < PHONE_REVERIFICATION_INTERVAL: 

354 context.abort_with_error_code(grpc.StatusCode.RESOURCE_EXHAUSTED, "reverification_too_early") 

355 

356 token = sms.generate_random_code() 

357 result = sms.send_sms(phone, sms.format_message(token)) 

358 

359 if result == "success": 

360 user.phone = phone 

361 user.phone_verification_verified = None 

362 user.phone_verification_token = token 

363 user.phone_verification_sent = now() 

364 user.phone_verification_attempts = 0 

365 

366 notify( 

367 session, 

368 user_id=user.id, 

369 topic_action=NotificationTopicAction.phone_number__change, 

370 key="", 

371 data=notification_data_pb2.PhoneNumberChange( 

372 phone=phone, 

373 ), 

374 ) 

375 

376 return empty_pb2.Empty() 

377 

378 context.abort(grpc.StatusCode.UNIMPLEMENTED, result) 

379 

380 def VerifyPhone( 

381 self, request: account_pb2.VerifyPhoneReq, context: CouchersContext, session: Session 

382 ) -> empty_pb2.Empty: 

383 if not sms.looks_like_a_code(request.token): 383 ↛ 384line 383 didn't jump to line 384 because the condition on line 383 was never true

384 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "wrong_sms_code") 

385 

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

387 if user.phone_verification_token is None: 

388 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "no_pending_verification") 

389 

390 if now() - user.phone_verification_sent > SMS_CODE_LIFETIME: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true

391 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "no_pending_verification") 

392 

393 if user.phone_verification_attempts > SMS_CODE_ATTEMPTS: 

394 context.abort_with_error_code(grpc.StatusCode.RESOURCE_EXHAUSTED, "too_many_sms_code_attempts") 

395 

396 if not verify_token(request.token, user.phone_verification_token): 

397 user.phone_verification_attempts += 1 

398 session.commit() 

399 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "wrong_sms_code") 

400 

401 # Delete verifications from everyone else that has this number 

402 session.execute( 

403 update(User) 

404 .where(User.phone == user.phone) 

405 .where(User.id != context.user_id) 

406 .values( 

407 { 

408 "phone_verification_verified": None, 

409 "phone_verification_attempts": 0, 

410 "phone_verification_token": None, 

411 "phone": None, 

412 } 

413 ) 

414 .execution_options(synchronize_session=False) 

415 ) 

416 

417 user.phone_verification_token = None 

418 user.phone_verification_verified = now() 

419 user.phone_verification_attempts = 0 

420 

421 notify( 

422 session, 

423 user_id=user.id, 

424 topic_action=NotificationTopicAction.phone_number__verify, 

425 key="", 

426 data=notification_data_pb2.PhoneNumberVerify( 

427 phone=user.phone, 

428 ), 

429 ) 

430 

431 return empty_pb2.Empty() 

432 

433 def InitiateStrongVerification( 

434 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

435 ) -> account_pb2.InitiateStrongVerificationRes: 

436 if not context.get_boolean_value("strong_verification_enabled", default=False): 

437 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "strong_verification_disabled") 

438 

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

440 existing_verification = session.execute( 

441 select(StrongVerificationAttempt) 

442 .where(StrongVerificationAttempt.user_id == user.id) 

443 .where(StrongVerificationAttempt.is_valid) 

444 ).scalar_one_or_none() 

445 if existing_verification: 445 ↛ 446line 445 didn't jump to line 446 because the condition on line 445 was never true

446 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "strong_verification_already_verified") 

447 

448 strong_verification_initiations_counter.labels(user.gender).inc() 

449 log_event(context, session, "verification.strong_initiated", {"gender": user.gender}) 

450 

451 verification_attempt_token = urlsafe_secure_token() 

452 # this is the iris reference data, they will return this on every callback, it also doubles as webhook auth given lack of it otherwise 

453 reference = b64encode( 

454 simple_encrypt( 

455 "iris_callback", 

456 internal_pb2.VerificationReferencePayload( 

457 verification_attempt_token=verification_attempt_token, 

458 user_id=user.id, 

459 ).SerializeToString(), 

460 ) 

461 ) 

462 response = requests.post( 

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

464 auth=(config.IRIS_ID_PUBKEY, config.IRIS_ID_SECRET), 

465 json={ 

466 "callback_url": f"{config.BACKEND_BASE_URL}/iris/webhook", 

467 "face_verification": False, 

468 "passport_only": True, 

469 "reference": reference, 

470 }, 

471 timeout=10, 

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

473 ) 

474 

475 if response.status_code != 200: 475 ↛ 476line 475 didn't jump to line 476 because the condition on line 475 was never true

476 raise Exception(f"Iris didn't return 200: {response.text}") 

477 

478 iris_session_id = response.json()["id"] 

479 token = response.json()["token"] 

480 session.add( 

481 StrongVerificationAttempt( 

482 user_id=user.id, 

483 verification_attempt_token=verification_attempt_token, 

484 iris_session_id=iris_session_id, 

485 iris_token=token, 

486 ) 

487 ) 

488 

489 redirect_params = { 

490 "token": token, 

491 "redirect_url": urls.complete_strong_verification_url( 

492 verification_attempt_token=verification_attempt_token 

493 ), 

494 } 

495 redirect_url = "https://passportreader.app/open?" + urlencode(redirect_params) 

496 

497 return account_pb2.InitiateStrongVerificationRes( 

498 verification_attempt_token=verification_attempt_token, 

499 redirect_url=redirect_url, 

500 ) 

501 

502 def GetStrongVerificationAttemptStatus( 

503 self, request: account_pb2.GetStrongVerificationAttemptStatusReq, context: CouchersContext, session: Session 

504 ) -> account_pb2.GetStrongVerificationAttemptStatusRes: 

505 verification_attempt = session.execute( 

506 select(StrongVerificationAttempt) 

507 .where(StrongVerificationAttempt.user_id == context.user_id) 

508 .where(StrongVerificationAttempt.is_visible) 

509 .where(StrongVerificationAttempt.verification_attempt_token == request.verification_attempt_token) 

510 ).scalar_one_or_none() 

511 if not verification_attempt: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true

512 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "strong_verification_attempt_not_found") 

513 status_to_pb = { 

514 StrongVerificationAttemptStatus.succeeded: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_SUCCEEDED, 

515 StrongVerificationAttemptStatus.in_progress_waiting_on_user_to_open_app: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_TO_OPEN_APP, 

516 StrongVerificationAttemptStatus.in_progress_waiting_on_user_in_app: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_IN_APP, 

517 StrongVerificationAttemptStatus.in_progress_waiting_on_backend: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND, 

518 StrongVerificationAttemptStatus.failed: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_FAILED, 

519 StrongVerificationAttemptStatus.duplicate: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_FAILED, 

520 } 

521 return account_pb2.GetStrongVerificationAttemptStatusRes( 

522 status=status_to_pb.get( 

523 verification_attempt.status, account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_UNKNOWN 

524 ), 

525 ) 

526 

527 def DeleteStrongVerificationData( 

528 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

529 ) -> empty_pb2.Empty: 

530 verification_attempts = ( 

531 session.execute( 

532 select(StrongVerificationAttempt) 

533 .where(StrongVerificationAttempt.user_id == context.user_id) 

534 .where(StrongVerificationAttempt.has_full_data) 

535 ) 

536 .scalars() 

537 .all() 

538 ) 

539 for verification_attempt in verification_attempts: 

540 verification_attempt.status = StrongVerificationAttemptStatus.deleted 

541 verification_attempt.has_full_data = False 

542 verification_attempt.passport_encrypted_data = None 

543 verification_attempt.passport_date_of_birth = None 

544 verification_attempt.passport_sex = None 

545 session.flush() 

546 # double check: 

547 verification_attempts = ( 

548 session.execute( 

549 select(StrongVerificationAttempt) 

550 .where(StrongVerificationAttempt.user_id == context.user_id) 

551 .where(StrongVerificationAttempt.has_full_data) 

552 ) 

553 .scalars() 

554 .all() 

555 ) 

556 assert len(verification_attempts) == 0 

557 

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

559 strong_verification_data_deletions_counter.labels(user.gender).inc() 

560 log_event(context, session, "verification.strong_data_deleted", {"gender": user.gender}) 

561 

562 return empty_pb2.Empty() 

563 

564 def DeleteAccount( 

565 self, request: account_pb2.DeleteAccountReq, context: CouchersContext, session: Session 

566 ) -> empty_pb2.Empty: 

567 """ 

568 Triggers email with token to confirm deletion 

569 

570 Frontend should confirm via unique string (i.e. username) before this is called 

571 """ 

572 if not request.confirm: 

573 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "must_confirm_account_delete") 

574 

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

576 

577 reason = request.reason.strip() 

578 if reason: 

579 deletion_reason = AccountDeletionReason(user_id=user.id, reason=reason) 

580 session.add(deletion_reason) 

581 session.flush() 

582 send_account_deletion_report_email(session, deletion_reason) 

583 

584 token = AccountDeletionToken(token=urlsafe_secure_token(), user_id=user.id, expiry=now() + timedelta(hours=2)) 

585 

586 notify( 

587 session, 

588 user_id=user.id, 

589 topic_action=NotificationTopicAction.account_deletion__start, 

590 key="", 

591 data=notification_data_pb2.AccountDeletionStart( 

592 deletion_token=token.token, 

593 ), 

594 ) 

595 session.add(token) 

596 

597 account_deletion_initiations_counter.labels(user.gender).inc() 

598 log_event(context, session, "account.deletion_initiated", {"gender": user.gender, "has_reason": bool(reason)}) 

599 

600 return empty_pb2.Empty() 

601 

602 def ListModNotes( 

603 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

604 ) -> account_pb2.ListModNotesRes: 

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

606 

607 notes = ( 

608 session.execute(select(ModNote).where(ModNote.user_id == user.id).order_by(ModNote.created.asc())) 

609 .scalars() 

610 .all() 

611 ) 

612 

613 return account_pb2.ListModNotesRes(mod_notes=[mod_note_to_pb(note) for note in notes]) 

614 

615 def ListActiveSessions( 

616 self, request: account_pb2.ListActiveSessionsReq, context: CouchersContext, session: Session 

617 ) -> account_pb2.ListActiveSessionsRes: 

618 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

619 page_token = dt_from_page_token(request.page_token) if request.page_token else now() 

620 

621 user_sessions = ( 

622 session.execute( 

623 select(UserSession) 

624 .where(UserSession.user_id == context.user_id) 

625 .where(UserSession.is_valid) 

626 .where(UserSession.is_api_key == False) 

627 .where(UserSession.last_seen <= page_token) 

628 .order_by(UserSession.last_seen.desc()) 

629 .limit(page_size + 1) 

630 ) 

631 .scalars() 

632 .all() 

633 ) 

634 

635 def _active_session_to_pb(user_session: UserSession) -> account_pb2.ActiveSession: 

636 user_agent = user_agents_parse(user_session.user_agent or "") 

637 return account_pb2.ActiveSession( 

638 created=Timestamp_from_datetime(user_session.created), 

639 expiry=Timestamp_from_datetime(user_session.expiry), 

640 last_seen=Timestamp_from_datetime(user_session.last_seen), 

641 operating_system=user_agent.os.family, 

642 browser=user_agent.browser.family, 

643 device=user_agent.device.family, 

644 approximate_location=geoip_approximate_location(user_session.ip_address) or "Unknown", 

645 is_current_session=user_session.token == context.token, 

646 ) 

647 

648 return account_pb2.ListActiveSessionsRes( 

649 active_sessions=list(map(_active_session_to_pb, user_sessions[:page_size])), 

650 next_page_token=dt_to_page_token(user_sessions[-1].last_seen) if len(user_sessions) > page_size else None, 

651 ) 

652 

653 def LogOutSession( 

654 self, request: account_pb2.LogOutSessionReq, context: CouchersContext, session: Session 

655 ) -> empty_pb2.Empty: 

656 session.execute( 

657 update(UserSession) 

658 .where(UserSession.token != context.token) 

659 .where(UserSession.user_id == context.user_id) 

660 .where(UserSession.is_valid) 

661 .where(UserSession.is_api_key == False) 

662 .where(UserSession.created == to_aware_datetime(request.created)) 

663 .values(expiry=func.now()) 

664 .execution_options(synchronize_session=False) 

665 ) 

666 return empty_pb2.Empty() 

667 

668 def LogOutOtherSessions( 

669 self, request: account_pb2.LogOutOtherSessionsReq, context: CouchersContext, session: Session 

670 ) -> empty_pb2.Empty: 

671 if not request.confirm: 

672 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "must_confirm_logout_other_sessions") 

673 

674 session.execute( 

675 update(UserSession) 

676 .where(UserSession.token != context.token) 

677 .where(UserSession.user_id == context.user_id) 

678 .where(UserSession.is_valid) 

679 .where(UserSession.is_api_key == False) 

680 .values(expiry=func.now()) 

681 .execution_options(synchronize_session=False) 

682 ) 

683 return empty_pb2.Empty() 

684 

685 def SetProfilePublicVisibility( 

686 self, request: account_pb2.SetProfilePublicVisibilityReq, context: CouchersContext, session: Session 

687 ) -> empty_pb2.Empty: 

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

689 user.public_visibility = profilepublicitysetting2sql[request.profile_public_visibility] # type: ignore[assignment] 

690 user.has_modified_public_visibility = True 

691 return empty_pb2.Empty() 

692 

693 def CreateInviteCode( 

694 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

695 ) -> account_pb2.CreateInviteCodeRes: 

696 code = generate_invite_code() 

697 session.add(InviteCode(id=code, creator_user_id=context.user_id)) 

698 

699 return account_pb2.CreateInviteCodeRes( 

700 code=code, 

701 url=urls.invite_code_link(code=code), 

702 ) 

703 

704 def DisableInviteCode( 

705 self, request: account_pb2.DisableInviteCodeReq, context: CouchersContext, session: Session 

706 ) -> empty_pb2.Empty: 

707 invite = session.execute( 

708 select(InviteCode).where(InviteCode.id == request.code, InviteCode.creator_user_id == context.user_id) 

709 ).scalar_one_or_none() 

710 

711 if not invite: 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true

712 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "not_found") 

713 

714 invite.disabled = func.now() 

715 session.commit() 

716 

717 return empty_pb2.Empty() 

718 

719 def ListInviteCodes( 

720 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

721 ) -> account_pb2.ListInviteCodesRes: 

722 results = session.execute( 

723 select( 

724 InviteCode.id, 

725 InviteCode.created, 

726 InviteCode.disabled, 

727 func.count(User.id).label("num_users"), 

728 ) 

729 .outerjoin(User, User.invite_code_id == InviteCode.id) 

730 .where(InviteCode.creator_user_id == context.user_id) 

731 .group_by(InviteCode.id, InviteCode.disabled) 

732 .order_by(func.count(User.id).desc(), InviteCode.disabled) 

733 ).all() 

734 

735 return account_pb2.ListInviteCodesRes( 

736 invite_codes=[ 

737 account_pb2.InviteCodeInfo( 

738 code=code_id, 

739 created=Timestamp_from_datetime(created), 

740 disabled=Timestamp_from_datetime(disabled) if disabled else None, 

741 uses=len_users, 

742 url=urls.invite_code_link(code=code_id), 

743 ) 

744 for code_id, created, disabled, len_users in results 

745 ] 

746 ) 

747 

748 def GetReminders(