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