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 maybe_send_contributor_form_email, send_email_changed_confirmation_to_new_email 

72from couchers.utils import ( 

73 Timestamp_from_datetime, 

74 create_lang_cookie, 

75 date_to_api, 

76 dt_from_page_token, 

77 dt_to_page_token, 

78 is_valid_email, 

79 now, 

80 to_aware_datetime, 

81) 

82 

83logger = logging.getLogger(__name__) 

84logger.setLevel(logging.DEBUG) 

85 

86contributeoption2sql = { 

87 auth_pb2.CONTRIBUTE_OPTION_UNSPECIFIED: None, 

88 auth_pb2.CONTRIBUTE_OPTION_YES: ContributeOption.yes, 

89 auth_pb2.CONTRIBUTE_OPTION_MAYBE: ContributeOption.maybe, 

90 auth_pb2.CONTRIBUTE_OPTION_NO: ContributeOption.no, 

91} 

92 

93contributeoption2api = { 

94 None: auth_pb2.CONTRIBUTE_OPTION_UNSPECIFIED, 

95 ContributeOption.yes: auth_pb2.CONTRIBUTE_OPTION_YES, 

96 ContributeOption.maybe: auth_pb2.CONTRIBUTE_OPTION_MAYBE, 

97 ContributeOption.no: auth_pb2.CONTRIBUTE_OPTION_NO, 

98} 

99 

100profilepublicitysetting2sql = { 

101 account_pb2.PROFILE_PUBLIC_VISIBILITY_UNKNOWN: None, 

102 account_pb2.PROFILE_PUBLIC_VISIBILITY_NOTHING: ProfilePublicVisibility.nothing, 

103 account_pb2.PROFILE_PUBLIC_VISIBILITY_MAP_ONLY: ProfilePublicVisibility.map_only, 

104 account_pb2.PROFILE_PUBLIC_VISIBILITY_LIMITED: ProfilePublicVisibility.limited, 

105 account_pb2.PROFILE_PUBLIC_VISIBILITY_MOST: ProfilePublicVisibility.most, 

106 account_pb2.PROFILE_PUBLIC_VISIBILITY_FULL: ProfilePublicVisibility.full, 

107} 

108 

109profilepublicitysetting2api = { 

110 None: account_pb2.PROFILE_PUBLIC_VISIBILITY_UNKNOWN, 

111 ProfilePublicVisibility.nothing: account_pb2.PROFILE_PUBLIC_VISIBILITY_NOTHING, 

112 ProfilePublicVisibility.map_only: account_pb2.PROFILE_PUBLIC_VISIBILITY_MAP_ONLY, 

113 ProfilePublicVisibility.limited: account_pb2.PROFILE_PUBLIC_VISIBILITY_LIMITED, 

114 ProfilePublicVisibility.most: account_pb2.PROFILE_PUBLIC_VISIBILITY_MOST, 

115 ProfilePublicVisibility.full: account_pb2.PROFILE_PUBLIC_VISIBILITY_FULL, 

116} 

117 

118MAX_PAGINATION_LENGTH = 50 

119 

120 

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

122 return account_pb2.ModNote( 

123 note_id=note.id, 

124 note_content=note.note_content, 

125 created=Timestamp_from_datetime(note.created), 

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

127 ) 

128 

129 

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

131 """ 

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

133 """ 

134 if len(password) < 8: 

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

136 

137 if len(password) > 256: 

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

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

140 

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

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

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

144 

145 

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

147 return account_pb2.GetMyVolunteerInfoRes( 

148 display_name=volunteer.display_name, 

149 display_location=volunteer.display_location, 

150 role=volunteer.role, 

151 started_volunteering=date_to_api(volunteer.started_volunteering), 

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

153 show_on_team_page=volunteer.show_on_team_page, 

154 **format_volunteer_link(volunteer, username), 

155 ) 

156 

157 

158class Account(account_pb2_grpc.AccountServicer): 

159 def GetAccountInfo( 

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

161 ) -> account_pb2.GetAccountInfoRes: 

162 user, volunteer = session.execute( 

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

164 ).one() 

165 

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

167 # Create 'test_growthbook_integration' in GrowthBook to test 

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

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

170 

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

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

173 # donated since the drive started see the banner. 

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

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

176 should_show_donation_banner = drive_start is not None and ( 

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

178 ) 

179 

