Corp/backend/communications/media_tickets.py
lizerk 2768d2f1b3
Some checks failed
Deploy production / deploy (push) Failing after 14m49s
Restore reliable chat realtime and media playback
2026-09-10 20:26:58 +03:00

78 lines
2.0 KiB
Python

"""Short-lived, scoped credentials for inline chat media playback."""
from __future__ import annotations
from dataclasses import dataclass
from django.conf import settings
from django.core import signing
MEDIA_TICKET_SALT = "eusrr.chat-media.ticket.v1"
DEFAULT_MEDIA_TICKET_MAX_AGE_SECONDS = 15 * 60
class ChatMediaTicketError(ValueError):
pass
@dataclass(frozen=True)
class ChatMediaTicketIdentity:
user_id: int
message_id: int
attachment_id: int
def media_ticket_max_age_seconds() -> int:
configured = getattr(
settings,
"CHAT_MEDIA_TICKET_MAX_AGE_SECONDS",
DEFAULT_MEDIA_TICKET_MAX_AGE_SECONDS,
)
try:
return max(30, min(int(configured), 60 * 60))
except (TypeError, ValueError):
return DEFAULT_MEDIA_TICKET_MAX_AGE_SECONDS
def issue_chat_media_ticket(
*, user_id: int, message_id: int, attachment_id: int
) -> str:
return signing.dumps(
{
"uid": int(user_id),
"mid": int(message_id),
"aid": int(attachment_id),
},
salt=MEDIA_TICKET_SALT,
compress=True,
)
def consume_chat_media_ticket(
raw_ticket: str,
*,
message_id: int,
attachment_id: int,
) -> ChatMediaTicketIdentity:
try:
payload = signing.loads(
raw_ticket,
salt=MEDIA_TICKET_SALT,
max_age=media_ticket_max_age_seconds(),
)
identity = ChatMediaTicketIdentity(
user_id=int(payload["uid"]),
message_id=int(payload["mid"]),
attachment_id=int(payload["aid"]),
)
except (signing.BadSignature, KeyError, TypeError, ValueError) as error:
raise ChatMediaTicketError("invalid_or_expired_ticket") from error
if identity.user_id <= 0:
raise ChatMediaTicketError("invalid_user")
if identity.message_id != int(message_id):
raise ChatMediaTicketError("message_mismatch")
if identity.attachment_id != int(attachment_id):
raise ChatMediaTicketError("attachment_mismatch")
return identity