Coverage for app/backend/src/couchers/servicers/admin.py: 79%
645 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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}
129# Doubles as the set of accepted values: MOD_NOTE_NOTIFICATION_UNSPECIFIED is rejected.
130_MOD_NOTE_NOTIFY_LOG_LABELS = {
131 admin_pb2.MOD_NOTE_NOTIFICATION_NONE: "No",
132 admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY: "Yes",
133 admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT: "Yes, including the note text",
134}
137def log_admin_action(
138 session: Session,
139 context: CouchersContext,
140 target_user: User,
141 action_type: str,
142 note: str | None = None,
143 data: object | None = None,
144 tag: str | None = None,
145 level: AdminActionLevel = AdminActionLevel.normal,
146) -> AdminAction:
147 action = AdminAction(
148 admin_user_id=context.user_id,
149 target_user_id=target_user.id,
150 action_type=action_type,
151 level=level,
152 note=note,
153 data=data,
154 tag=tag,
155 )
156 session.add(action)
157 session.flush()
158 return action
161def _live_ota_package_ids(session: Session) -> set[int]:
162 # The live package per (platform, fingerprint) is the newest non-banned one by manifest_created_at,
163 # matching what GetNativeUpdateManifest resolves. DISTINCT ON picks the row with the leading ORDER BY
164 # value per (platform, fingerprint) group in a single index-friendly query.
165 return set(
166 session.scalars(
167 select(OTAPackage.id)
168 .where(OTAPackage.banned_at.is_(None))
169 .distinct(OTAPackage.platform, OTAPackage.fingerprint)
170 .order_by(
171 OTAPackage.platform,
172 OTAPackage.fingerprint,
173 OTAPackage.manifest_created_at.desc(),
174 OTAPackage.id.desc(),
175 )
176 )
177 )
180def _extract_ota_manifest(body: bytes) -> dict[str, Any] | None:
181 # The manifest object is the JSON in the "manifest" part of the signed multipart/mixed body.
182 marker = body.find(b'name="manifest"')
183 if marker == -1:
184 return None
185 body_start = body.find(b"\r\n\r\n", marker)
186 if body_start == -1: 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 return None
188 body_end = body.find(b"\r\n--", body_start + 4)
189 if body_end == -1: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 return None
191 try:
192 manifest = json.loads(body[body_start + 4 : body_end])
193 except json.JSONDecodeError:
194 return None
195 return manifest if isinstance(manifest, dict) else None
198def _ota_package_to_pb(package: OTAPackage, live_ids: set[int]) -> admin_pb2.OTAPackage:
199 return admin_pb2.OTAPackage(
200 ota_package_id=package.id,
201 created=Timestamp_from_datetime(package.created),
202 creator_user_id=package.creator_user_id,
203 platform=otaplatform2api[package.platform],
204 fingerprint=package.fingerprint,
205 version=package.version,
206 manifest_created_at=Timestamp_from_datetime(package.manifest_created_at),
207 manifest_id=package.manifest_id,
208 banned=package.banned_at is not None,
209 banned_at=Timestamp_from_datetime(package.banned_at) if package.banned_at else None,
210 banned_by_user_id=package.banned_by_user_id or 0,
211 banned_reason=package.banned_reason or "",
212 live=package.id in live_ids,
213 )
216def _user_to_details(session: Session, user: User) -> admin_pb2.UserDetails:
217 # Query admin actions for this user
218 actions = session.execute(
219 select(AdminAction, User.username)
220 .join(User, AdminAction.admin_user_id == User.id)
221 .where(AdminAction.target_user_id == user.id)
222 .order_by(AdminAction.created.asc())
223 ).all()
225 action_pbs = []
226 for action, admin_username in actions:
227 action_pbs.append(
228 admin_pb2.AdminActionLog(
229 admin_action_id=action.id,
230 created=Timestamp_from_datetime(action.created),
231 admin_user_id=action.admin_user_id,
232 admin_username=admin_username,
233 action_type=action.action_type,
234 level=adminactionlevel2api[action.level],
235 note=action.note or "",
236 data=json.dumps(action.data) if action.data is not None else "",
237 tag=action.tag or "",
238 target_user_id=action.target_user_id,
239 target_username=user.username,
240 )
241 )
243 # Query admin tags
244 admin_tags = (
245 session.execute(
246 select(AdminTag.tag)
247 .join(UserAdminTag, UserAdminTag.admin_tag_id == AdminTag.id)
248 .where(UserAdminTag.user_id == user.id)
249 .order_by(AdminTag.tag)
250 )
251 .scalars()
252 .all()
253 )
255 last_mod_note_acknowledged = session.execute(
256 select(func.max(ModNote.acknowledged)).where(ModNote.user_id == user.id)
257 ).scalar()
259 return admin_pb2.UserDetails(
260 user_id=user.id,
261 username=user.username,
262 name=user.name,
263 email=user.email,
264 gender=user.gender,
265 birthdate=date_to_api(user.birthdate),
266 banned=user.banned_at is not None,
267 deleted=user.deleted_at is not None,
268 shadowed=user.shadowed_at is not None,
269 do_not_email=user.do_not_email,
270 badges=[badge.badge_id for badge in user.badges],
271 **get_strong_verification_fields(session, user),
272 has_passport_sex_gender_exception=user.has_passport_sex_gender_exception,
273 pending_mod_notes_count=user.mod_notes.where(ModNote.is_pending).count(),
274 acknowledged_mod_notes_count=user.mod_notes.where(~ModNote.is_pending).count(),
275 last_mod_note_acknowledged=(
276 Timestamp_from_datetime(last_mod_note_acknowledged) if last_mod_note_acknowledged else None
277 ),
278 admin_actions=action_pbs,
279 admin_tags=list(admin_tags),
280 mod_score=user.mod_score,
281 ui_language_preference=user.ui_language_preference,
282 )
285def _content_report_to_pb(content_report: ContentReport) -> admin_pb2.ContentReport:
286 return admin_pb2.ContentReport(
287 content_report_id=content_report.id,
288 time=Timestamp_from_datetime(content_report.time),
289 reporting_user_id=content_report.reporting_user_id,
290 author_user_id=content_report.author_user_id,
291 reason=content_report.reason,
292 description=content_report.description,
293 content_ref=content_report.content_ref,
294 user_agent=content_report.user_agent,
295 page=content_report.page,
296 )
299def _upload_metadata_to_pb(upload: Upload) -> admin_pb2.UploadMetadata:
300 return admin_pb2.UploadMetadata(
301 parsed_json=json.dumps(upload.metadata_parsed, sort_keys=True) if upload.metadata_parsed else None,
302 parse_error=upload.metadata_parse_error,
303 original_filename=upload.original_filename,
304 original_format=upload.original_format,
305 original_size=upload.original_size,
306 original_width=upload.original_width,
307 original_height=upload.original_height,
308 )
311def _make_chat_user_info_getter(session: Session) -> Callable[[int], admin_pb2.ChatUserInfo]:
312 user_info_cache: dict[int, admin_pb2.ChatUserInfo] = {}
314 def get_chat_user_info(user_id: int) -> admin_pb2.ChatUserInfo:
315 if user_id not in user_info_cache:
316 u = session.execute(select(User).where(User.id == user_id)).scalar_one()
317 user_info_cache[user_id] = admin_pb2.ChatUserInfo(
318 user_id=u.id,
319 username=u.username,
320 name=u.name,
321 birthdate=date_to_api(u.birthdate),
322 gender=u.gender,
323 )
324 return user_info_cache[user_id]
326 return get_chat_user_info
329def _reference_to_pb(
330 reference: Reference, get_chat_user_info: Callable[[int], admin_pb2.ChatUserInfo]
331) -> admin_pb2.AdminReference:
332 host_request = reference.host_request
333 return admin_pb2.AdminReference(
334 reference_id=reference.id,
335 from_user_id=reference.from_user_id,
336 to_user_id=reference.to_user_id,
337 from_user=get_chat_user_info(reference.from_user_id),
338 to_user=get_chat_user_info(reference.to_user_id),
339 reference_type=reference.reference_type.name,
340 text=reference.text,
341 private_text=reference.private_text or "",
342 time=Timestamp_from_datetime(reference.time),
343 host_request_id=reference.host_request_id or 0,
344 rating=reference.rating,
345 was_appropriate=reference.was_appropriate,
346 moderation_visibility=not_none(reference.moderation_state.visibility).name,
347 hosting_city=host_request.hosting_city if host_request else "",
348 from_date=date_to_api(host_request.from_date) if host_request else "",
349 to_date=date_to_api(host_request.to_date) if host_request else "",
350 status=host_request.status.name if host_request else "",
351 )
354class Admin(admin_pb2_grpc.AdminServicer):
355 def GetUserDetails(
356 self, request: admin_pb2.GetUserDetailsReq, context: CouchersContext, session: Session
357 ) -> admin_pb2.UserDetails:
358 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
359 if not user: 359 ↛ 360line 359 didn't jump to line 360 because the condition on line 359 was never true
360 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
361 return _user_to_details(session, user)
363 def GetUser(self, request: admin_pb2.GetUserReq, context: CouchersContext, session: Session) -> api_pb2.User:
364 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
365 if not user: 365 ↛ 366line 365 didn't jump to line 366 because the condition on line 365 was never true
366 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
367 return user_model_to_pb(user, session, context, is_admin_see_ghosts=True)
369 def SearchUsers(
370 self, request: admin_pb2.SearchUsersReq, context: CouchersContext, session: Session
371 ) -> admin_pb2.SearchUsersRes:
372 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
373 next_user_id = int(request.page_token) if request.page_token else 0
374 statement = select(User)
375 if request.username: 375 ↛ 376line 375 didn't jump to line 376 because the condition on line 375 was never true
376 statement = statement.where(User.username.ilike(request.username))
377 if request.email: 377 ↛ 378line 377 didn't jump to line 378 because the condition on line 377 was never true
378 statement = statement.where(User.email.ilike(request.email))
379 if request.name: 379 ↛ 380line 379 didn't jump to line 380 because the condition on line 379 was never true
380 statement = statement.where(User.name.ilike(request.name))
381 if request.admin_action_log:
382 statement = statement.where(
383 User.id.in_(select(AdminAction.target_user_id).where(AdminAction.note.ilike(request.admin_action_log)))
384 )
385 if request.city: 385 ↛ 386line 385 didn't jump to line 386 because the condition on line 385 was never true
386 statement = statement.where(User.city.ilike(request.city))
387 if request.min_user_id: 387 ↛ 388line 387 didn't jump to line 388 because the condition on line 387 was never true
388 statement = statement.where(User.id >= request.min_user_id)
389 if request.max_user_id: 389 ↛ 390line 389 didn't jump to line 390 because the condition on line 389 was never true
390 statement = statement.where(User.id <= request.max_user_id)
391 if request.min_birthdate: 391 ↛ 392line 391 didn't jump to line 392 because the condition on line 391 was never true
392 statement = statement.where(User.birthdate >= parse_date(request.min_birthdate))
393 if request.max_birthdate: 393 ↛ 394line 393 didn't jump to line 394 because the condition on line 393 was never true
394 statement = statement.where(User.birthdate <= parse_date(request.max_birthdate))
395 if request.genders: 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true
396 statement = statement.where(User.gender.in_(request.genders))
397 if request.min_joined_date: 397 ↛ 398line 397 didn't jump to line 398 because the condition on line 397 was never true
398 statement = statement.where(User.joined >= parse_date(request.min_joined_date))
399 if request.max_joined_date: 399 ↛ 400line 399 didn't jump to line 400 because the condition on line 399 was never true
400 statement = statement.where(User.joined <= parse_date(request.max_joined_date))
401 if request.min_last_active_date: 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true
402 statement = statement.where(User.last_active >= parse_date(request.min_last_active_date))
403 if request.max_last_active_date: 403 ↛ 404line 403 didn't jump to line 404 because the condition on line 403 was never true
404 statement = statement.where(User.last_active <= parse_date(request.max_last_active_date))
405 if request.genders: 405 ↛ 406line 405 didn't jump to line 406 because the condition on line 405 was never true
406 statement = statement.where(User.gender.in_(request.genders))
407 if request.language_codes: 407 ↛ 408line 407 didn't jump to line 408 because the condition on line 407 was never true
408 statement = statement.join(
409 LanguageAbility,
410 and_(LanguageAbility.user_id == User.id, LanguageAbility.language_code.in_(request.language_codes)),
411 )
412 if request.HasField("is_deleted"): 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 statement = statement.where((User.deleted_at != None) == request.is_deleted.value)
414 if request.HasField("is_banned"): 414 ↛ 415line 414 didn't jump to line 415 because the condition on line 414 was never true
415 statement = statement.where((User.banned_at != None) == request.is_banned.value)
416 if request.HasField("is_shadowed"): 416 ↛ 417line 416 didn't jump to line 417 because the condition on line 416 was never true
417 statement = statement.where((User.shadowed_at != None) == request.is_shadowed.value)
418 if request.HasField("has_avatar"): 418 ↛ 419line 418 didn't jump to line 419 because the condition on line 418 was never true
419 statement = statement.where(has_avatar_photo_expression(User) == request.has_avatar.value)
420 if request.admin_tags:
421 for tag_name in request.admin_tags:
422 statement = statement.where(
423 User.id.in_(
424 select(UserAdminTag.user_id)
425 .join(AdminTag, UserAdminTag.admin_tag_id == AdminTag.id)
426 .where(AdminTag.tag == tag_name)
427 )
428 )
429 users = (
430 session.execute(
431 statement.where(User.id >= next_user_id)
432 .order_by(User.id)
433 .limit(page_size + 1)
434 .options(selectinload(User.badges))
435 )
436 .scalars()
437 .all()
438 )
439 logger.info(users)
440 return admin_pb2.SearchUsersRes(
441 users=[_user_to_details(session, user) for user in users[:page_size]],
442 next_page_token=str(users[-1].id) if len(users) > page_size else None,
443 )
445 def ChangeUserGender(
446 self, request: admin_pb2.ChangeUserGenderReq, context: CouchersContext, session: Session
447 ) -> admin_pb2.UserDetails:
448 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
449 if not user: 449 ↛ 450line 449 didn't jump to line 450 because the condition on line 449 was never true
450 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
451 old_gender = user.gender
452 user.gender = request.gender
453 log_admin_action(
454 session, context, user, "change_gender", note=f"Changed from '{old_gender}' to '{request.gender}'"
455 )
456 session.commit()
458 notify(
459 session,
460 user_id=user.id,
461 topic_action=NotificationTopicAction.gender__change,
462 key="",
463 data=notification_data_pb2.GenderChange(
464 gender=request.gender,
465 ),
466 )
468 return _user_to_details(session, user)
470 def ChangeUserBirthdate(
471 self, request: admin_pb2.ChangeUserBirthdateReq, context: CouchersContext, session: Session
472 ) -> admin_pb2.UserDetails:
473 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
474 if not user: 474 ↛ 475line 474 didn't jump to line 475 because the condition on line 474 was never true
475 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
476 if not (birthdate := parse_date(request.birthdate)): 476 ↛ 477line 476 didn't jump to line 477 because the condition on line 476 was never true
477 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_birthdate")
479 old_birthdate = user.birthdate
480 user.birthdate = birthdate
481 log_admin_action(
482 session, context, user, "change_birthdate", note=f"Changed from {old_birthdate} to {request.birthdate}"
483 )
484 session.commit()
486 notify(
487 session,
488 user_id=user.id,
489 topic_action=NotificationTopicAction.birthdate__change,
490 key="",
491 data=notification_data_pb2.BirthdateChange(
492 birthdate=request.birthdate,
493 ),
494 )
496 return _user_to_details(session, user)
498 def AddBadge(
499 self, request: admin_pb2.AddBadgeReq, context: CouchersContext, session: Session
500 ) -> admin_pb2.UserDetails:
501 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
502 if not user: 502 ↛ 503line 502 didn't jump to line 503 because the condition on line 502 was never true
503 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
505 badge = get_badge_dict().get(request.badge_id)
506 if not badge:
507 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "badge_not_found")
509 if not badge.admin_editable:
510 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:cannot_edit_badge")
512 if badge.id in [b.badge_id for b in user.badges]:
513 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_already_has_badge")
515 user_add_badge(session, user.id, request.badge_id)
516 log_admin_action(session, context, user, "add_badge", note=f"Added badge {request.badge_id}")
518 return _user_to_details(session, user)
520 def RemoveBadge(
521 self, request: admin_pb2.RemoveBadgeReq, context: CouchersContext, session: Session
522 ) -> admin_pb2.UserDetails:
523 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
524 if not user: 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.NOT_FOUND, "user_not_found")
527 badge = get_badge_dict().get(request.badge_id)
528 if not badge: 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true
529 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "badge_not_found")
531 if not badge.admin_editable: 531 ↛ 532line 531 didn't jump to line 532 because the condition on line 531 was never true
532 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:cannot_edit_badge")
534 user_badge = session.execute(
535 select(UserBadge).where(UserBadge.user_id == user.id, UserBadge.badge_id == badge.id)
536 ).scalar_one_or_none()
537 if not user_badge:
538 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_does_not_have_badge")
540 user_remove_badge(session, user.id, request.badge_id)
541 log_admin_action(session, context, user, "remove_badge", note=f"Removed badge {request.badge_id}")
543 return _user_to_details(session, user)
545 def SetPassportSexGenderException(
546 self, request: admin_pb2.SetPassportSexGenderExceptionReq, context: CouchersContext, session: Session
547 ) -> admin_pb2.UserDetails:
548 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
549 if not user: 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
551 old_exception = user.has_passport_sex_gender_exception
552 user.has_passport_sex_gender_exception = request.passport_sex_gender_exception
553 log_admin_action(
554 session,
555 context,
556 user,
557 "set_passport_sex_gender_exception",
558 note=f"Changed from {old_exception} to {request.passport_sex_gender_exception}",
559 )
560 return _user_to_details(session, user)
562 def BanUser(
563 self, request: admin_pb2.BanUserReq, context: CouchersContext, session: Session
564 ) -> admin_pb2.UserDetails:
565 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
566 if not user: 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true
567 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
568 if not request.admin_note.strip(): 568 ↛ 569line 568 didn't jump to line 569 because the condition on line 568 was never true
569 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
570 log_admin_action(session, context, user, "ban", note=request.admin_note, level=AdminActionLevel.high)
571 user.banned_at = now()
572 return _user_to_details(session, user)
574 def UnbanUser(
575 self, request: admin_pb2.UnbanUserReq, context: CouchersContext, session: Session
576 ) -> admin_pb2.UserDetails:
577 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
578 if not user: 578 ↛ 579line 578 didn't jump to line 579 because the condition on line 578 was never true
579 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
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, "unban", note=request.admin_note, level=AdminActionLevel.high)
583 user.banned_at = None
584 return _user_to_details(session, user)
586 def ShadowUser(
587 self, request: admin_pb2.ShadowUserReq, context: CouchersContext, session: Session
588 ) -> admin_pb2.UserDetails:
589 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
590 if not user: 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.NOT_FOUND, "user_not_found")
592 if not request.admin_note.strip():
593 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
594 log_admin_action(session, context, user, "shadow", note=request.admin_note, level=AdminActionLevel.high)
595 user.shadowed_at = now()
596 # Bulk-shadow all UMS-governed content authored by this user so existing visible content is hidden too
597 bulk_set_user_content_visibility(
598 session=session,
599 user=user,
600 new_visibility=ModerationVisibility.shadowed,
601 moderator_user_id=context.user_id,
602 reason=f"User {user.id} shadowed: {request.admin_note}",
603 )
604 return _user_to_details(session, user)
606 def UnshadowUser(
607 self, request: admin_pb2.UnshadowUserReq, context: CouchersContext, session: Session
608 ) -> admin_pb2.UserDetails:
609 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
610 if not user: 610 ↛ 611line 610 didn't jump to line 611 because the condition on line 610 was never true
611 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
612 if not request.admin_note.strip(): 612 ↛ 613line 612 didn't jump to line 613 because the condition on line 612 was never true
613 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_cant_be_empty")
614 log_admin_action(session, context, user, "unshadow", note=request.admin_note, level=AdminActionLevel.high)
615 user.shadowed_at = None
616 # Sweep content shadowed by the cascade back to visible; leave hidden/unlisted content where moderators put it
617 bulk_set_user_content_visibility(
618 session=session,
619 user=user,
620 new_visibility=ModerationVisibility.visible,
621 moderator_user_id=context.user_id,
622 from_visibilities={ModerationVisibility.shadowed},
623 reason=f"User {user.id} unshadowed: {request.admin_note}",
624 )
625 return _user_to_details(session, user)
627 def AddAdminNote(
628 self, request: admin_pb2.AddAdminNoteReq, context: CouchersContext, session: Session
629 ) -> admin_pb2.UserDetails:
630 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
631 if not user: 631 ↛ 632line 631 didn't jump to line 632 because the condition on line 631 was never true
632 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
633 has_note = bool(request.admin_note.strip())
634 has_data = bool(request.data.strip())
635 if has_note == has_data:
636 context.abort_with_error_code(
637 grpc.StatusCode.INVALID_ARGUMENT, "admin:note_requires_exactly_one_of_note_or_data"
638 )
639 data = None
640 if has_data:
641 try:
642 data = json.loads(request.data)
643 except json.JSONDecodeError:
644 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:note_data_must_be_valid_json")
645 level = api2adminactionlevel.get(request.level, AdminActionLevel.normal)
646 log_admin_action(
647 session,
648 context,
649 user,
650 "note",
651 note=request.admin_note if has_note else None,
652 data=data,
653 level=level,
654 )
655 return _user_to_details(session, user)
657 def GetContentReport(
658 self, request: admin_pb2.GetContentReportReq, context: CouchersContext, session: Session
659 ) -> admin_pb2.GetContentReportRes:
660 content_report = session.execute(
661 select(ContentReport).where(ContentReport.id == request.content_report_id)
662 ).scalar_one_or_none()
663 if not content_report:
664 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:content_report_not_found")
665 return admin_pb2.GetContentReportRes(
666 content_report=_content_report_to_pb(content_report),
667 )
669 def GetContentReportsForAuthor(
670 self, request: admin_pb2.GetContentReportsForAuthorReq, context: CouchersContext, session: Session
671 ) -> admin_pb2.GetContentReportsForAuthorRes:
672 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
673 if not user: 673 ↛ 674line 673 didn't jump to line 674 because the condition on line 673 was never true
674 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
675 content_reports = (
676 session.execute(
677 select(ContentReport).where(ContentReport.author_user_id == user.id).order_by(ContentReport.id.desc())
678 )
679 .scalars()
680 .all()
681 )
682 return admin_pb2.GetContentReportsForAuthorRes(
683 content_reports=[_content_report_to_pb(content_report) for content_report in content_reports],
684 )
686 def SendModNote(
687 self, request: admin_pb2.SendModNoteReq, context: CouchersContext, session: Session
688 ) -> admin_pb2.UserDetails:
689 if request.notification not in _MOD_NOTE_NOTIFY_LOG_LABELS:
690 context.abort_with_error_code(
691 grpc.StatusCode.INVALID_ARGUMENT, "admin:mod_note_notification_must_be_specified"
692 )
693 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
694 if not user: 694 ↛ 695line 694 didn't jump to line 695 because the condition on line 694 was never true
695 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
696 session.add(
697 ModNote(
698 user_id=user.id,
699 internal_id=request.internal_id,
700 creator_user_id=context.user_id,
701 note_content=request.content,
702 )
703 )
704 session.flush()
705 log_admin_action(
706 session,
707 context,
708 user,
709 "send_mod_note",
710 note=f"Notify user: {_MOD_NOTE_NOTIFY_LOG_LABELS[request.notification]}\n\n{request.content}",
711 )
713 if request.notification != admin_pb2.MOD_NOTE_NOTIFICATION_NONE:
714 notify(
715 session,
716 user_id=user.id,
717 topic_action=NotificationTopicAction.modnote__create,
718 key="",
719 data=notification_data_pb2.ModNoteCreate(
720 markdown_text=(
721 request.content
722 if request.notification == admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT
723 else ""
724 ),
725 ),
726 )
728 return _user_to_details(session, user)
730 def MarkUserNeedsLocationUpdate(
731 self, request: admin_pb2.MarkUserNeedsLocationUpdateReq, context: CouchersContext, session: Session
732 ) -> admin_pb2.UserDetails:
733 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
734 if not user: 734 ↛ 735line 734 didn't jump to line 735 because the condition on line 734 was never true
735 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
736 user.needs_to_update_location = True
737 log_admin_action(
738 session, context, user, "mark_needs_location_update", note="Marked user as needing location update"
739 )
740 return _user_to_details(session, user)
742 def DeleteUser(
743 self, request: admin_pb2.DeleteUserReq, context: CouchersContext, session: Session
744 ) -> admin_pb2.UserDetails:
745 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
746 if not user: 746 ↛ 747line 746 didn't jump to line 747 because the condition on line 746 was never true
747 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
748 user.deleted_at = now()
749 log_admin_action(session, context, user, "delete_user", level=AdminActionLevel.high)
750 return _user_to_details(session, user)
752 def RecoverDeletedUser(
753 self, request: admin_pb2.RecoverDeletedUserReq, context: CouchersContext, session: Session
754 ) -> admin_pb2.UserDetails:
755 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
756 if not user: 756 ↛ 757line 756 didn't jump to line 757 because the condition on line 756 was never true
757 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
758 user.deleted_at = None
759 user.undelete_token = None
760 user.undelete_until = None
761 log_admin_action(session, context, user, "recover_user", level=AdminActionLevel.high)
762 return _user_to_details(session, user)
764 def CreateApiKey(
765 self, request: admin_pb2.CreateApiKeyReq, context: CouchersContext, session: Session
766 ) -> admin_pb2.CreateApiKeyRes:
767 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
768 if not user: 768 ↛ 769line 768 didn't jump to line 769 because the condition on line 768 was never true
769 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
770 token, expiry = create_session(
771 context, session, user, long_lived=True, is_api_key=True, duration=timedelta(days=365), set_cookie=False
772 )
773 log_admin_action(session, context, user, "create_api_key")
775 notify(
776 session,
777 user_id=user.id,
778 topic_action=NotificationTopicAction.api_key__create,
779 key="",
780 data=notification_data_pb2.ApiKeyCreate(
781 api_key=token,
782 expiry=Timestamp_from_datetime(expiry),
783 ),
784 )
786 return admin_pb2.CreateApiKeyRes()
788 def GetChats(
789 self, request: admin_pb2.GetChatsReq, context: CouchersContext, session: Session
790 ) -> admin_pb2.GetChatsRes:
791 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
792 if not user: 792 ↛ 793line 792 didn't jump to line 793 because the condition on line 792 was never true
793 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
795 get_chat_user_info = _make_chat_user_info_getter(session)
797 def message_to_pb(message: Message) -> admin_pb2.ChatMessage:
798 return admin_pb2.ChatMessage(
799 message_id=message.id,
800 author=get_chat_user_info(message.author_id),
801 time=Timestamp_from_datetime(message.time),
802 message_type=message.message_type.name if message.message_type else "",
803 text=message.text or "",
804 host_request_status_target=(
805 message.host_request_status_target.name if message.host_request_status_target else ""
806 ),
807 target=get_chat_user_info(message.target_id) if message.target_id else None,
808 )
810 def get_messages_for_conversation(conversation_id: int) -> list[admin_pb2.ChatMessage]:
811 messages = (
812 session.execute(
813 select(Message).where(Message.conversation_id == conversation_id).order_by(Message.id.asc())
814 )
815 .scalars()
816 .all()
817 )
818 return [message_to_pb(msg) for msg in messages]
820 def get_host_request_pb(host_request: HostRequest) -> admin_pb2.AdminHostRequest:
821 return admin_pb2.AdminHostRequest(
822 host_request_id=host_request.conversation_id,
823 surfer=get_chat_user_info(host_request.surfer_user_id),
824 host=get_chat_user_info(host_request.host_user_id),
825 status=host_request.status.name if host_request.status else "",
826 from_date=date_to_api(host_request.from_date),
827 to_date=date_to_api(host_request.to_date),
828 created=Timestamp_from_datetime(host_request.conversation.created),
829 messages=get_messages_for_conversation(host_request.conversation_id),
830 )
832 def get_group_chat_pb(group_chat: GroupChat) -> admin_pb2.AdminGroupChat:
833 subs = (
834 session.execute(
835 select(GroupChatSubscription)
836 .where(GroupChatSubscription.group_chat_id == group_chat.conversation_id)
837 .order_by(GroupChatSubscription.joined.asc())
838 )
839 .scalars()
840 .all()
841 )
842 members = [
843 admin_pb2.GroupChatMember(
844 user=get_chat_user_info(sub.user_id),
845 joined=Timestamp_from_datetime(sub.joined),
846 left=Timestamp_from_datetime(sub.left) if sub.left else None,
847 role=sub.role.name if sub.role else "",
848 )
849 for sub in subs
850 ]
851 return admin_pb2.AdminGroupChat(
852 group_chat_id=group_chat.conversation_id,
853 title=group_chat.title or "",
854 is_dm=group_chat.is_dm,
855 creator=get_chat_user_info(group_chat.creator_id),
856 members=members,
857 messages=get_messages_for_conversation(group_chat.conversation_id),
858 )
860 # Get all host requests for the user
861 host_requests = (
862 session.execute(
863 select(HostRequest)
864 .where(or_(HostRequest.recipient_user_id == user.id, HostRequest.initiator_user_id == user.id))
865 .order_by(HostRequest.conversation_id.desc())
866 )
867 .scalars()
868 .all()
869 )
871 # Get all group chats for the user
872 group_chat_ids = (
873 session.execute(
874 select(GroupChatSubscription.group_chat_id)
875 .where(GroupChatSubscription.user_id == user.id)
876 .order_by(GroupChatSubscription.joined.desc())
877 )
878 .scalars()
879 .all()
880 )
881 group_chats = (
882 session.execute(select(GroupChat).where(GroupChat.conversation_id.in_(group_chat_ids))).scalars().all()
883 )
885 # Build protobuf objects, then sort by latest message time (most recent first)
886 host_request_pbs = [get_host_request_pb(hr) for hr in host_requests]
887 host_request_pbs.sort(key=lambda hr: hr.messages[-1].time.seconds if hr.messages else 0, reverse=True)
889 group_chat_pbs = [get_group_chat_pb(gc) for gc in group_chats]
890 group_chat_pbs.sort(key=lambda gc: gc.messages[-1].time.seconds if gc.messages else 0, reverse=True)
892 return admin_pb2.GetChatsRes(
893 user=get_chat_user_info(user.id),
894 host_requests=host_request_pbs,
895 group_chats=group_chat_pbs,
896 )
898 def DeleteEvent(
899 self, request: admin_pb2.DeleteEventReq, context: CouchersContext, session: Session
900 ) -> empty_pb2.Empty:
901 res = session.execute(
902 select(Event, EventOccurrence)
903 .where(EventOccurrence.id == request.event_id)
904 .where(EventOccurrence.event_id == Event.id)
905 .where(~EventOccurrence.is_deleted)
906 ).one_or_none()
908 if not res: 908 ↛ 909line 908 didn't jump to line 909 because the condition on line 908 was never true
909 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "event_not_found")
911 event, occurrence = res
913 occurrence.is_deleted = True
915 queue_job(
916 session,
917 job=generate_event_delete_notifications,
918 payload=jobs_pb2.GenerateEventDeleteNotificationsPayload(
919 occurrence_id=occurrence.id,
920 ),
921 )
923 return empty_pb2.Empty()
925 def ListUserIds(
926 self, request: admin_pb2.ListUserIdsReq, context: CouchersContext, session: Session
927 ) -> admin_pb2.ListUserIdsRes:
928 start_date = to_aware_datetime(request.start_time)
929 end_date = to_aware_datetime(request.end_time)
931 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
932 next_user_id = int(request.page_token) if request.page_token else 0
934 user_ids = (
935 session.execute(
936 select(User.id)
937 .where(or_(User.id <= next_user_id, to_bool(next_user_id == 0)))
938 .where(User.joined >= start_date)
939 .where(User.joined <= end_date)
940 .order_by(User.id.desc())
941 .limit(page_size + 1)
942 )
943 .scalars()
944 .all()
945 )
947 return admin_pb2.ListUserIdsRes(
948 user_ids=user_ids[:page_size],
949 next_page_token=str(user_ids[-1]) if len(user_ids) > page_size else None,
950 )
952 def EditReferenceText(
953 self, request: admin_pb2.EditReferenceTextReq, context: CouchersContext, session: Session
954 ) -> empty_pb2.Empty:
955 reference = session.execute(select(Reference).where(Reference.id == request.reference_id)).scalar_one_or_none()
957 if reference is None: 957 ↛ 958line 957 didn't jump to line 958 because the condition on line 957 was never true
958 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:reference_not_found")
960 if not request.new_text.strip(): 960 ↛ 961line 960 didn't jump to line 961 because the condition on line 960 was never true
961 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "reference_no_text")
963 reference.text = request.new_text.strip()
964 # Log action against the reference author
965 author = session.execute(select(User).where(User.id == reference.from_user_id)).scalar_one()
966 log_admin_action(session, context, author, "edit_reference", note=f"Edited reference {reference.id}")
967 return empty_pb2.Empty()
969 def DeleteReference(
970 self, request: admin_pb2.DeleteReferenceReq, context: CouchersContext, session: Session
971 ) -> empty_pb2.Empty:
972 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:deletereference_deprecated_use_ums")
974 def GetUserReferences(
975 self, request: admin_pb2.GetUserReferencesReq, context: CouchersContext, session: Session
976 ) -> admin_pb2.GetUserReferencesRes:
977 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
978 if not user:
979 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
981 get_chat_user_info = _make_chat_user_info_getter(session)
983 references_from = (
984 session.execute(
985 select(Reference)
986 .where(Reference.from_user_id == user.id)
987 .options(selectinload(Reference.host_request), selectinload(Reference.moderation_state))
988 .order_by(Reference.id.desc())
989 )
990 .scalars()
991 .all()
992 )
994 references_to = (
995 session.execute(
996 select(Reference)
997 .where(Reference.to_user_id == user.id)
998 .options(selectinload(Reference.host_request), selectinload(Reference.moderation_state))
999 .order_by(Reference.id.desc())
1000 )
1001 .scalars()
1002 .all()
1003 )
1005 return admin_pb2.GetUserReferencesRes(
1006 references_from=[_reference_to_pb(ref, get_chat_user_info) for ref in references_from],
1007 references_to=[_reference_to_pb(ref, get_chat_user_info) for ref in references_to],
1008 )
1010 def GetFriendRequests(
1011 self, request: admin_pb2.GetFriendRequestsReq, context: CouchersContext, session: Session
1012 ) -> admin_pb2.GetFriendRequestsRes:
1013 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1014 if not user:
1015 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1017 get_chat_user_info = _make_chat_user_info_getter(session)
1019 def friend_request_to_pb(rel: FriendRelationship) -> admin_pb2.AdminFriendRequest:
1020 return admin_pb2.AdminFriendRequest(
1021 friend_request_id=rel.id,
1022 from_user=get_chat_user_info(rel.from_user_id),
1023 to_user=get_chat_user_info(rel.to_user_id),
1024 status=rel.status.name if rel.status else "",
1025 time_sent=Timestamp_from_datetime(rel.time_sent),
1026 time_responded=Timestamp_from_datetime(rel.time_responded) if rel.time_responded else None,
1027 moderation_visibility=not_none(rel.moderation_state.visibility).name,
1028 )
1030 sent = (
1031 session.execute(
1032 select(FriendRelationship)
1033 .where(FriendRelationship.from_user_id == user.id)
1034 .order_by(FriendRelationship.id.desc())
1035 )
1036 .scalars()
1037 .all()
1038 )
1040 received = (
1041 session.execute(
1042 select(FriendRelationship)
1043 .where(FriendRelationship.to_user_id == user.id)
1044 .order_by(FriendRelationship.id.desc())
1045 )
1046 .scalars()
1047 .all()
1048 )
1050 return admin_pb2.GetFriendRequestsRes(
1051 sent=[friend_request_to_pb(rel) for rel in sent],
1052 received=[friend_request_to_pb(rel) for rel in received],
1053 )
1055 def GetUserBlocks(
1056 self, request: admin_pb2.GetUserBlocksReq, context: CouchersContext, session: Session
1057 ) -> admin_pb2.GetUserBlocksRes:
1058 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1059 if not user:
1060 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1062 def user_block_to_pb(block: UserBlock, other_user: User) -> admin_pb2.AdminUserBlock:
1063 return admin_pb2.AdminUserBlock(
1064 user=admin_pb2.ChatUserInfo(
1065 user_id=other_user.id,
1066 username=other_user.username,
1067 name=other_user.name,
1068 birthdate=date_to_api(other_user.birthdate),
1069 gender=other_user.gender,
1070 ),
1071 time_blocked=Timestamp_from_datetime(block.time_blocked),
1072 )
1074 blocked_user = aliased(User)
1075 blocked_users = session.execute(
1076 select(UserBlock, blocked_user)
1077 .join(blocked_user, UserBlock.blocked_user_id == blocked_user.id)
1078 .where(UserBlock.blocking_user_id == user.id)
1079 .order_by(UserBlock.time_blocked.desc(), UserBlock.id.desc())
1080 ).all()
1082 blocking_user = aliased(User)
1083 blocking_users = session.execute(
1084 select(UserBlock, blocking_user)
1085 .join(blocking_user, UserBlock.blocking_user_id == blocking_user.id)
1086 .where(UserBlock.blocked_user_id == user.id)
1087 .order_by(UserBlock.time_blocked.desc(), UserBlock.id.desc())
1088 ).all()
1090 return admin_pb2.GetUserBlocksRes(
1091 blocked_users=[user_block_to_pb(block, other_user) for block, other_user in blocked_users],
1092 blocking_users=[user_block_to_pb(block, other_user) for block, other_user in blocking_users],
1093 )
1095 def GetNonvisibleUserAccessLog(
1096 self, request: admin_pb2.GetNonvisibleUserAccessLogReq, context: CouchersContext, session: Session
1097 ) -> admin_pb2.GetNonvisibleUserAccessLogRes:
1098 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1099 if not user: 1099 ↛ 1100line 1099 didn't jump to line 1100 because the condition on line 1099 was never true
1100 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1102 actor = aliased(User)
1103 rows = session.execute(
1104 select(NonvisibleUserAccess, actor.username)
1105 .outerjoin(actor, NonvisibleUserAccess.actor_user_id == actor.id)
1106 .where(NonvisibleUserAccess.target_user_id == user.id)
1107 .order_by(NonvisibleUserAccess.time.desc())
1108 .limit(MAX_PAGINATION_LENGTH)
1109 ).all()
1111 return admin_pb2.GetNonvisibleUserAccessLogRes(
1112 entries=[
1113 admin_pb2.NonvisibleUserAccessLogEntry(
1114 time=Timestamp_from_datetime(access.time),
1115 access_type=nonvisibleuseraccesstype2api[access.access_type],
1116 target_state=nonvisibleuserstate2api[access.target_state],
1117 target_user_id=access.target_user_id,
1118 actor_user_id=Int64Value(value=access.actor_user_id) if access.actor_user_id is not None else None,
1119 actor_username=actor_username or "",
1120 ip_address=access.ip_address or "",
1121 user_agent=access.user_agent or "",
1122 sofa=access.sofa or "",
1123 )
1124 for access, actor_username in rows
1125 ]
1126 )
1128 def EditDiscussion(
1129 self, request: admin_pb2.EditDiscussionReq, context: CouchersContext, session: Session
1130 ) -> empty_pb2.Empty:
1131 discussion = session.execute(
1132 select(Discussion).where(Discussion.id == request.discussion_id)
1133 ).scalar_one_or_none()
1134 if not discussion:
1135 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found")
1136 if request.new_title:
1137 discussion.title = request.new_title.strip()
1138 if request.new_content:
1139 discussion.content = request.new_content.strip()
1140 return empty_pb2.Empty()
1142 def DeleteDiscussion(
1143 self, request: admin_pb2.AdminDeleteDiscussionReq, context: CouchersContext, session: Session
1144 ) -> empty_pb2.Empty:
1145 discussion = session.execute(
1146 select(Discussion).where(Discussion.id == request.discussion_id)
1147 ).scalar_one_or_none()
1148 if not discussion: 1148 ↛ 1149line 1148 didn't jump to line 1149 because the condition on line 1148 was never true
1149 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "discussion_not_found")
1150 if discussion.deleted is not None: 1150 ↛ 1151line 1150 didn't jump to line 1151 because the condition on line 1150 was never true
1151 return empty_pb2.Empty()
1152 session.add(
1153 DiscussionVersion(
1154 discussion_id=discussion.id,
1155 editor_user_id=context.user_id,
1156 change_type=ContentChangeType.delete,
1157 old_title=discussion.title,
1158 new_title=None,
1159 old_content=discussion.content,
1160 new_content=None,
1161 )
1162 )
1163 discussion.deleted = now()
1164 return empty_pb2.Empty()
1166 def EditReply(self, request: admin_pb2.EditReplyReq, context: CouchersContext, session: Session) -> empty_pb2.Empty:
1167 database_id, depth = unpack_thread_id(request.reply_id)
1168 if depth == 1:
1169 obj: Comment | Reply | None = session.execute(
1170 select(Comment).where(Comment.id == database_id)
1171 ).scalar_one_or_none()
1172 elif depth == 2:
1173 obj = session.execute(select(Reply).where(Reply.id == database_id)).scalar_one_or_none()
1174 else:
1175 obj = None
1177 if not obj:
1178 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:object_not_found")
1179 old_content = obj.content
1180 new_content = request.new_content.strip()
1181 if depth == 1:
1182 session.add(
1183 CommentVersion(
1184 comment_id=database_id,
1185 editor_user_id=context.user_id,
1186 change_type=ContentChangeType.edit,
1187 old_content=old_content,
1188 new_content=new_content,
1189 )
1190 )
1191 else:
1192 session.add(
1193 ReplyVersion(
1194 reply_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 obj.content = new_content
1202 return empty_pb2.Empty()
1204 def AddUsersToModerationUserList(
1205 self, request: admin_pb2.AddUsersToModerationUserListReq, context: CouchersContext, session: Session
1206 ) -> admin_pb2.AddUsersToModerationUserListRes:
1207 """Add multiple users to a moderation user list. If no moderation list is provided, a new one is created.
1208 Id of the moderation list is returned."""
1209 req_users = request.users
1210 users = []
1212 for req_user in req_users:
1213 user = session.execute(select(User).where(username_or_email_or_id(req_user))).scalar_one_or_none()
1214 if not user:
1215 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1216 users.append(user)
1218 if request.moderation_list_id:
1219 moderation_user_list = session.get(ModerationUserList, request.moderation_list_id)
1220 if not moderation_user_list:
1221 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:moderation_user_list_not_found")
1222 # Create a new moderation user list if no one is provided
1223 else:
1224 moderation_user_list = ModerationUserList()
1225 session.add(moderation_user_list)
1226 session.flush()
1228 # Add users to the moderation list only if not already in it
1229 for user in users:
1230 if user not in moderation_user_list.users: 1230 ↛ 1232line 1230 didn't jump to line 1232 because the condition on line 1230 was always true
1231 moderation_user_list.users.append(user)
1232 log_admin_action(session, context, user, "add_to_moderation_list")
1234 return admin_pb2.AddUsersToModerationUserListRes(moderation_list_id=moderation_user_list.id)
1236 def ListModerationUserLists(
1237 self, request: admin_pb2.ListModerationUserListsReq, context: CouchersContext, session: Session
1238 ) -> admin_pb2.ListModerationUserListsRes:
1239 """Lists all moderation user lists for a user."""
1240 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1241 if not user: 1241 ↛ 1242line 1241 didn't jump to line 1242 because the condition on line 1241 was never true
1242 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1244 moderation_lists = [
1245 admin_pb2.ModerationList(
1246 moderation_list_id=ml.id,
1247 members=[_user_to_details(session, u) for u in ml.users],
1248 )
1249 for ml in user.moderation_user_lists
1250 ]
1251 return admin_pb2.ListModerationUserListsRes(moderation_lists=moderation_lists)
1253 def RemoveUserFromModerationUserList(
1254 self, request: admin_pb2.RemoveUserFromModerationUserListReq, context: CouchersContext, session: Session
1255 ) -> empty_pb2.Empty:
1256 """Removes a user from a provided moderation user list."""
1257 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1258 if not user:
1259 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1260 if not request.moderation_list_id:
1261 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:missing_moderation_user_list_id")
1263 moderation_user_list = session.get(ModerationUserList, request.moderation_list_id)
1264 if not moderation_user_list: 1264 ↛ 1265line 1264 didn't jump to line 1265 because the condition on line 1264 was never true
1265 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:moderation_user_list_not_found")
1266 if user not in moderation_user_list.users:
1267 context.abort_with_error_code(
1268 grpc.StatusCode.FAILED_PRECONDITION, "admin:user_not_in_the_moderation_user_list"
1269 )
1271 moderation_user_list.users.remove(user)
1272 log_admin_action(session, context, user, "remove_from_moderation_list")
1274 if len(moderation_user_list.users) == 0:
1275 session.delete(moderation_user_list)
1277 return empty_pb2.Empty()
1279 def CreateAccountDeletionLink(
1280 self, request: admin_pb2.CreateAccountDeletionLinkReq, context: CouchersContext, session: Session
1281 ) -> admin_pb2.CreateAccountDeletionLinkRes:
1282 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1283 if not user: 1283 ↛ 1284line 1283 didn't jump to line 1284 because the condition on line 1283 was never true
1284 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1285 token = AccountDeletionToken(token=urlsafe_secure_token(), user_id=user.id, expiry=now() + timedelta(hours=2))
1286 session.add(token)
1287 log_admin_action(session, context, user, "create_account_deletion_link", level=AdminActionLevel.high)
1288 return admin_pb2.CreateAccountDeletionLinkRes(
1289 account_deletion_confirm_url=urls.delete_account_link(account_deletion_token=token.token)
1290 )
1292 def AccessStats(
1293 self, request: admin_pb2.AccessStatsReq, context: CouchersContext, session: Session
1294 ) -> admin_pb2.AccessStatsRes:
1295 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1296 if not user: 1296 ↛ 1297line 1296 didn't jump to line 1297 because the condition on line 1296 was never true
1297 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
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 = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1350 if not user:
1351 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1353 if request.HasField("last_donated"):
1354 user.last_donated = to_aware_datetime(request.last_donated)
1355 else:
1356 user.last_donated = None
1358 log_admin_action(session, context, user, "set_last_donated")
1359 return _user_to_details(session, user)
1361 def CreateAdminTag(
1362 self, request: admin_pb2.CreateAdminTagReq, context: CouchersContext, session: Session
1363 ) -> admin_pb2.AdminTagInfo:
1364 if not request.tag.strip():
1365 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:tag_cant_be_empty")
1366 existing = session.execute(select(AdminTag).where(AdminTag.tag == request.tag.strip())).scalar_one_or_none()
1367 if existing:
1368 context.abort_with_error_code(grpc.StatusCode.ALREADY_EXISTS, "admin:tag_already_exists")
1369 admin_tag = AdminTag(tag=request.tag.strip())
1370 session.add(admin_tag)
1371 session.flush()
1372 return admin_pb2.AdminTagInfo(admin_tag_id=admin_tag.id, tag=admin_tag.tag)
1374 def ListAdminTags(
1375 self, request: admin_pb2.ListAdminTagsReq, context: CouchersContext, session: Session
1376 ) -> admin_pb2.ListAdminTagsRes:
1377 tags = session.execute(select(AdminTag).order_by(AdminTag.tag)).scalars().all()
1378 return admin_pb2.ListAdminTagsRes(
1379 tags=[admin_pb2.AdminTagInfo(admin_tag_id=tag.id, tag=tag.tag) for tag in tags]
1380 )
1382 def AddAdminTagToUser(
1383 self, request: admin_pb2.AddAdminTagToUserReq, context: CouchersContext, session: Session
1384 ) -> admin_pb2.UserDetails:
1385 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1386 if not user: 1386 ↛ 1387line 1386 didn't jump to line 1387 because the condition on line 1386 was never true
1387 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1388 admin_tag = session.execute(select(AdminTag).where(AdminTag.tag == request.tag)).scalar_one_or_none()
1389 if not admin_tag:
1390 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:tag_not_found")
1391 existing = session.execute(
1392 select(UserAdminTag).where(UserAdminTag.user_id == user.id, UserAdminTag.admin_tag_id == admin_tag.id)
1393 ).scalar_one_or_none()
1394 if existing:
1395 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_already_has_admin_tag")
1396 session.add(UserAdminTag(user_id=user.id, admin_tag_id=admin_tag.id))
1397 session.flush()
1398 log_admin_action(session, context, user, "add_tag", tag=request.tag)
1399 return _user_to_details(session, user)
1401 def RemoveAdminTagFromUser(
1402 self, request: admin_pb2.RemoveAdminTagFromUserReq, context: CouchersContext, session: Session
1403 ) -> admin_pb2.UserDetails:
1404 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1405 if not user: 1405 ↛ 1406line 1405 didn't jump to line 1406 because the condition on line 1405 was never true
1406 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1407 admin_tag = session.execute(select(AdminTag).where(AdminTag.tag == request.tag)).scalar_one_or_none()
1408 if not admin_tag: 1408 ↛ 1409line 1408 didn't jump to line 1409 because the condition on line 1408 was never true
1409 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:tag_not_found")
1410 user_admin_tag = session.execute(
1411 select(UserAdminTag).where(UserAdminTag.user_id == user.id, UserAdminTag.admin_tag_id == admin_tag.id)
1412 ).scalar_one_or_none()
1413 if not user_admin_tag:
1414 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:user_does_not_have_admin_tag")
1415 session.delete(user_admin_tag)
1416 session.flush()
1417 log_admin_action(session, context, user, "remove_tag", tag=request.tag)
1418 return _user_to_details(session, user)
1420 def SetModScore(
1421 self, request: admin_pb2.SetModScoreReq, context: CouchersContext, session: Session
1422 ) -> admin_pb2.UserDetails:
1423 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1424 if not user:
1425 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1426 user.mod_score = request.mod_score
1427 log_admin_action(session, context, user, "set_mod_score", note=f"mod_score={request.mod_score}")
1428 return _user_to_details(session, user)
1430 def ListAdminActions(
1431 self, request: admin_pb2.ListAdminActionsReq, context: CouchersContext, session: Session
1432 ) -> admin_pb2.ListAdminActionsRes:
1433 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
1435 admin_user = aliased(User)
1436 target_user = aliased(User)
1438 statement = (
1439 select(AdminAction, admin_user.username, target_user.username)
1440 .join(admin_user, AdminAction.admin_user_id == admin_user.id)
1441 .join(target_user, AdminAction.target_user_id == target_user.id)
1442 )
1444 if request.admin_user_id:
1445 statement = statement.where(AdminAction.admin_user_id == request.admin_user_id)
1446 if request.target_user_id:
1447 statement = statement.where(AdminAction.target_user_id == request.target_user_id)
1448 if request.page_token:
1449 statement = statement.where(AdminAction.id < int(request.page_token))
1451 statement = statement.order_by(AdminAction.id.desc()).limit(page_size + 1)
1453 rows = session.execute(statement).all()
1455 action_pbs = [
1456 admin_pb2.AdminActionLog(
1457 admin_action_id=action.id,
1458 created=Timestamp_from_datetime(action.created),
1459 admin_user_id=action.admin_user_id,
1460 admin_username=admin_username,
1461 action_type=action.action_type,
1462 level=adminactionlevel2api[action.level],
1463 note=action.note or "",
1464 data=json.dumps(action.data) if action.data is not None else "",
1465 tag=action.tag or "",
1466 target_user_id=action.target_user_id,
1467 target_username=target_username,
1468 )
1469 for action, admin_username, target_username in rows[:page_size]
1470 ]
1472 return admin_pb2.ListAdminActionsRes(
1473 admin_actions=action_pbs,
1474 next_page_token=str(rows[page_size - 1][0].id) if len(rows) > page_size else None,
1475 )
1477 def ListUserUploads(
1478 self, request: admin_pb2.ListUserUploadsReq, context: CouchersContext, session: Session
1479 ) -> admin_pb2.ListUserUploadsRes:
1480 user = session.execute(select(User).where(username_or_email_or_id(request.user))).scalar_one_or_none()
1481 if not user:
1482 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found")
1484 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH)
1486 statement = (
1487 select(Upload)
1488 .where(Upload.creator_user_id == user.id)
1489 .options(
1490 undefer(Upload.metadata_parsed),
1491 undefer(Upload.metadata_parse_error),
1492 undefer(Upload.original_filename),
1493 undefer(Upload.original_format),
1494 undefer(Upload.original_size),
1495 undefer(Upload.original_width),
1496 undefer(Upload.original_height),
1497 )
1498 )
1499 if request.page_token:
1500 cursor_created = session.execute(
1501 select(Upload.created).where(Upload.key == request.page_token)
1502 ).scalar_one()
1503 statement = statement.where(tuple_(Upload.created, Upload.key) < (cursor_created, request.page_token))
1505 uploads = (
1506 session.execute(statement.order_by(Upload.created.desc(), Upload.key.desc()).limit(page_size + 1))
1507 .scalars()
1508 .all()
1509 )
1511 page = uploads[:page_size]
1512 uses_by_key = get_upload_uses_for_keys(session, [upload.key for upload in page])
1514 return admin_pb2.ListUserUploadsRes(
1515 uploads=[
1516 admin_pb2.UserUpload(
1517 key=upload.key,
1518 filename=upload.filename,
1519 full_url=upload.full_url,
1520 thumbnail_url=upload.thumbnail_url,
1521 credit=upload.credit or "",
1522 created=Timestamp_from_datetime(upload.created),
1523 metadata=_upload_metadata_to_pb(upload),
1524 uses=[
1525 admin_pb2.UploadUse(
1526 type=uploadusetype2api[use.use_type],
1527 is_current=use.is_current,
1528 user_id=use.user_id,
1529 event_id=use.event_id,
1530 page_id=use.page_id,
1531 url=use.url,
1532 )
1533 for use in uses_by_key.get(upload.key, [])
1534 ],
1535 )
1536 for upload in page
1537 ],
1538 next_page_token=uploads[page_size - 1].key if len(uploads) > page_size else None,
1539 )
1541 def CreateOTAPackage(
1542 self, request: admin_pb2.CreateOTAPackageReq, context: CouchersContext, session: Session
1543 ) -> admin_pb2.OTAPackage:
1544 platform = api2otaplatform.get(request.platform)
1545 if platform is None: 1545 ↛ 1546line 1545 didn't jump to line 1546 because the condition on line 1545 was never true
1546 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_platform")
1548 if not request.version:
1549 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_version")
1551 existing = session.execute(
1552 select(OTAPackage.id).where(OTAPackage.platform == platform).where(OTAPackage.version == request.version)
1553 ).scalar_one_or_none()
1554 if existing is not None:
1555 context.abort_with_error_code(grpc.StatusCode.FAILED_PRECONDITION, "admin:ota_package_already_exists")
1557 # Read the keying/ordering fields out of the manifest we're about to serve, so the row can't
1558 # disagree with the bytes on the CDN.
1559 cdn_root = context.get_string_value("native_ota_cdn_root", "https://cdn.couchers.org/native/ota")
1560 _content_type, body = _fetch_signed_manifest(
1561 _native_ota_manifest_url(cdn_root=cdn_root, version=request.version, platform=platform.name)
1562 )
1563 manifest = _extract_ota_manifest(body)
1564 fingerprint = manifest.get("runtimeVersion") if manifest else None
1565 manifest_id = manifest.get("id") if manifest else None
1566 created_at_raw = manifest.get("createdAt") if manifest else None
1567 if (
1568 manifest is None
1569 or not isinstance(fingerprint, str)
1570 or not fingerprint
1571 or not isinstance(manifest_id, str)
1572 or not manifest_id
1573 or not isinstance(created_at_raw, str)
1574 or not created_at_raw
1575 ):
1576 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_manifest")
1577 try:
1578 manifest_created_at = datetime.fromisoformat(created_at_raw)
1579 except ValueError:
1580 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_manifest")
1581 if manifest_created_at.tzinfo is None: 1581 ↛ 1582line 1581 didn't jump to line 1582 because the condition on line 1581 was never true
1582 manifest_created_at = manifest_created_at.replace(tzinfo=UTC)
1584 package = OTAPackage(
1585 creator_user_id=context.user_id,
1586 platform=platform,
1587 fingerprint=fingerprint,
1588 version=request.version,
1589 manifest_created_at=manifest_created_at,
1590 manifest_id=manifest_id,
1591 )
1592 session.add(package)
1593 session.flush()
1595 return _ota_package_to_pb(package, _live_ota_package_ids(session))
1597 def ListOTAPackages(
1598 self, request: admin_pb2.ListOTAPackagesReq, context: CouchersContext, session: Session
1599 ) -> admin_pb2.ListOTAPackagesRes:
1600 statement = select(OTAPackage).order_by(OTAPackage.manifest_created_at.desc(), OTAPackage.id.desc())
1601 if request.platform != admin_pb2.OTA_PLATFORM_UNSPECIFIED:
1602 platform = api2otaplatform.get(request.platform)
1603 if platform is None: 1603 ↛ 1604line 1603 didn't jump to line 1604 because the condition on line 1603 was never true
1604 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:invalid_ota_platform")
1605 statement = statement.where(OTAPackage.platform == platform)
1606 if request.fingerprint: 1606 ↛ 1607line 1606 didn't jump to line 1607 because the condition on line 1606 was never true
1607 statement = statement.where(OTAPackage.fingerprint == request.fingerprint)
1608 if not request.include_banned:
1609 statement = statement.where(OTAPackage.banned_at.is_(None))
1611 packages = session.execute(statement).scalars().all()
1612 live_ids = _live_ota_package_ids(session)
1613 return admin_pb2.ListOTAPackagesRes(packages=[_ota_package_to_pb(package, live_ids) for package in packages])
1615 def BanOTAPackage(
1616 self, request: admin_pb2.BanOTAPackageReq, context: CouchersContext, session: Session
1617 ) -> admin_pb2.OTAPackage:
1618 # Bans are irreversible — to roll back an accidental ban, republish the bundle as a new
1619 # package — so a reason is required for the audit trail.
1620 if not request.reason.strip():
1621 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "admin:ota_ban_reason_required")
1623 package = session.execute(
1624 select(OTAPackage).where(OTAPackage.id == request.ota_package_id)
1625 ).scalar_one_or_none()
1626 if package is None:
1627 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "admin:ota_package_not_found")
1629 if package.banned_at is None: 1629 ↛ 1633line 1629 didn't jump to line 1633 because the condition on line 1629 was always true
1630 package.banned_at = now()
1631 package.banned_by_user_id = context.user_id
1632 package.banned_reason = request.reason
1633 session.flush()
1635 return _ota_package_to_pb(package, _live_ota_package_ids(session))