180 return account_pb2.GetAccountInfoRes( 

181 username=user.username, 

182 email=user.email, 

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

184 has_donated=user.last_donated is not None, 

185 phone_verified=user.phone_is_verified, 

186 profile_complete=has_completed_profile(session, user), 

187 my_home_complete=user.has_completed_my_home, 

188 timezone=user.timezone, 

189 is_superuser=user.is_superuser, 

190 ui_language_preference=user.ui_language_preference, 

191 profile_public_visibility=profilepublicitysetting2api[user.public_visibility], 

192 is_volunteer=volunteer is not None, 

193 should_show_donation_banner=should_show_donation_banner, 

194 **get_strong_verification_fields(session, user), 

195 ) 

196 

197 def ChangePasswordV2( 

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

199 ) -> empty_pb2.Empty: 

200 """ 

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

202 

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

204 """ 

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

206 

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

208 # wrong password 

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

210 

211 abort_on_invalid_password(request.new_password, context) 

212 user.hashed_password = hash_password(request.new_password) 

213 

214 session.commit() 

215 

216 notify( 

217 session, 

218 user_id=user.id, 

219 topic_action=NotificationTopicAction.password__change, 

220 key="", 

221 ) 

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

223 

224 return empty_pb2.Empty() 

225 

226 def ChangeEmailV2( 

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

228 ) -> empty_pb2.Empty: 

229 """ 

230 Change the user's email address. 

231 

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

233 

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

235 

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

237 """ 

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

239 

240 # check password first 

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

242 # wrong password 

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

244 

245 # not a valid email 

246 if not is_valid_email(request.new_email): 

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

248 

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

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

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

252 

253 user.new_email = request.new_email 

254 user.new_email_token = urlsafe_secure_token() 

255 user.new_email_token_created = now() 

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

257 

258 send_email_changed_confirmation_to_new_email(context, session, user) 

259 

260 # will still go into old email 

261 notify( 

262 session, 

263 user_id=user.id, 

264 topic_action=NotificationTopicAction.email_address__change, 

265 key="", 

266 data=notification_data_pb2.EmailAddressChange( 

267 new_email=request.new_email, 

268 ), 

269 ) 

270 

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

272 

273 # session autocommit 

274 return empty_pb2.Empty() 

275 

276 def ChangeLanguagePreference( 

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

278 ) -> empty_pb2.Empty: 

279 # select the user from the db 

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

281 

282 # update the user's preference 

283 user.ui_language_preference = request.ui_language_preference 

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

285 

286 return empty_pb2.Empty() 

287 

288 def FillContributorForm( 

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

290 ) -> empty_pb2.Empty: 

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

292 

293 form = request.contributor_form 

294 

295 form = ContributorForm( 

296 user_id=user.id, 

297 ideas=form.ideas or None, 

298 features=form.features or None, 

299 experience=form.experience or None, 

300 contribute=contributeoption2sql[form.contribute], 

301 contribute_ways=form.contribute_ways, 

302 expertise=form.expertise or None, 

303 ) 

304 

305 session.add(form) 

306 session.flush() 

307 maybe_send_contributor_form_email(session, form) 

308 

309 user.filled_contributor_form = True 

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

311 

312 return empty_pb2.Empty() 

313 

314 def GetContributorFormInfo( 

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

316 ) -> account_pb2.GetContributorFormInfoRes: 

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

318 

319 return account_pb2.GetContributorFormInfoRes( 

320 filled_contributor_form=user.filled_contributor_form, 

321 ) 

322 

323 def ChangePhone( 

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

325 ) -> empty_pb2.Empty: 

326 phone = request.phone 

327 # early quick validation 

328 if phone and not is_e164_format(phone): 

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

330 

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

332 if user.last_donated is None: 

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

334 

335 if not phone: 

336 user.phone = None 

337 user.phone_verification_verified = None 

338 user.phone_verification_token = None 

339 user.phone_verification_attempts = 0 

340 return empty_pb2.Empty() 

341 

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

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

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

345 

346 if not is_known_operator(phone): 

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

348 

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

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

351 

352 token = sms.generate_random_code() 

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

354 

355 if result == "success": 

356 user.phone = phone 

357 user.phone_verification_verified = None 

358 user.phone_verification_token = token 

359 user.phone_verification_sent = now() 

360 user.phone_verification_attempts = 0 

361 

362 notify( 

363 session, 

364 user_id=user.id, 

365 topic_action=NotificationTopicAction.phone_number__change, 

366 key="", 

367 data=notification_data_pb2.PhoneNumberChange( 

368 phone=phone, 

369 ), 

370 ) 

371 

372 return empty_pb2.Empty() 

373 

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

375 

376 def VerifyPhone( 

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

378 ) -> empty_pb2.Empty: 

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

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

381 

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

383 if user.phone_verification_token is None: 

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

385 

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

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

388 

389 if user.phone_verification_attempts > SMS_CODE_ATTEMPTS: 

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

