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 } 

520 return account_pb2.GetStrongVerificationAttemptStatusRes( 

521 status=status_to_pb.get( 

522 verification_attempt.status, account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_UNKNOWN 

523 ), 

524 ) 

525 

526 def DeleteStrongVerificationData( 

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

528 ) -> empty_pb2.Empty: 

529 verification_attempts = ( 

530 session.execute( 

531 select(StrongVerificationAttempt) 

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

533 .where(StrongVerificationAttempt.has_full_data) 

534 ) 

535 .scalars() 

536 .all() 

537 ) 

538 for verification_attempt in verification_attempts: 

539 verification_attempt.status = StrongVerificationAttemptStatus.deleted 

540 verification_attempt.has_full_data = False 

541 verification_attempt.passport_encrypted_data = None 

542 verification_attempt.passport_date_of_birth = None 

543 verification_attempt.passport_sex = None 

544 session.flush() 

545 # double check: 

546 verification_attempts = ( 

547 session.execute( 

548 select(StrongVerificationAttempt) 

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

550 .where(StrongVerificationAttempt.has_full_data) 

551 ) 

552 .scalars() 

553 .all() 

554 ) 

555 assert len(verification_attempts) == 0 

556 

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

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

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

560 

561 return empty_pb2.Empty() 

562 

563 def DeleteAccount( 

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

565 ) -> empty_pb2.Empty: 

566 """ 

567 Triggers email with token to confirm deletion 

568 

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

570 """ 

571 if not request.confirm: 

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

573 

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

575 

576 reason = request.reason.strip() 

577 if reason: 

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

579 session.add(deletion_reason) 

580 session.flush() 

581 send_account_deletion_report_email(session, deletion_reason) 

582 

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

584 

585 notify( 

586 session, 

587 user_id=user.id, 

588 topic_action=NotificationTopicAction.account_deletion__start, 

589 key="", 

590 data=notification_data_pb2.AccountDeletionStart( 

591 deletion_token=token.token, 

592 ), 

593 ) 

594 session.add(token) 

595 

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

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

598 

599 return empty_pb2.Empty() 

600 

601 def ListModNotes( 

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

603 ) -> account_pb2.ListModNotesRes: 

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

605 

606 notes = ( 

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

608 .scalars() 

609 .all() 

610 ) 

611 

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

613 

614 def ListActiveSessions( 

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

616 ) -> account_pb2.ListActiveSessionsRes: 

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

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

619 

620 user_sessions = ( 

621 session.execute( 

622 select(UserSession) 

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

624 .where(UserSession.is_valid) 

625 .where(UserSession.is_api_key == False) 

626 .where(UserSession.last_seen <= page_token) 

627 .order_by(UserSession.last_seen.desc()) 

628 .limit(page_size + 1) 

629 ) 

630 .scalars() 

631 .all() 

632 ) 

633 

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

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

636 return account_pb2.ActiveSession( 

637 created=Timestamp_from_datetime(user_session.created), 

638 expiry=Timestamp_from_datetime(user_session.expiry), 

639 last_seen=Timestamp_from_datetime(user_session.last_seen), 

640 operating_system=user_agent.os.family, 

641 browser=user_agent.browser.family, 

642 device=user_agent.device.family, 

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

644 is_current_session=user_session.token == context.token, 

645 ) 

646 

647 return account_pb2.ListActiveSessionsRes( 

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

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

650 ) 

651 

652 def LogOutSession( 

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

654 ) -> empty_pb2.Empty: 

655 session.execute( 

656 update(UserSession) 

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

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

659 .where(UserSession.is_valid) 

660 .where(UserSession.is_api_key == False) 

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

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

663 .execution_options(synchronize_session=False) 

664 ) 

665 return empty_pb2.Empty() 

666 

667 def LogOutOtherSessions( 

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

669 ) -> empty_pb2.Empty: 

670 if not request.confirm: 

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

672 

673 session.execute( 

674 update(UserSession) 

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

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

677 .where(UserSession.is_valid) 

678 .where(UserSession.is_api_key == False) 

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

680 .execution_options(synchronize_session=False) 

681 ) 

682 return empty_pb2.Empty() 

683 

684 def SetProfilePublicVisibility( 

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

686 ) -> empty_pb2.Empty: 

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

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

689 user.has_modified_public_visibility = True 

690 return empty_pb2.Empty() 

691 

692 def CreateInviteCode( 

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

694 ) -> account_pb2.CreateInviteCodeRes: 

695 code = generate_invite_code() 

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

697 

698 return account_pb2.CreateInviteCodeRes( 

699 code=code, 

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

701 ) 

702 

703 def DisableInviteCode( 

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

705 ) -> empty_pb2.Empty: 

706 invite = session.execute( 

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

708 ).scalar_one_or_none() 

709 

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

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

712 

713 invite.disabled = func.now() 

714 session.commit() 

715 

716 return empty_pb2.Empty() 

717 

718 def ListInviteCodes( 

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

720 ) -> account_pb2.ListInviteCodesRes: 

721 results = session.execute( 

722 select( 

723 InviteCode.id, 

724 InviteCode.created, 

725 InviteCode.disabled, 

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

727 ) 

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

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

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

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

732 ).all() 

733 

734 return account_pb2.ListInviteCodesRes( 

735 invite_codes=[ 

736 account_pb2.InviteCodeInfo( 

737 code=code_id, 

738 created=Timestamp_from_datetime(created), 

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

740 uses=len_users, 

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

742 ) 

743 for code_id, created, disabled, len_users in results 

744 ] 

745 ) 

746 

747 def GetReminders( 

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