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