391 

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

393 user.phone_verification_attempts += 1 

394 session.commit() 

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

396 

397 # Delete verifications from everyone else that has this number 

398 session.execute( 

399 update(User) 

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

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

402 .values( 

403 { 

404 "phone_verification_verified": None, 

405 "phone_verification_attempts": 0, 

406 "phone_verification_token": None, 

407 "phone": None, 

408 } 

409 ) 

410 .execution_options(synchronize_session=False) 

411 ) 

412 

413 user.phone_verification_token = None 

414 user.phone_verification_verified = now() 

415 user.phone_verification_attempts = 0 

416 

417 notify( 

418 session, 

419 user_id=user.id, 

420 topic_action=NotificationTopicAction.phone_number__verify, 

421 key="", 

422 data=notification_data_pb2.PhoneNumberVerify( 

423 phone=user.phone, 

424 ), 

425 ) 

426 

427 return empty_pb2.Empty() 

428 

429 def InitiateStrongVerification( 

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

431 ) -> account_pb2.InitiateStrongVerificationRes: 

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

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

434 

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

436 existing_verification = session.execute( 

437 select(StrongVerificationAttempt) 

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

439 .where(StrongVerificationAttempt.is_valid) 

440 ).scalar_one_or_none() 

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

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

443 

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

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

446 

447 verification_attempt_token = urlsafe_secure_token() 

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

449 reference = b64encode( 

450 simple_encrypt( 

451 "iris_callback", 

452 internal_pb2.VerificationReferencePayload( 

453 verification_attempt_token=verification_attempt_token, 

454 user_id=user.id, 

455 ).SerializeToString(), 

456 ) 

457 ) 

458 response = requests.post( 

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

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

461 json={ 

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

463 "face_verification": False, 

464 "passport_only": True, 

465 "reference": reference, 

466 }, 

467 timeout=10, 

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

469 ) 

470 

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

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

473 

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

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

476 session.add( 

477 StrongVerificationAttempt( 

478 user_id=user.id, 

479 verification_attempt_token=verification_attempt_token, 

480 iris_session_id=iris_session_id, 

481 iris_token=token, 

482 ) 

483 ) 

484 

485 redirect_params = { 

486 "token": token, 

487 "redirect_url": urls.complete_strong_verification_url( 

488 verification_attempt_token=verification_attempt_token 

489 ), 

490 } 

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

492 

493 return account_pb2.InitiateStrongVerificationRes( 

494 verification_attempt_token=verification_attempt_token, 

495 redirect_url=redirect_url, 

496 ) 

497 

498 def GetStrongVerificationAttemptStatus( 

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

500 ) -> account_pb2.GetStrongVerificationAttemptStatusRes: 

501 verification_attempt = session.execute( 

502 select(StrongVerificationAttempt) 

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

504 .where(StrongVerificationAttempt.is_visible) 

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

506 ).scalar_one_or_none() 

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

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

509 status_to_pb = { 

510 StrongVerificationAttemptStatus.succeeded: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_SUCCEEDED, 

511 StrongVerificationAttemptStatus.in_progress_waiting_on_user_to_open_app: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_TO_OPEN_APP, 

512 StrongVerificationAttemptStatus.in_progress_waiting_on_user_in_app: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_USER_IN_APP, 

513 StrongVerificationAttemptStatus.in_progress_waiting_on_backend: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_IN_PROGRESS_WAITING_ON_BACKEND, 

514 StrongVerificationAttemptStatus.failed: account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_FAILED, 

515 } 

516 return account_pb2.GetStrongVerificationAttemptStatusRes( 

517 status=status_to_pb.get( 

518 verification_attempt.status, account_pb2.STRONG_VERIFICATION_ATTEMPT_STATUS_UNKNOWN 

519 ), 

520 ) 

521 

522 def DeleteStrongVerificationData( 

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

524 ) -> empty_pb2.Empty: 

525 verification_attempts = ( 

526 session.execute( 

527 select(StrongVerificationAttempt) 

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

529 .where(StrongVerificationAttempt.has_full_data) 

530 ) 

531 .scalars() 

532 .all() 

533 ) 

534 for verification_attempt in verification_attempts: 

535 verification_attempt.status = StrongVerificationAttemptStatus.deleted 

536 verification_attempt.has_full_data = False 

537 verification_attempt.passport_encrypted_data = None 

538 verification_attempt.passport_date_of_birth = None 

539 verification_attempt.passport_sex = None 

540 session.flush() 

541 # double check: 

