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