Coverage for app/backend/src/couchers/servicers/admin.py: 84%
616 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1import json
2import logging
3from collections.abc import Callable
4from datetime import UTC, datetime, timedelta
5from typing import Any
7import grpc
8from google.protobuf import empty_pb2
9from google.protobuf.json_format import MessageToJson
10from google.protobuf.wrappers_pb2 import Int64Value
11from sqlalchemy import select, tuple_
12from sqlalchemy.orm import Session, aliased, selectinload, undefer
13from sqlalchemy.sql import and_, func, or_
14from user_agents import parse as user_agents_parse
16from couchers import urls
17from couchers.constants import MODERATION_AUTO_APPROVE_DEADLINE_MAX_SECONDS
18from couchers.context import CouchersContext
19from couchers.crypto import urlsafe_secure_token
20from couchers.data_export import generate_user_data_export
21from couchers.helpers.badges import user_add_badge, user_remove_badge
22from couchers.helpers.geoip import geoip_approximate_location, geoip_asn
23from couchers.helpers.strong_verification import get_strong_verification_fields
24from couchers.helpers.upload_uses import UploadUseType, get_upload_uses_for_keys
25from couchers.jobs.enqueue import queue_job
26from couchers.models import (
27 AccountDeletionToken,
28 AdminAction,
29 AdminActionLevel,
30 AdminTag,
31 Comment,
32 ContentReport,
33 Discussion,
34 Event,
35 EventOccurrence,
36 FriendRelationship,
37 GroupChat,
38 GroupChatSubscription,
39 HostRequest,
40 LanguageAbility,
41 Message,
42 ModerationUserList,
43 ModerationVisibility,
44 ModNote,
45 NonvisibleUserAccess,
46 NonvisibleUserAccessType,
47 NonvisibleUserState,
48 OTAPackage,
49 OTAPlatform,
50 Reference,
51 Reply,
52 User,
53 UserActivity,
54 UserAdminTag,
55 UserBadge,
56 UserBlock,
57)
58from couchers.models.discussions import (
59 CommentVersion,
60 ContentChangeType,
61 DiscussionVersion,
62 ReplyVersion,
63)
64from couchers.models.notifications import NotificationTopicAction
65from couchers.models.uploads import Upload, has_avatar_photo_expression
66from couchers.notifications.notify import notify
67from couchers.proto import admin_pb2, admin_pb2_grpc, api_pb2, notification_data_pb2
68from couchers.proto.internal import jobs_pb2
69from couchers.resources import get_badge_dict
70from couchers.servicers.account import abort_if_email_reused
71from couchers.servicers.api import user_model_to_pb
72from couchers.servicers.auth import create_session
73from couchers.servicers.bugs import _fetch_signed_manifest, _native_ota_manifest_url
74from couchers.servicers.events import generate_event_delete_notifications
75from couchers.servicers.moderation import bulk_set_user_content_visibility
76from couchers.servicers.threads import unpack_thread_id
77from couchers.sql import to_bool, username_or_email_or_id
78from couchers.utils import Timestamp_from_datetime, date_to_api, not_none, now, parse_date, to_aware_datetime
80logger = logging.getLogger(__name__)
82MAX_PAGINATION_LENGTH = 250
85adminactionlevel2api = {
86 AdminActionLevel.trace: admin_pb2.ADMIN_ACTION_LEVEL_TRACE,
87 AdminActionLevel.debug: admin_pb2.ADMIN_ACTION_LEVEL_DEBUG,
88 AdminActionLevel.normal: admin_pb2.ADMIN_ACTION_LEVEL_NORMAL,
89 AdminActionLevel.high: admin_pb2.ADMIN_ACTION_LEVEL_HIGH,
90}
92api2adminactionlevel = {
93 admin_pb2.ADMIN_ACTION_LEVEL_TRACE: AdminActionLevel.trace,
94 admin_pb2.ADMIN_ACTION_LEVEL_DEBUG: AdminActionLevel.debug,
95 admin_pb2.ADMIN_ACTION_LEVEL_NORMAL: AdminActionLevel.normal,
96 admin_pb2.ADMIN_ACTION_LEVEL_HIGH: AdminActionLevel.high,
97}
99uploadusetype2api = {
100 None: admin_pb2.UPLOAD_USE_TYPE_UNSPECIFIED,
101 UploadUseType.profile_gallery_photo: admin_pb2.UPLOAD_USE_TYPE_PROFILE_GALLERY_PHOTO,
102 UploadUseType.profile_gallery_photo_avatar: admin_pb2.UPLOAD_USE_TYPE_PROFILE_GALLERY_PHOTO_AVATAR,
103 UploadUseType.event: admin_pb2.UPLOAD_USE_TYPE_EVENT,
104 UploadUseType.page: admin_pb2.UPLOAD_USE_TYPE_PAGE,
105}
107otaplatform2api = {
108 None: admin_pb2.OTA_PLATFORM_UNSPECIFIED,
109 OTAPlatform.ios: admin_pb2.OTA_PLATFORM_IOS,
110 OTAPlatform.android: admin_pb2.OTA_PLATFORM_ANDROID,
111}
113api2otaplatform = {
114 admin_pb2.OTA_PLATFORM_UNSPECIFIED: None,
115 admin_pb2.OTA_PLATFORM_IOS: OTAPlatform.ios,
116 admin_pb2.OTA_PLATFORM_ANDROID: OTAPlatform.android,
117}
119nonvisibleuseraccesstype2api = {
120 None: admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_UNSPECIFIED,
121 NonvisibleUserAccessType.login_attempt: admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_LOGIN_ATTEMPT,
122 NonvisibleUserAccessType.profile_view: admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_PROFILE_VIEW,
123 NonvisibleUserAccessType.ghost_served: admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_GHOST_SERVED,
124}
126nonvisibleuserstate2api = {
127 None: admin_pb2.NONVISIBLE_USER_STATE_UNSPECIFIED,
128 NonvisibleUserState.banned: admin_pb2.NONVISIBLE_USER_STATE_BANNED,
129 NonvisibleUserState.shadowed: admin_pb2.NONVISIBLE_USER_STATE_SHADOWED,
130 NonvisibleUserState.deleted: admin_pb2.NONVISIBLE_USER_STATE_DELETED,
131}
133# Doubles as the set of accepted values: MOD_NOTE_NOTIFICATION_UNSPECIFIED is rejected.
134_MOD_NOTE_NOTIFY_LOG_LABELS = {
135 admin_pb2.MOD_NOTE_NOTIFICATION_NONE: "No",
136 admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY: "Yes",
137 admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT: "Yes, including the note text",
138}
141def log_admin_action(
142 session: Session,
143 context: CouchersContext,
144 target_user: User,
145 action_type: str,
146 note: str | None = None,
147 data: object | None = None,
148 tag: str | None = None,
149 level: AdminActionLevel = AdminActionLevel.normal,
150) -> AdminAction:
151 action = AdminAction(
152 admin_user_id=context.user_id,
153 target_user_id=target_user.id,
154 action_type=action_type,
155 level=level,
156 note=note,
157 data=data,
158 tag=tag,
159 )
160 session.add(action)
161 session.flush()
162 return action
165def _live_ota_package_ids(session: Session) -> set[int]:
166 # The live package per (platform, fingerprint) is the newest non-banned one by manifest_created_at,
167 # matching what GetNativeUpdateManifest resolves. DISTINCT ON picks the row with the leading ORDER BY
168 # value per (platform, fingerprint) group in a single index-friendly query.
169 return set(
170 session.scalars(
171 select(OTAPackage.id)
172 .where(OTAPackage.banned_at.is_(None))
173 .distinct(OTAPackage.platform, OTAPackage.fingerprint)
174 .order_by(
175 OTAPackage.platform,
176 OTAPackage.fingerprint,
177 OTAPackage.manifest_created_at.desc(),
178 OTAPackage.id.desc(),
179 )
180 )
181 )
184def _extract_ota_manifest(body: bytes) -> dict[str, Any] | None:
185 # The manifest object is the JSON in the "manifest" part of the signed multipart/mixed body.
186 marker = body.find(b'name="manifest"')
187 if marker == -1:
188 return None
189 body_start = body.find(b"\r\n\r\n", marker)
190 if body_start == -1: 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true
191 return None
192 body_end = body.find(b"\r\n--", body_start + 4)
193 if body_end == -1: 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true
194 return None
195 try:
196 manifest = json.loads(body[body_start + 4 : body_end])
197 except json.JSONDecodeError:
198 return None
199 return manifest if isinstance(manifest, dict) else None
202def _ota_package_to_pb(package: OTAPackage, live_ids: set[int]) -> admin_pb2.OTAPackage:
203 return admin_pb2.OTAPackage(
204 ota_package_id=package.id,
205 created=Timestamp_from_datetime(package.created),
206 creator_user_id=package.creator_user_id,
207 platform=otaplatform2api[package.platform],
208 fingerprint=package.fingerprint,
209 version=package.version,
210 manifest_created_at=Timestamp_from_datetime(package.manifest_created_at),
211 manifest_id=package.manifest_id,
212 banned=package.banned_at is not None,
213 banned_at=Timestamp_from_datetime(package.banned_at) if package.banned_at else None,
214 banned_by_user_id=package.banned_by_user_id or 0,
215 banned_reason=package.banned_reason or "",
216 live=package.id in live_ids,
217 )
220def find_user_by_username_or_email_or_id(session: Session, context: CouchersContext, user: str) -> User:
221 users = session.execute(select(User).where(username_or_email_or_id(user))).scalars().all()
222 if not users:
223 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
224 if len(users) > 1:
225 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:email_used_by_multiple_users")
226 return users[0]
229def _user_to_details(session: Session, user: User) -> admin_pb2.UserDetails:
230 # Query admin actions for this user
231 actions = session.execute(
232 select(AdminAction, User.username)
233 .join(User, AdminAction.admin_user_id == User.id)
234 .where(AdminAction.target_user_id == user.id)
235 .order_by(AdminAction.created.asc())
236 ).all()
238 action_pbs = []
239 for action, admin_username in actions:
240 action_pbs.append(
241 admin_pb2.AdminActionLog(
242 admin_action_id=action.id,
243 created=Timestamp_from_datetime(action.created),
244 admin_user_id=action.admin_user_id,
245 admin_username=admin_username,
246 action_type=action.action_type,
247 level=adminactionlevel2api[action.level],
248 note=action.note or "",
249 data=json.dumps(action.data) if action.data is not None else "",
250 tag=action.tag or "",
251 target_user_id=action.target_user_id,
252 target_username=user.username,
253 )
254 )
256 # Query admin tags
257 admin_tags = (
258 session.execute(
259 select(AdminTag.tag)
260 .join(UserAdminTag, UserAdminTag.admin_tag_id == AdminTag.id)
261 .where(UserAdminTag.user_id == user.id)
262 .order_by(AdminTag.tag)
263 )
264 .scalars()
265 .all()
266 )
268 last_mod_note_acknowledged = session.execute(
269 select(func.max(ModNote.acknowledged)).where(ModNote.user_id == user.id)
270 ).scalar()
272 return admin_pb2.UserDetails(
273 user_id=user.id,
274 username=user.username,
275 name=user.name,
276 email=user.email,
277 gender=user.gender,
278 birthdate=date_to_api(user.birthdate),
279 banned=user.banned_at is not None,
280 deleted=user.deleted_at is not None,
281 shadowed=user.shadowed_at is not None,
282 do_not_email=user.do_not_email,
283 badges=[badge.badge_id for badge in user.badges],
284 **get_strong_verification_fields(session, user),
285 has_passport_sex_gender_exception=user.has_passport_sex_gender_exception,
286 pending_mod_notes_count=user.mod_notes.where(ModNote.is_pending).count(),
287 acknowledged_mod_notes_count=user.mod_notes.where(~ModNote.is_pending).count(),
288 last_mod_note_acknowledged=(
289 Timestamp_from_datetime(last_mod_note_acknowledged) if last_mod_note_acknowledged else None
290 ),
291 admin_actions=action_pbs,
292 admin_tags=list(admin_tags),
293 mod_score=user.mod_score,
294 ui_language_preference=user.ui_language_preference,
295 moderation_auto_approve_seconds=(
296 Int64Value(value=user.moderation_auto_approve_deadline_seconds)
297 if user.moderation_auto_approve_deadline_seconds is not None
298 else None
299 ),
300 )
303def _format_auto_approve_deadline(deadline_seconds: int | None) -> str:
304 if deadline_seconds is None:
305 return "default"
306 if deadline_seconds == 0:
307 return "never"
308 return f"{deadline_seconds} seconds"
311def _content_report_to_pb(content_report: ContentReport) -> admin_pb2.ContentReport:
312 return admin_pb2.ContentReport(
313 content_report_id=content_report.id,
314 time=Timestamp_from_datetime(content_report.time),
315 reporting_user_id=content_report.reporting_user_id,
316 author_user_id=content_report.author_user_id,
317 reason=content_report.reason,
318 description=content_report.description,
319 content_ref=content_report.content_ref,
320 user_agent=content_report.user_agent,
321 page=content_report.page,
322 )
325def _upload_metadata_to_pb(upload: Upload) -> admin_pb2.UploadMetadata:
326 return admin_pb2.UploadMetadata(
327 parsed_json=json.dumps(upload.metadata_parsed, sort_keys=True) if upload.metadata_parsed else None,
328 parse_error=upload.metadata_parse_error,
329 original_filename=upload.original_filename,
330 original_format=upload.original_format,
331 original_size=upload.original_size,
332 original_width=upload.original_width,
333 original_height=upload.original_height,
334 )
337def _make_chat_user_info_getter(session: Session) -> Callable[[int], admin_pb2.ChatUserInfo]:
338 user_info_cache: dict[int, admin_pb2.ChatUserInfo] = {}
340 def get_chat_user_info(user_id: int) -> admin_pb2.ChatUserInfo:
341 if user_id not in user_info_cache:
342 u = session.execute(select(User).where(User.id == user_id)).scalar_one()
343 user_info_cache[user_id] = admin_pb2.ChatUserInfo(
344 user_id=u.id,
345 username=u.username,
346 name=u.name,
347 birthdate=date_to_api(u.birthdate),
348 gender=u.gender,
349 )
350 return user_info_cache[user_id]
352 return get_chat_user_info
355def _reference_to_pb(
356 reference: Reference, get_chat_user_info: Callable[[int], admin_pb2.ChatUserInfo]
357) -> admin_pb2.AdminReference:
358 host_request = reference.host_request
359 return admin_pb2.AdminReference(
360 reference_id=reference.id,
361 from_user_id=reference.from_user_id,
362 to_user_id=reference.to_user_id,
363 from_user=get_chat_user_info(reference.from_user_id),
364 to_user=get_chat_user_info(reference.to_user_id),
365 reference_type=reference.reference_type.name,
366 text=reference.text,
367 private_text=reference.private_text or "",
368 time=Timestamp_from_datetime(reference.time),
369 host_request_id=reference.host_request_id or 0,
370 rating=reference.rating,
371 was_appropriate=reference.was_appropriate,
372 moderation_visibility=not_none(reference.moderation_state.visibility).name,
373 hosting_city=host_request.hosting_city if host_request else "",
374 from_date=date_to_api(host_request.from_date) if host_request else "",
375 to_date=date_to_api(host_request.to_date) if host_request else "",
376 status=host_request.status.name if host_request else "",
377 )
380class Admin(admin_pb2_grpc.AdminServicer):
381 def GetUserDetails(
382 self, request: admin_pb2.GetUserDetailsReq, context: CouchersContext, session: Session
383 ) -> admin_pb2.UserDetails:
384 user = find_user_by_username_or_email_or_id(session, context, request.user)
385 return _user_to_details(session, user)
387 def GetUser(self, request: admin_pb2.GetUserReq, context: CouchersContext, session: Session) -> api_pb2.User:
388 user = find_user_by_username_or_email_or_id(session, context, request.user)
389 return user_model_to_pb(user, session, context, is_admin_see_ghosts=True)
391 def SearchUsers(
392 self, request: admin_pb2.SearchUsersReq, context: CouchersContext, session: Session
393 ) -> admin_pb2.SearchUsersRes:
394 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
395 next_user_id = int(request.page_token) if request.page_token else 0
396 statement = select(User)
397 if request.username: 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true
398 statement = statement.where(User.username.ilike(request.username))
399 if request.email:
400 statement = statement.where(User.email.ilike(request.email))
401 if request.exact_email:
402 statement = statement.where(User.email == request.exact_email.lower())
403 if request.name: 403 ↛ 404line 403 didn't jump to line 404 because the condition on line 403 was never true
404 statement = statement.where(User.name.ilike(request.name))
405 if request.admin_action_log:
406 statement = statement.where(
407 User.id.in_(select(AdminAction.target_user_id).where(AdminAction.note.ilike(request.admin_action_log)))
408 )
409 if request.city: 409 ↛ 410line 409 didn't jump to line 410 because the condition on line 409 was never true
410 statement = statement.where(User.city.ilike(request.city))
411 if request.min_user_id: 411 ↛ 412line 411 didn't jump to line 412 because the condition on line 411 was never true
412 statement = statement.where(User.id >= request.min_user_id)
413 if request.max_user_id: 413 ↛ 414line 413 didn't jump to line 414 because the condition on line 413 was never true
414 statement = statement.where(User.id <= request.max_user_id)
415 if request.min_birthdate: 415 ↛ 416line 415 didn't jump to line 416 because the condition on line 415 was never true
416 statement = statement.where(User.birthdate >= parse_date(request.min_birthdate))
417 if request.max_birthdate: 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 statement = statement.where(User.birthdate <= parse_date(request.max_birthdate))
419 if request.genders: 419 ↛ 420line 419 didn't jump to line 420 because the condition on line 419 was never true
420 statement = statement.where(User.gender.in_(request.genders))
421 if request.min_joined_date: 421 ↛ 422line 421 didn't jump to line 422 because the condition on line 421 was never true
422 statement = statement.where(User.joined >= parse_date(request.min_joined_date))
423 if request.max_joined_date: 423 ↛ 424line 423 didn't jump to line 424 because the condition on line 423 was never true
424 statement = statement.where(User.joined <= parse_date(request.max_joined_date))
425 if request.min_last_active_date: 425 ↛ 426line 425 didn't jump to line 426 because the condition on line 425 was never true
426 statement = statement.where(User.last_active >= parse_date(request.min_last_active_date))
427 if request.max_last_active_date: 427 ↛ 428line 427 didn't jump to line 428 because the condition on line 427 was never true
428 statement = statement.where(User.last_active <= parse_date(request.max_last_active_date))
429 if request.genders: 429 ↛ 430line 429 didn't jump to line 430 because the condition on line 429 was never true
430 statement = statement.where(User.gender.in_(request.genders))
431 if request.language_codes: 431 ↛ 432line 431 didn't jump to line 432 because the condition on line 431 was never true
432 statement = statement.join(
433 LanguageAbility,
434 and_(LanguageAbility.user_id == User.id, LanguageAbility.language_code.in_(request.language_codes)),
435 )
436 if request.HasField("is_deleted"): 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true
437 statement = statement.where((User.deleted_at != None) == request.is_deleted.value)
438 if request.HasField("is_banned"): 438 ↛ 439line 438 didn't jump to line 439 because the condition on line 438 was never true
439 statement = statement.where((User.banned_at != None) == request.is_banned.value)
440 if request.HasField("is_shadowed"): 440 ↛ 441line 440 didn't jump to line 441 because the condition on line 440 was never true
441 statement = statement.where((User.shadowed_at != None) == request.is_shadowed.value)
442 if request.HasField("has_avatar"): 442 ↛ 443line 442 didn't jump to line 443 because the condition on line 442 was never true
443 statement = statement.where(has_avatar_photo_expression(User) == request.has_avatar.value)
444 if request.admin_tags:
445 for tag_name in request.admin_tags:
446 statement = statement.where(
447 User.id.in_(
448 select(UserAdminTag.user_id)
449 .join(AdminTag, UserAdminTag.admin_tag_id == AdminTag.id)
450 .where(AdminTag.tag == tag_name)
451 )
452 )
453 users = (
454 session.execute(
455 statement.where(User.id >= next_user_id)
456 .order_by(User.id)
457 .limit(page_size + 1)
458 .options(selectinload(User.badges))
459 )
460 .scalars()
461 .all()
462 )
463 logger.info(users)
464 return admin_pb2.SearchUsersRes(
465 users=[_user_to_details(session, user) for user in users[:page_size]],
466 next_page_token=str(users[-1].id) if len(users) > page_size else None,
467 )
469 def ChangeUserGender(
470 self, request: admin_pb2.ChangeUserGenderReq, context: CouchersContext, session: Session
471 ) -> admin_pb2.UserDetails:
472 user = find_user_by_username_or_email_or_id(session, context, request.user)
473 old_gender = user.gender
474 user.gender = request.gender
475 log_admin_action(
476 session, context, user, "change_gender", note=f"Changed from '{old_gender}' to '{request.gender}'"
477 )
478 session.commit()
480 notify(
481 session,
482 user_id=user.id,
483 topic_action=NotificationTopicAction.gender__change,
484 key="",
485 data=notification_data_pb2.GenderChange(
486 gender=request.gender,
487 ),
488 )
490 return _user_to_details(session, user)
492 def ChangeUserBirthdate(
493 self, request: admin_pb2.ChangeUserBirthdateReq, context: CouchersContext, session: Session
494 ) -> admin_pb2.UserDetails:
495 user = find_user_by_username_or_email_or_id(session, context, request.user)
496 if not (birthdate := parse_date(request.birthdate)): 496 ↛ 497line 496 didn't jump to line 497 because the condition on line 496 was never true
497 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_birthdate")
499 old_birthdate = user.birthdate
500 user.birthdate = birthdate
501 log_admin_action(
502 session, context, user, "change_birthdate", note=f"Changed from {old_birthdate} to {request.birthdate}"
503 )
504 session.commit()
506 notify(
507 session,
508 user_id=user.id,
509 topic_action=NotificationTopicAction.birthdate__change,
510 key="",
511 data=notification_data_pb2.BirthdateChange(
512 birthdate=request.birthdate,
513 ),
514 )
516 return _user_to_details(session, user)
518 def AddBadge(
519 self, request: admin_pb2.AddBadgeReq, context: CouchersContext, session: Session
520 ) -> admin_pb2.UserDetails:
521 user = find_user_by_username_or_email_or_id(session, context, request.user)
523 badge = get_badge_dict().get(request.badge_id)
524 if not badge:
525 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "badge_not_found")
527 if not badge.admin_editable:
528 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:cannot_edit_badge")
530 if badge.id in [b.badge_id for b in user.badges]:
531 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_already_has_badge")
533 user_add_badge(session, user.id, request.badge_id)
534 log_admin_action(session, context, user, "add_badge", note=f"Added badge {request.badge_id}")
536 return _user_to_details(session, user)
538 def RemoveBadge(
539 self, request: admin_pb2.RemoveBadgeReq, context: CouchersContext, session: Session
540 ) -> admin_pb2.UserDetails:
541 user = find_user_by_username_or_email_or_id(session, context, request.user)
543 badge = get_badge_dict().get(request.badge_id)
544 if not badge: 544 ↛ 545line 544 didn't jump to line 545 because the condition on line 544 was never true
545 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "badge_not_found")
547 if not badge.admin_editable: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true
548 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:cannot_edit_badge")
550 user_badge = session.execute(
551 select(UserBadge).where(UserBadge.user_id == user.id, UserBadge.badge_id == badge.id)
552 ).scalar_one_or_none()
553 if not user_badge:
554 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_does_not_have_badge")
556 user_remove_badge(session, user.id, request.badge_id)
557 log_admin_action(session, context, user, "remove_badge", note=f"Removed badge {request.badge_id}")
559 return _user_to_details(session, user)
561 def SetPassportSexGenderException(
562 self, request: admin_pb2.SetPassportSexGenderExceptionReq, context: CouchersContext, session: Session
563 ) -> admin_pb2.UserDetails:
564 user = find_user_by_username_or_email_or_id(session, context, request.user)
565 old_exception = user.has_passport_sex_gender_exception
566 user.has_passport_sex_gender_exception = request.passport_sex_gender_exception
567 log_admin_action(
568 session,
569 context,
570 user,
571 "set_passport_sex_gender_exception",
572 note=f"Changed from {old_exception} to {request.passport_sex_gender_exception}",
573 )
574 return _user_to_details(session, user)
576 def BanUser(
577 self, request: admin_pb2.BanUserReq, context: CouchersContext, session: Session
578 ) -> admin_pb2.UserDetails:
579 user = find_user_by_username_or_email_or_id(session, context, request.user)
580 if not request.admin_note.strip(): 580 ↛ 581line 580 didn't jump to line 581 because the condition on line 580 was never true
581 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
582 log_admin_action(session, context, user, "ban", note=request.admin_note, level=AdminActionLevel.high)
583 user.banned_at = now()
584 return _user_to_details(session, user)
586 def UnbanUser(
587 self, request: admin_pb2.UnbanUserReq, context: CouchersContext, session: Session
588 ) -> admin_pb2.UserDetails:
589 user = find_user_by_username_or_email_or_id(session, context, request.user)
590 if not request.admin_note.strip(): 590 ↛ 591line 590 didn't jump to line 591 because the condition on line 590 was never true
591 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
592 log_admin_action(session, context, user, "unban", note=request.admin_note, level=AdminActionLevel.high)
593 user.banned_at = None
594 return _user_to_details(session, user)
596 def ShadowUser(
597 self, request: admin_pb2.ShadowUserReq, context: CouchersContext, session: Session
598 ) -> admin_pb2.UserDetails:
599 user = find_user_by_username_or_email_or_id(session, context, request.user)
600 if not request.admin_note.strip():
601 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
602 log_admin_action(session, context, user, "shadow", note=request.admin_note, level=AdminActionLevel.high)
603 user.shadowed_at = now()
604 # Bulk-shadow all UMS-governed content authored by this user so existing visible content is hidden too
605 bulk_set_user_content_visibility(
606 session=session,
607 user=user,
608 new_visibility=ModerationVisibility.shadowed,
609 moderator_user_id=context.user_id,
610 reason=f"User {user.id} shadowed: {request.admin_note}",
611 )
612 return _user_to_details(session, user)
614 def UnshadowUser(
615 self, request: admin_pb2.UnshadowUserReq, context: CouchersContext, session: Session
616 ) -> admin_pb2.UserDetails:
617 user = find_user_by_username_or_email_or_id(session, context, request.user)
618 if not request.admin_note.strip(): 618 ↛ 619line 618 didn't jump to line 619 because the condition on line 618 was never true
619 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
620 log_admin_action(session, context, user, "unshadow", note=request.admin_note, level=AdminActionLevel.high)
621 user.shadowed_at = None
622 # Sweep content shadowed by the cascade back to visible; leave hidden/unlisted content where moderators put it
623 bulk_set_user_content_visibility(
624 session=session,
625 user=user,
626 new_visibility=ModerationVisibility.visible,
627 moderator_user_id=context.user_id,
628 from_visibilities={ModerationVisibility.shadowed},
629 reason=f"User {user.id} unshadowed: {request.admin_note}",
630 )
631 return _user_to_details(session, user)
633 def SetModerationAutoApproveDeadline(
634 self, request: admin_pb2.SetModerationAutoApproveDeadlineReq, context: CouchersContext, session: Session
635 ) -> admin_pb2.UserDetails:
636 user = find_user_by_username_or_email_or_id(session, context, request.user)
637 deadline_seconds = None
638 if request.HasField("auto_approve_seconds"):
639 deadline_seconds = request.auto_approve_seconds.value
640 if not 0 <= deadline_seconds <= MODERATION_AUTO_APPROVE_DEADLINE_MAX_SECONDS:
641 context.abort_with_error_code(
642 grpc.StatusCode.INVALID_ARGUMENT,
643 "admin:auto_approve_deadline_out_of_range",
644 substitutions={"max_hours": MODERATION_AUTO_APPROVE_DEADLINE_MAX_SECONDS // 3600},
645 )
646 old_deadline_seconds = user.moderation_auto_approve_deadline_seconds
647 user.moderation_auto_approve_deadline_seconds = deadline_seconds
648 log_admin_action(
649 session,
650 context,
651 user,
652 "set_moderation_auto_approve_deadline",
653 note=(
654 f"Changed from {_format_auto_approve_deadline(old_deadline_seconds)} "
655 f"to {_format_auto_approve_deadline(deadline_seconds)}"
656 ),
657 )
658 return _user_to_details(session, user)
660 def AddAdminNote(
661 self, request: admin_pb2.AddAdminNoteReq, context: CouchersContext, session: Session
662 ) -> admin_pb2.UserDetails:
663 user = find_user_by_username_or_email_or_id(session, context, request.user)
664 has_note = bool(request.admin_note.strip())
665 has_data = bool(request.data.strip())
666 if has_note == has_data:
667 context.abort_with_error_code(
668 grpc.StatusCode.INVALID_ARGUMENT, "admin:note_requires_exactly_one_of_note_or_data"
669 )
670 data = None
671 if has_data:
672 try:
673 data = json.loads(request.data)
674 except json.JSONDecodeError:
675 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_data_must_be_valid_json")
676 level = api2adminactionlevel.get(request.level, AdminActionLevel.normal)
677 log_admin_action(
678 session,
679 context,
680 user,
681 "note",
682 note=request.admin_note if has_note else None,
683 data=data,
684 level=level,
685 )
686 return _user_to_details(session, user)
688 def GetContentReport(
689 self, request: admin_pb2.GetContentReportReq, context: CouchersContext, session: Session
690 ) -> admin_pb2.GetContentReportRes:
691 content_report = session.execute(
692 select(ContentReport).where(ContentReport.id == request.content_report_id)
693 ).scalar_one_or_none()
694 if not content_report:
695 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:content_report_not_found")
696 return admin_pb2.GetContentReportRes(
697 content_report=_content_report_to_pb(content_report),
698 )
700 def GetContentReportsForAuthor(
701 self, request: admin_pb2.GetContentReportsForAuthorReq, context: CouchersContext, session: Session
702 ) -> admin_pb2.GetContentReportsForAuthorRes:
703 user = find_user_by_username_or_email_or_id(session, context, request.user)
704 content_reports = (
705 session.execute(
706 select(ContentReport).where(ContentReport.author_user_id == user.id).order_by(ContentReport.id.desc())
707 )
708 .scalars()
709 .all()
710 )
711 return admin_pb2.GetContentReportsForAuthorRes(
712 content_reports=[_content_report_to_pb(content_report) for content_report in content_reports],
713 )
715 def SendModNote(
716 self, request: admin_pb2.SendModNoteReq, context: CouchersContext, session: Session
717 ) -> admin_pb2.UserDetails:
718 if request.notification not in _MOD_NOTE_NOTIFY_LOG_LABELS:
719 context.abort_with_error_code(
720 grpc.StatusCode.INVALID_ARGUMENT, "admin:mod_note_notification_must_be_specified"
721 )
722 user = find_user_by_username_or_email_or_id(session, context, request.user)
723 session.add(
724 ModNote(
725 user_id=user.id,
726 internal_id=request.internal_id,
727 creator_user_id=context.user_id,
728 note_content=request.content,
729 )
730 )
731 session.flush()
732 log_admin_action(
733 session,
734 context,
735 user,
736 "send_mod_note",
737 note=f"Notify user: {_MOD_NOTE_NOTIFY_LOG_LABELS[request.notification]}\n\n{request.content}",
738 )
740 if request.notification != admin_pb2.MOD_NOTE_NOTIFICATION_NONE:
741 notify(
742 session,
743 user_id=user.id,
744 topic_action=NotificationTopicAction.modnote__create,
745 key="",
746 data=notification_data_pb2.ModNoteCreate(
747 markdown_text=(
748 request.content
749 if request.notification == admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT
750 else ""
751 ),
752 ),
753 )
755 return _user_to_details(session, user)
757 def MarkUserNeedsLocationUpdate(
758 self, request: admin_pb2.MarkUserNeedsLocationUpdateReq, context: CouchersContext, session: Session
759 ) -> admin_pb2.UserDetails:
760 user = find_user_by_username_or_email_or_id(session, context, request.user)
761 user.needs_to_update_location = True
762 log_admin_action(
763 session, context, user, "mark_needs_location_update", note="Marked user as needing location update"
764 )
765 return _user_to_details(session, user)
767 def DeleteUser(
768 self, request: admin_pb2.DeleteUserReq, context: CouchersContext, session: Session
769 ) -> admin_pb2.UserDetails:
770 user = find_user_by_username_or_email_or_id(session, context, request.user)
771 user.deleted_at = now()
772 log_admin_action(session, context, user, "delete_user", level=AdminActionLevel.high)
773 return _user_to_details(session, user)
775 def RecoverDeletedUser(
776 self, request: admin_pb2.RecoverDeletedUserReq, context: CouchersContext, session: Session
777 ) -> admin_pb2.UserDetails:
778 user = find_user_by_username_or_email_or_id(session, context, request.user)
779 abort_if_email_reused(context, session, user)
780 user.deleted_at = None
781 user.undelete_token = None
782 user.undelete_until = None
783 log_admin_action(session, context, user, "recover_user", level=AdminActionLevel.high)
784 return _user_to_details(session, user)
786 def CreateApiKey(
787 self, request: admin_pb2.CreateApiKeyReq, context: CouchersContext, session: Session
788 ) -> admin_pb2.CreateApiKeyRes:
789 user = find_user_by_username_or_email_or_id(session, context, request.user)
790 token, expiry = create_session(
791 context, session, user, long_lived=True, is_api_key=True, duration=timedelta(days=365), set_cookie=False
792 )
793 log_admin_action(session, context, user, "create_api_key")
795 notify(
796 session,
797 user_id=user.id,
798 topic_action=NotificationTopicAction.api_key__create,
799 key="",
800 data=notification_data_pb2.ApiKeyCreate(
801 api_key=token,
802 expiry=Timestamp_from_datetime(expiry),
803 ),
804 )
806 return admin_pb2.CreateApiKeyRes()
808 def GetChats(
809 self, request: admin_pb2.GetChatsReq, context: CouchersContext, session: Session
810 ) -> admin_pb2.GetChatsRes:
811 user = find_user_by_username_or_email_or_id(session, context, request.user)
813 get_chat_user_info = _make_chat_user_info_getter(session)
815 def message_to_pb(message: Message) -> admin_pb2.ChatMessage:
816 return admin_pb2.ChatMessage(
817 message_id=message.id,
818 author=get_chat_user_info(message.author_id),
819 time=Timestamp_from_datetime(message.time),
820 message_type=message.message_type.name if message.message_type else "",
821 text=message.text or "",
822 host_request_status_target=(
823 message.host_request_status_target.name if message.host_request_status_target else ""
824 ),
825 target=get_chat_user_info(message.target_id) if message.target_id else None,
826 )
828 def get_messages_for_conversation(conversation_id: int) -> list[admin_pb2.ChatMessage]:
829 messages = (
830 session.execute(
831 select(Message).where(Message.conversation_id == conversation_id).order_by(Message.id.asc())
832 )
833 .scalars()
834 .all()
835 )
836 return [message_to_pb(msg) for msg in messages]
838 def get_host_request_pb(host_request: HostRequest) -> admin_pb2.AdminHostRequest:
839 return admin_pb2.AdminHostRequest(
840 host_request_id=host_request.conversation_id,
841 surfer=get_chat_user_info(host_request.surfer_user_id),
842 host=get_chat_user_info(host_request.host_user_id),
843 status=host_request.status.name if host_request.status else "",
844 from_date=date_to_api(host_request.from_date),
845 to_date=date_to_api(host_request.to_date),
846 created=Timestamp_from_datetime(host_request.conversation.created),
847 messages=get_messages_for_conversation(host_request.conversation_id),
848 )
850 def get_group_chat_pb(group_chat: GroupChat) -> admin_pb2.AdminGroupChat:
851 subs = (
852 session.execute(
853 select(GroupChatSubscription)
854 .where(GroupChatSubscription.group_chat_id == group_chat.conversation_id)
855 .order_by(GroupChatSubscription.joined.asc())
856 )
857 .scalars()
858 .all()
859 )
860 members = [
861 admin_pb2.GroupChatMember(
862 user=get_chat_user_info(sub.user_id),
863 joined=Timestamp_from_datetime(sub.joined),
864 left=Timestamp_from_datetime(sub.left) if sub.left else None,
865 role=sub.role.name if sub.role else "",
866 )
867 for sub in subs
868 ]
869 return admin_pb2.AdminGroupChat(
870 group_chat_id=group_chat.conversation_id,
871 title=group_chat.title or "",
872 is_dm=group_chat.is_dm,
873 creator=get_chat_user_info(group_chat.creator_id),
874 members=members,
875 messages=get_messages_for_conversation(group_chat.conversation_id),
876 )
878 # Get all host requests for the user
879 host_requests = (
880 session.execute(
881 select(HostRequest)
882 .where(or_(HostRequest.recipient_user_id == user.id, HostRequest.initiator_user_id == user.id))
883 .order_by(HostRequest.conversation_id.desc())
884 )
885 .scalars()
886 .all()
887 )
889 # Get all group chats for the user
890 group_chat_ids = (
891 session.execute(
892 select(GroupChatSubscription.group_chat_id)
893 .where(GroupChatSubscription.user_id == user.id)
894 .order_by(GroupChatSubscription.joined.desc())
895 )
896 .scalars()
897 .all()
898 )
899 group_chats = (
900 session.execute(select(GroupChat).where(GroupChat.conversation_id.in_(group_chat_ids))).scalars().all()
901 )
903 # Build protobuf objects, then sort by latest message time (most recent first)
904 host_request_pbs = [get_host_request_pb(hr) for hr in host_requests]
905 host_request_pbs.sort(key=lambda hr: hr.messages[-1].time.seconds if hr.messages else 0, reverse=True)
907 group_chat_pbs = [get_group_chat_pb(gc) for gc in group_chats]
908 group_chat_pbs.sort(key=lambda gc: gc.messages[-1].time.seconds if gc.messages else 0, reverse=True)
910 return admin_pb2.GetChatsRes(
911 user=get_chat_user_info(user.id),
912 host_requests=host_request_pbs,
913 group_chats=group_chat_pbs,
914 )
916 def DeleteEvent(
917 self, request: admin_pb2.DeleteEventReq, context: CouchersContext, session: Session
918 ) -> empty_pb2.Empty:
919 res = session.execute(
920 select(Event, EventOccurrence)
921 .where(EventOccurrence.id == request.event_id)
922 .where(EventOccurrence.event_id == Event.id)
923 .where(~EventOccurrence.is_deleted)
924 ).one_or_none()
926 if not res: 926 ↛ 927line 926 didn't jump to line 927 because the condition on line 926 was never true
927 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
929 event, occurrence = res
931 occurrence.is_deleted = True
933 queue_job(
934 session,
935 job=generate_event_delete_notifications,
936 payload=jobs_pb2.GenerateEventDeleteNotificationsPayload(
937 occurrence_id=occurrence.id,
938 ),
939 )
941 return empty_pb2.Empty()
943 def ListUserIds(
944 self, request: admin_pb2.ListUserIdsReq, context: CouchersContext, session: Session
945 ) -> admin_pb2.ListUserIdsRes:
946 start_date = to_aware_datetime(request.start_time)
947 end_date = to_aware_datetime(request.end_time)
949 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
950 next_user_id = int(request.page_token) if request.page_token else 0
952 user_ids = (
953 session.execute(
954 select(User.id)
955 .where(or_(User.id <= next_user_id, to_bool(next_user_id == 0)))
956 .where(User.joined >= start_date)
957 .where(User.joined <= end_date)
958 .order_by(User.id.desc())
959 .limit(page_size + 1)
960 )
961 .scalars()
962 .all()
963 )
965 return admin_pb2.ListUserIdsRes(
966 user_ids=user_ids[:page_size],
967 next_page_token=str(user_ids[-1]) if len(user_ids) > page_size else None,
968 )
970 def EditReferenceText(
971 self, request: admin_pb2.EditReferenceTextReq, context: CouchersContext, session: Session
972 ) -> empty_pb2.Empty:
973 reference = session.execute(select(Reference).where(Reference.id == request.reference_id)).scalar_one_or_none()
975 if reference is None: 975 ↛ 976line 975 didn't jump to line 976 because the condition on line 975 was never true
976 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:reference_not_found")
978 if not request.new_text.strip(): 978 ↛ 979line 978 didn't jump to line 979 because the condition on line 978 was never true
979 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "reference_no_text")
981 reference.text = request.new_text.strip()
982 # Log action against the reference author
983 author = session.execute(select(User).where(User.id == reference.from_user_id)).scalar_one()
984 log_admin_action(session, context, author, "edit_reference", note=f"Edited reference {reference.id}")
985 return empty_pb2.Empty()
987 def DeleteReference(
988 self, request: admin_pb2.DeleteReferenceReq, context: CouchersContext, session: Session
989 ) -> empty_pb2.Empty:
990 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:deletereference_deprecated_use_ums")
992 def GetUserReferences(
993 self, request: admin_pb2.GetUserReferencesReq, context: CouchersContext, session: Session
994 ) -> admin_pb2.GetUserReferencesRes:
995 user = find_user_by_username_or_email_or_id(session, context, request.user)
997 get_chat_user_info = _make_chat_user_info_getter(session)
999 references_from = (
1000 session.execute(
1001 select(Reference)
1002 .where(Reference.from_user_id == user.id)
1003 .options(selectinload(Reference.host_request), selectinload(Reference.moderation_state))
1004 .order_by(Reference.id.desc())
1005 )
1006 .scalars()
1007 .all()
1008 )
1010 references_to = (
1011 session.execute(
1012 select(Reference)
1013 .where(Reference.to_user_id == user.id)
1014 .options(selectinload(Reference.host_request), selectinload(Reference.moderation_state))
1015 .order_by(Reference.id.desc())
1016 )
1017 .scalars()
1018 .all()
1019 )
1021 return admin_pb2.GetUserReferencesRes(
1022 references_from=[_reference_to_pb(ref, get_chat_user_info) for ref in references_from],
1023 references_to=[_reference_to_pb(ref, get_chat_user_info) for ref in references_to],
1024 )
1026 def GetFriendRequests(
1027 self, request: admin_pb2.GetFriendRequestsReq, context: CouchersContext, session: Session
1028 ) -> admin_pb2.GetFriendRequestsRes:
1029 user = find_user_by_username_or_email_or_id(session, context, request.user)
1031 get_chat_user_info = _make_chat_user_info_getter(session)
1033 def friend_request_to_pb(rel: FriendRelationship) -> admin_pb2.AdminFriendRequest:
1034 return admin_pb2.AdminFriendRequest(
1035 friend_request_id=rel.id,
1036 from_user=get_chat_user_info(rel.from_user_id),
1037 to_user=get_chat_user_info(rel.to_user_id),
1038 status=rel.status.name if rel.status else "",
1039 time_sent=Timestamp_from_datetime(rel.time_sent),
1040 time_responded=Timestamp_from_datetime(rel.time_responded) if rel.time_responded else None,
1041 moderation_visibility=not_none(rel.moderation_state.visibility).name,
1042 )
1044 sent = (
1045 session.execute(
1046 select(FriendRelationship)
1047 .where(FriendRelationship.from_user_id == user.id)
1048 .order_by(FriendRelationship.id.desc())
1049 )
1050 .scalars()
1051 .all()
1052 )
1054 received = (
1055 session.execute(
1056 select(FriendRelationship)
1057 .where(FriendRelationship.to_user_id == user.id)
1058 .order_by(FriendRelationship.id.desc())
1059 )
1060 .scalars()
1061 .all()
1062 )
1064 return admin_pb2.GetFriendRequestsRes(
1065 sent=[friend_request_to_pb(rel) for rel in sent],
1066 received=[friend_request_to_pb(rel) for rel in received],
1067 )
1069 def GetUserBlocks(
1070 self, request: admin_pb2.GetUserBlocksReq, context: CouchersContext, session: Session
1071 ) -> admin_pb2.GetUserBlocksRes:
1072 user = find_user_by_username_or_email_or_id(session, context, request.user)
1074 def user_block_to_pb(block: UserBlock, other_user: User) -> admin_pb2.AdminUserBlock:
1075 return admin_pb2.AdminUserBlock(
1076 user=admin_pb2.ChatUserInfo(
1077 user_id=other_user.id,
1078 username=other_user.username,
1079 name=other_user.name,
1080 birthdate=date_to_api(other_user.birthdate),
1081 gender=other_user.gender,
1082 ),
1083 time_blocked=Timestamp_from_datetime(block.time_blocked),
1084 )
1086 blocked_user = aliased(User)
1087 blocked_users = session.execute(
1088 select(UserBlock, blocked_user)
1089 .join(blocked_user, UserBlock.blocked_user_id == blocked_user.id)
1090 .where(UserBlock.blocking_user_id == user.id)
1091 .order_by(UserBlock.time_blocked.desc(), UserBlock.id.desc())
1092 ).all()
1094 blocking_user = aliased(User)
1095 blocking_users = session.execute(
1096 select(UserBlock, blocking_user)
1097 .join(blocking_user, UserBlock.blocking_user_id == blocking_user.id)
1098 .where(UserBlock.blocked_user_id == user.id)
1099 .order_by(UserBlock.time_blocked.desc(), UserBlock.id.desc())
1100 ).all()
1102 return admin_pb2.GetUserBlocksRes(
1103 blocked_users=[user_block_to_pb(block, other_user) for block, other_user in blocked_users],
1104 blocking_users=[user_block_to_pb(block, other_user) for block, other_user in blocking_users],
1105 )
1107 def GetNonvisibleUserAccessLog(
1108 self, request: admin_pb2.GetNonvisibleUserAccessLogReq, context: CouchersContext, session: Session
1109 ) -> admin_pb2.GetNonvisibleUserAccessLogRes:
1110 user = find_user_by_username_or_email_or_id(session, context, request.user)
1112 actor = aliased(User)
1113 rows = session.execute(
1114 select(NonvisibleUserAccess, actor.username)
1115 .outerjoin(actor, NonvisibleUserAccess.actor_user_id == actor.id)
1116 .where(NonvisibleUserAccess.target_user_id == user.id)
1117 .order_by(NonvisibleUserAccess.time.desc())
1118 .limit(MAX_PAGINATION_LENGTH)
1119 ).all()
1121 return admin_pb2.GetNonvisibleUserAccessLogRes(
1122 entries=[
1123 admin_pb2.NonvisibleUserAccessLogEntry(
1124 time=Timestamp_from_datetime(access.time),
1125 access_type=nonvisibleuseraccesstype2api[access.access_type],
1126 target_state=nonvisibleuserstate2api[access.target_state],
1127 target_user_id=access.target_user_id,
1128 actor_user_id=Int64Value(value=access.actor_user_id) if access.actor_user_id is not None else None,
1129 actor_username=actor_username or "",
1130 ip_address=access.ip_address or "",
1131 user_agent=access.user_agent or "",
1132 sofa=access.sofa or "",
1133 )
1134 for access, actor_username in rows
1135 ]
1136 )
1138 def EditDiscussion(
1139 self, request: admin_pb2.EditDiscussionReq, context: CouchersContext, session: Session
1140 ) -> empty_pb2.Empty:
1141 discussion = session.execute(
1142 select(Discussion).where(Discussion.id == request.discussion_id)
1143 ).scalar_one_or_none()
1144 if not discussion:
1145 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found")
1146 if request.new_title:
1147 discussion.title = request.new_title.strip()
1148 if request.new_content:
1149 discussion.content = request.new_content.strip()
1150 return empty_pb2.Empty()
1152 def DeleteDiscussion(
1153 self, request: admin_pb2.AdminDeleteDiscussionReq, context: CouchersContext, session: Session
1154 ) -> empty_pb2.Empty:
1155 discussion = session.execute(
1156 select(Discussion).where(Discussion.id == request.discussion_id)
1157 ).scalar_one_or_none()
1158 if not discussion: 1158 ↛ 1159line 1158 didn't jump to line 1159 because the condition on line 1158 was never true
1159 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found")
1160 if discussion.deleted is not None: 1160 ↛ 1161line 1160 didn't jump to line 1161 because the condition on line 1160 was never true
1161 return empty_pb2.Empty()
1162 session.add(
1163 DiscussionVersion(
1164 discussion_id=discussion.id,
1165 editor_user_id=context.user_id,
1166 change_type=ContentChangeType.delete,
1167 old_title=discussion.title,
1168 new_title=None,
1169 old_content=discussion.content,
1170 new_content=None,
1171 )
1172 )
1173 discussion.deleted = now()
1174 return empty_pb2.Empty()
1176 def EditReply(self, request: admin_pb2.EditReplyReq, context: CouchersContext, session: Session) -> empty_pb2.Empty:
1177 database_id, depth = unpack_thread_id(request.reply_id)
1178 if depth == 1:
1179 obj: Comment | Reply | None = session.execute(
1180 select(Comment).where(Comment.id == database_id)
1181 ).scalar_one_or_none()
1182 elif depth == 2:
1183 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
1184 else:
1185 obj = None
1187 if not obj:
1188 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:object_not_found")
1189 old_content = obj.content
1190 new_content = request.new_content.strip()
1191 if depth == 1:
1192 session.add(
1193 CommentVersion(
1194 comment_id=database_id,
1195 editor_user_id=context.user_id,
1196 change_type=ContentChangeType.edit,
1197 old_content=old_content,
1198 new_content=new_content,
1199 )
1200 )
1201 else:
1202 session.add(
1203 ReplyVersion(
1204 reply_id=database_id,
1205 editor_user_id=context.user_id,
1206 change_type=ContentChangeType.edit,
1207 old_content=old_content,
1208 new_content=new_content,
1209 )
1210 )
1211 obj.content = new_content
1212 return empty_pb2.Empty()
1214 def AddUsersToModerationUserList(
1215 self, request: admin_pb2.AddUsersToModerationUserListReq, context: CouchersContext, session: Session
1216 ) -> admin_pb2.AddUsersToModerationUserListRes:
1217 """Add multiple users to a moderation user list. If no moderation list is provided, a new one is created.
1218 Id of the moderation list is returned."""
1219 req_users = request.users
1220 users = []
1222 for req_user in req_users:
1223 user = find_user_by_username_or_email_or_id(session, context, req_user)
1224 users.append(user)
1226 if request.moderation_list_id:
1227 moderation_user_list = session.get(ModerationUserList, request.moderation_list_id)
1228 if not moderation_user_list:
1229 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:moderation_user_list_not_found")
1230 # Create a new moderation user list if no one is provided
1231 else:
1232 moderation_user_list = ModerationUserList()
1233 session.add(moderation_user_list)
1234 session.flush()
1236 # Add users to the moderation list only if not already in it
1237 for user in users:
1238 if user not in moderation_user_list.users: 1238 ↛ 1240line 1238 didn't jump to line 1240 because the condition on line 1238 was always true
1239 moderation_user_list.users.append(user)
1240 log_admin_action(session, context, user, "add_to_moderation_list")
1242 return admin_pb2.AddUsersToModerationUserListRes(moderation_list_id=moderation_user_list.id)
1244 def ListModerationUserLists(
1245 self, request: admin_pb2.ListModerationUserListsReq, context: CouchersContext, session: Session
1246 ) -> admin_pb2.ListModerationUserListsRes:
1247 """Lists all moderation user lists for a user."""
1248 user = find_user_by_username_or_email_or_id(session, context, request.user)
1250 moderation_lists = [
1251 admin_pb2.ModerationList(
1252 moderation_list_id=ml.id,
1253 members=[_user_to_details(session, u) for u in ml.users],
1254 )
1255 for ml in user.moderation_user_lists
1256 ]
1257 return admin_pb2.ListModerationUserListsRes(moderation_lists=moderation_lists)
1259 def RemoveUserFromModerationUserList(
1260 self, request: admin_pb2.RemoveUserFromModerationUserListReq, context: CouchersContext, session: Session
1261 ) -> empty_pb2.Empty:
1262 """Removes a user from a provided moderation user list."""
1263 user = find_user_by_username_or_email_or_id(session, context, request.user)
1264 if not request.moderation_list_id:
1265 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:missing_moderation_user_list_id")
1267 moderation_user_list = session.get(ModerationUserList, request.moderation_list_id)
1268 if not moderation_user_list: 1268 ↛ 1269line 1268 didn't jump to line 1269 because the condition on line 1268 was never true
1269 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:moderation_user_list_not_found")
1270 if user not in moderation_user_list.users:
1271 context.abort_with_error_code(
1272 grpc.StatusCode.FAILED_PRECONDITION, "admin:user_not_in_the_moderation_user_list"
1273 )
1275 moderation_user_list.users.remove(user)
1276 log_admin_action(session, context, user, "remove_from_moderation_list")
1278 if len(moderation_user_list.users) == 0:
1279 session.delete(moderation_user_list)
1281 return empty_pb2.Empty()
1283 def CreateAccountDeletionLink(
1284 self, request: admin_pb2.CreateAccountDeletionLinkReq, context: CouchersContext, session: Session
1285 ) -> admin_pb2.CreateAccountDeletionLinkRes:
1286 user = find_user_by_username_or_email_or_id(session, context, request.user)
1287 token = AccountDeletionToken(token=urlsafe_secure_token(), user_id=user.id, expiry=now() + timedelta(hours=2))
1288 session.add(token)
1289 log_admin_action(session, context, user, "create_account_deletion_link", level=AdminActionLevel.high)
1290 return admin_pb2.CreateAccountDeletionLinkRes(
1291 account_deletion_confirm_url=urls.delete_account_link(account_deletion_token=token.token)
1292 )
1294 def AccessStats(
1295 self, request: admin_pb2.AccessStatsReq, context: CouchersContext, session: Session
1296 ) -> admin_pb2.AccessStatsRes:
1297 user = find_user_by_username_or_email_or_id(session, context, request.user)
1299 start_time = (
1300 to_aware_datetime(request.start_time) if request.HasField("start_time") else now() - timedelta(days=90)
1301 )
1302 end_time = to_aware_datetime(request.end_time) if request.HasField("end_time") else now()
1304 user_activity = session.execute(
1305 select(
1306 UserActivity.ip_address,
1307 UserActivity.user_agent,
1308 func.sum(UserActivity.api_calls),
1309 func.count(UserActivity.period),
1310 func.min(UserActivity.period),
1311 func.max(UserActivity.period),
1312 )
1313 .where(UserActivity.user_id == user.id)
1314 .where(UserActivity.period >= start_time)
1315 .where(UserActivity.period <= end_time)
1316 .order_by(func.max(UserActivity.period).desc())
1317 .group_by(UserActivity.ip_address, UserActivity.user_agent)
1318 ).all()
1320 out = admin_pb2.AccessStatsRes()
1322 for ip_address, user_agent, api_call_count, periods_count, first_seen, last_seen in user_activity:
1323 ip_address_str = str(ip_address) if ip_address is not None else None
1324 user_agent_data = user_agents_parse(user_agent or "")
1325 asn = geoip_asn(ip_address_str)
1326 out.stats.append(
1327 admin_pb2.AccessStat(
1328 ip_address=ip_address_str,
1329 asn=str(asn[0]) if asn else None,
1330 asorg=str(asn[1]) if asn else None,
1331 asnetwork=str(asn[2]) if asn else None,
1332 user_agent=user_agent,
1333 operating_system=user_agent_data.os.family,
1334 browser=user_agent_data.browser.family,
1335 device=user_agent_data.device.family,
1336 approximate_location=geoip_approximate_location(ip_address_str) or "Unknown",
1337 api_call_count=api_call_count,
1338 periods_count=periods_count,
1339 first_seen=Timestamp_from_datetime(first_seen),
1340 last_seen=Timestamp_from_datetime(last_seen),
1341 )
1342 )
1344 return out
1346 def SetLastDonated(
1347 self, request: admin_pb2.SetLastDonatedReq, context: CouchersContext, session: Session
1348 ) -> admin_pb2.UserDetails:
1349 user = find_user_by_username_or_email_or_id(session, context, request.user)
1351 if request.HasField("last_donated"):
1352 user.last_donated = to_aware_datetime(request.last_donated)
1353 else:
1354 user.last_donated = None
1356 log_admin_action(session, context, user, "set_last_donated")
1357 return _user_to_details(session, user)
1359 def CreateAdminTag(
1360 self, request: admin_pb2.CreateAdminTagReq, context: CouchersContext, session: Session
1361 ) -> admin_pb2.AdminTagInfo:
1362 if not request.tag.strip():
1363 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:tag_cant_be_empty")
1364 existing = session.execute(select(AdminTag).where(AdminTag.tag == request.tag.strip())).scalar_one_or_none()
1365 if existing:
1366 context.abort_with_error_code(grpc.StatusCode.ALREADY_EXISTS, "admin:tag_already_exists")
1367 admin_tag = AdminTag(tag=request.tag.strip())
1368 session.add(admin_tag)
1369 session.flush()
1370 return admin_pb2.AdminTagInfo(admin_tag_id=admin_tag.id, tag=admin_tag.tag)
1372 def ListAdminTags(
1373 self, request: admin_pb2.ListAdminTagsReq, context: CouchersContext, session: Session
1374 ) -> admin_pb2.ListAdminTagsRes:
1375 tags = session.execute(select(AdminTag).order_by(AdminTag.tag)).scalars().all()
1376 return admin_pb2.ListAdminTagsRes(
1377 tags=[admin_pb2.AdminTagInfo(admin_tag_id=tag.id, tag=tag.tag) for tag in tags]
1378 )
1380 def AddAdminTagToUser(
1381 self, request: admin_pb2.AddAdminTagToUserReq, context: CouchersContext, session: Session
1382 ) -> admin_pb2.UserDetails:
1383 user = find_user_by_username_or_email_or_id(session, context, request.user)
1384 admin_tag = session.execute(select(AdminTag).where(AdminTag.tag == request.tag)).scalar_one_or_none()
1385 if not admin_tag:
1386 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:tag_not_found")
1387 existing = session.execute(
1388 select(UserAdminTag).where(UserAdminTag.user_id == user.id, UserAdminTag.admin_tag_id == admin_tag.id)
1389 ).scalar_one_or_none()
1390 if existing:
1391 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_already_has_admin_tag")
1392 session.add(UserAdminTag(user_id=user.id, admin_tag_id=admin_tag.id))
1393 session.flush()
1394 log_admin_action(session, context, user, "add_tag", tag=request.tag)
1395 return _user_to_details(session, user)
1397 def RemoveAdminTagFromUser(
1398 self, request: admin_pb2.RemoveAdminTagFromUserReq, context: CouchersContext, session: Session
1399 ) -> admin_pb2.UserDetails:
1400 user = find_user_by_username_or_email_or_id(session, context, request.user)
1401 admin_tag = session.execute(select(AdminTag).where(AdminTag.tag == request.tag)).scalar_one_or_none()
1402 if not admin_tag: 1402 ↛ 1403line 1402 didn't jump to line 1403 because the condition on line 1402 was never true
1403 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:tag_not_found")
1404 user_admin_tag = session.execute(
1405 select(UserAdminTag).where(UserAdminTag.user_id == user.id, UserAdminTag.admin_tag_id == admin_tag.id)
1406 ).scalar_one_or_none()
1407 if not user_admin_tag:
1408 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_does_not_have_admin_tag")
1409 session.delete(user_admin_tag)
1410 session.flush()
1411 log_admin_action(session, context, user, "remove_tag", tag=request.tag)
1412 return _user_to_details(session, user)
1414 def SetModScore(
1415 self, request: admin_pb2.SetModScoreReq, context: CouchersContext, session: Session
1416 ) -> admin_pb2.UserDetails:
1417 user = find_user_by_username_or_email_or_id(session, context, request.user)
1418 user.mod_score = request.mod_score
1419 log_admin_action(session, context, user, "set_mod_score", note=f"mod_score={request.mod_score}")
1420 return _user_to_details(session, user)
1422 def ListAdminActions(
1423 self, request: admin_pb2.ListAdminActionsReq, context: CouchersContext, session: Session
1424 ) -> admin_pb2.ListAdminActionsRes:
1425 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
1427 admin_user = aliased(User)
1428 target_user = aliased(User)
1430 statement = (
1431 select(AdminAction, admin_user.username, target_user.username)
1432 .join(admin_user, AdminAction.admin_user_id == admin_user.id)
1433 .join(target_user, AdminAction.target_user_id == target_user.id)
1434 )
1436 if request.admin_user_id:
1437 statement = statement.where(AdminAction.admin_user_id == request.admin_user_id)
1438 if request.target_user_id:
1439 statement = statement.where(AdminAction.target_user_id == request.target_user_id)
1440 if request.page_token:
1441 statement = statement.where(AdminAction.id < int(request.page_token))
1443 statement = statement.order_by(AdminAction.id.desc()).limit(page_size + 1)
1445 rows = session.execute(statement).all()
1447 action_pbs = [
1448 admin_pb2.AdminActionLog(
1449 admin_action_id=action.id,
1450 created=Timestamp_from_datetime(action.created),
1451 admin_user_id=action.admin_user_id,
1452 admin_username=admin_username,
1453 action_type=action.action_type,
1454 level=adminactionlevel2api[action.level],
1455 note=action.note or "",
1456 data=json.dumps(action.data) if action.data is not None else "",
1457 tag=action.tag or "",
1458 target_user_id=action.target_user_id,
1459 target_username=target_username,
1460 )
1461 for action, admin_username, target_username in rows[:page_size]
1462 ]
1464 return admin_pb2.ListAdminActionsRes(
1465 admin_actions=action_pbs,
1466 next_page_token=str(rows[page_size - 1][0].id) if len(rows) > page_size else None,
1467 )
1469 def ListUserUploads(
1470 self, request: admin_pb2.ListUserUploadsReq, context: CouchersContext, session: Session
1471 ) -> admin_pb2.ListUserUploadsRes:
1472 user = find_user_by_username_or_email_or_id(session, context, request.user)
1474 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
1476 statement = (
1477 select(Upload)
1478 .where(Upload.creator_user_id == user.id)
1479 .options(
1480 undefer(Upload.metadata_parsed),
1481 undefer(Upload.metadata_parse_error),
1482 undefer(Upload.original_filename),
1483 undefer(Upload.original_format),
1484 undefer(Upload.original_size),
1485 undefer(Upload.original_width),
1486 undefer(Upload.original_height),
1487 )
1488 )
1489 if request.page_token:
1490 cursor_created = session.execute(
1491 select(Upload.created).where(Upload.key == request.page_token)
1492 ).scalar_one()
1493 statement = statement.where(tuple_(Upload.created, Upload.key) < (cursor_created, request.page_token))
1495 uploads = (
1496 session.execute(statement.order_by(Upload.created.desc(), Upload.key.desc()).limit(page_size + 1))
1497 .scalars()
1498 .all()
1499 )
1501 page = uploads[:page_size]
1502 uses_by_key = get_upload_uses_for_keys(session, [upload.key for upload in page])
1504 return admin_pb2.ListUserUploadsRes(
1505 uploads=[
1506 admin_pb2.UserUpload(
1507 key=upload.key,
1508 filename=upload.filename,
1509 full_url=upload.full_url,
1510 thumbnail_url=upload.thumbnail_url,
1511 credit=upload.credit or "",
1512 created=Timestamp_from_datetime(upload.created),
1513 metadata=_upload_metadata_to_pb(upload),
1514 uses=[
1515 admin_pb2.UploadUse(
1516 type=uploadusetype2api[use.use_type],
1517 is_current=use.is_current,
1518 user_id=use.user_id,
1519 event_id=use.event_id,
1520 page_id=use.page_id,
1521 url=use.url,
1522 )
1523 for use in uses_by_key.get(upload.key, [])
1524 ],
1525 )
1526 for upload in page
1527 ],
1528 next_page_token=uploads[page_size - 1].key if len(uploads) > page_size else None,
1529 )
1531 def ExportUserData(
1532 self, request: admin_pb2.ExportUserDataReq, context: CouchersContext, session: Session
1533 ) -> admin_pb2.ExportUserDataRes:
1534 user = find_user_by_username_or_email_or_id(session, context, request.user)
1536 generated = now()
1537 export = generate_user_data_export(session, user, generated)
1538 log_admin_action(session, context, user, "export_user_data")
1540 return admin_pb2.ExportUserDataRes(
1541 user_id=user.id,
1542 generated=Timestamp_from_datetime(generated),
1543 filename=f"couchers-data-export-{user.username}-{generated.date().isoformat()}.json",
1544 json=MessageToJson(
1545 export, indent=2, preserving_proto_field_name=True, always_print_fields_with_no_presence=True
1546 ),
1547 )
1549 def CreateOTAPackage(
1550 self, request: admin_pb2.CreateOTAPackageReq, context: CouchersContext, session: Session
1551 ) -> admin_pb2.OTAPackage:
1552 platform = api2otaplatform.get(request.platform)
1553 if platform is None: 1553 ↛ 1554line 1553 didn't jump to line 1554 because the condition on line 1553 was never true
1554 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_platform")
1556 if not request.version:
1557 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_version")
1559 existing = session.execute(
1560 select(OTAPackage.id).where(OTAPackage.platform == platform).where(OTAPackage.version == request.version)
1561 ).scalar_one_or_none()
1562 if existing is not None:
1563 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:ota_package_already_exists")
1565 # Read the keying/ordering fields out of the manifest we're about to serve, so the row can't
1566 # disagree with the bytes on the CDN.
1567 cdn_root = context.get_string_value("native_ota_cdn_root", "https://cdn.couchers.org/native/ota")
1568 _content_type, body = _fetch_signed_manifest(
1569 _native_ota_manifest_url(cdn_root=cdn_root, version=request.version, platform=platform.name)
1570 )
1571 manifest = _extract_ota_manifest(body)
1572 fingerprint = manifest.get("runtimeVersion") if manifest else None
1573 manifest_id = manifest.get("id") if manifest else None
1574 created_at_raw = manifest.get("createdAt") if manifest else None
1575 if (
1576 manifest is None
1577 or not isinstance(fingerprint, str)
1578 or not fingerprint
1579 or not isinstance(manifest_id, str)
1580 or not manifest_id
1581 or not isinstance(created_at_raw, str)
1582 or not created_at_raw
1583 ):
1584 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_manifest")
1585 try:
1586 manifest_created_at = datetime.fromisoformat(created_at_raw)
1587 except ValueError:
1588 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_manifest")
1589 if manifest_created_at.tzinfo is None: 1589 ↛ 1590line 1589 didn't jump to line 1590 because the condition on line 1589 was never true
1590 manifest_created_at = manifest_created_at.replace(tzinfo=UTC)
1592 package = OTAPackage(
1593 creator_user_id=context.user_id,
1594 platform=platform,
1595 fingerprint=fingerprint,
1596 version=request.version,
1597 manifest_created_at=manifest_created_at,
1598 manifest_id=manifest_id,
1599 )
1600 session.add(package)
1601 session.flush()
1603 return _ota_package_to_pb(package, _live_ota_package_ids(session))
1605 def ListOTAPackages(
1606 self, request: admin_pb2.ListOTAPackagesReq, context: CouchersContext, session: Session
1607 ) -> admin_pb2.ListOTAPackagesRes:
1608 statement = select(OTAPackage).order_by(OTAPackage.manifest_created_at.desc(), OTAPackage.id.desc())
1609 if request.platform != admin_pb2.OTA_PLATFORM_UNSPECIFIED:
1610 platform = api2otaplatform.get(request.platform)
1611 if platform is None: 1611 ↛ 1612line 1611 didn't jump to line 1612 because the condition on line 1611 was never true
1612 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_platform")
1613 statement = statement.where(OTAPackage.platform == platform)
1614 if request.fingerprint: 1614 ↛ 1615line 1614 didn't jump to line 1615 because the condition on line 1614 was never true
1615 statement = statement.where(OTAPackage.fingerprint == request.fingerprint)
1616 if not request.include_banned:
1617 statement = statement.where(OTAPackage.banned_at.is_(None))
1619 packages = session.execute(statement).scalars().all()
1620 live_ids = _live_ota_package_ids(session)
1621 return admin_pb2.ListOTAPackagesRes(packages=[_ota_package_to_pb(package, live_ids) for package in packages])
1623 def BanOTAPackage(
1624 self, request: admin_pb2.BanOTAPackageReq, context: CouchersContext, session: Session
1625 ) -> admin_pb2.OTAPackage:
1626 # Bans are irreversible — to roll back an accidental ban, republish the bundle as a new
1627 # package — so a reason is required for the audit trail.
1628 if not request.reason.strip():
1629 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:ota_ban_reason_required")
1631 package = session.execute(
1632 select(OTAPackage).where(OTAPackage.id == request.ota_package_id)
1633 ).scalar_one_or_none()
1634 if package is None:
1635 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:ota_package_not_found")
1637 if package.banned_at is None: 1637 ↛ 1641line 1637 didn't jump to line 1641 because the condition on line 1637 was always true
1638 package.banned_at = now()
1639 package.banned_by_user_id = context.user_id
1640 package.banned_reason = request.reason
1641 session.flush()
1643 return _ota_package_to_pb(package, _live_ota_package_ids(session))