542 verification_attempts = ( 

543 session.execute( 

544 select(StrongVerificationAttempt) 

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

546 .where(StrongVerificationAttempt.has_full_data) 

547 ) 

548 .scalars() 

549 .all() 

550 ) 

551 assert len(verification_attempts) == 0 

552 

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

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

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

556 

557 return empty_pb2.Empty() 

558 

559 def DeleteAccount( 

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

561 ) -> empty_pb2.Empty: 

562 """ 

563 Triggers email with token to confirm deletion 

564 

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

566 """ 

567 if not request.confirm: 

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

569 

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

571 

572 reason = request.reason.strip() 

573 if reason: 

574 session.add(AccountDeletionReason(user_id=user.id, reason=reason)) 

575 

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

577 

578 notify( 

579 session, 

580 user_id=user.id, 

581 topic_action=NotificationTopicAction.account_deletion__start, 

582 key="", 

583 data=notification_data_pb2.AccountDeletionStart( 

584 deletion_token=token.token, 

585 ), 

586 ) 

587 session.add(token) 

588 

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

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

591 

592 return empty_pb2.Empty() 

593 

594 def ListModNotes( 

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

596 ) -> account_pb2.ListModNotesRes: 

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

598 

599 notes = ( 

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

601 .scalars() 

602 .all() 

603 ) 

604 

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

606 

607 def ListActiveSessions( 

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

609 ) -> account_pb2.ListActiveSessionsRes: 

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

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

612 

613 user_sessions = ( 

614 session.execute( 

615 select(UserSession) 

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

617 .where(UserSession.is_valid) 

618 .where(UserSession.is_api_key == False) 

619 .where(UserSession.last_seen <= page_token) 

620 .order_by(UserSession.last_seen.desc()) 

621 .limit(page_size + 1) 

622 ) 

623 .scalars() 

624 .all() 

625 ) 

626 

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

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

629 return account_pb2.ActiveSession( 

630 created=Timestamp_from_datetime(user_session.created), 

631 expiry=Timestamp_from_datetime(user_session.expiry), 

632 last_seen=Timestamp_from_datetime(user_session.last_seen), 

633 operating_system=user_agent.os.family, 

634 browser=user_agent.browser.family, 

635 device=user_agent.device.family, 

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

637 is_current_session=user_session.token == context.token, 

638 ) 

639 

640 return account_pb2.ListActiveSessionsRes( 

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

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

643 ) 

644 

645 def LogOutSession( 

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

647 ) -> empty_pb2.Empty: 

648 session.execute( 

649 update(UserSession) 

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

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

652 .where(UserSession.is_valid) 

653 .where(UserSession.is_api_key == False) 

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

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

656 .execution_options(synchronize_session=False) 

657 ) 

658 return empty_pb2.Empty() 

659 

660 def LogOutOtherSessions( 

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

662 ) -> empty_pb2.Empty: 

663 if not request.confirm: 

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

665 

666 session.execute( 

667 update(UserSession) 

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

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

670 .where(UserSession.is_valid) 

671 .where(UserSession.is_api_key == False) 

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

673 .execution_options(synchronize_session=False) 

674 ) 

675 return empty_pb2.Empty() 

676 

677 def SetProfilePublicVisibility( 

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

679 ) -> empty_pb2.Empty: 

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

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

682 user.has_modified_public_visibility = True 

683 return empty_pb2.Empty() 

684 

685 def CreateInviteCode( 

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

687 ) -> account_pb2.CreateInviteCodeRes: 

688 code = generate_invite_code() 

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

690 

691 return account_pb2.CreateInviteCodeRes( 

692 code=code, 

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

694 ) 

695 

696 def DisableInviteCode( 

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

698 ) -> empty_pb2.Empty: 

699 invite = session.execute( 

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

701 ).scalar_one_or_none() 

702 

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

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

705 

706 invite.disabled = func.now() 

707 session.commit() 

708 

709 return empty_pb2.Empty() 

710 

711 def ListInviteCodes( 

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

713 ) -> account_pb2.ListInviteCodesRes: 

714 results = session.execute( 

715 select( 

716 InviteCode.id, 

717 InviteCode.created, 

718 InviteCode.disabled, 

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

720 ) 

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

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

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

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

725 ).all() 

726 

727 return account_pb2.ListInviteCodesRes( 

728 invite_codes=[ 

729 account_pb2.InviteCodeInfo( 

730 code=code_id, 

731 created=Timestamp_from_datetime(created), 

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

733 uses=len_users, 

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

735 ) 

736 for code_id, created, disabled, len_users in results 

737 ] 

738 ) 

739 

740 def GetReminders( 

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

742 ) -> account_pb2.GetRemindersRes: 

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

744