mi
252 lines
7.7 KiB
Python
252 lines
7.7 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import uuid
|
|
from base64 import urlsafe_b64decode, urlsafe_b64encode
|
|
from datetime import datetime
|
|
from enum import StrEnum
|
|
from typing import Annotated, Literal
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator
|
|
|
|
from app.chat_settings import CHAT_MESSAGE_TRANSPORT_MAX_LENGTH
|
|
|
|
|
|
class StrictModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class OtpSettingsResponse(StrictModel):
|
|
max_send_attempts_per_24h: int = Field(strict=True, gt=0)
|
|
min_seconds_between_attempts: int = Field(strict=True, ge=0)
|
|
max_verify_attempts: int = Field(strict=True, gt=0)
|
|
code_length: int = Field(strict=True, ge=4, le=10)
|
|
ttl_seconds: int = Field(strict=True, ge=60, le=900, multiple_of=60)
|
|
sms_order_timeout_ms: int = Field(strict=True, gt=0)
|
|
version: str = Field(min_length=1, max_length=64)
|
|
cache_ttl_seconds: int = Field(strict=True, gt=0)
|
|
|
|
|
|
class PublicAuthConfig(StrictModel):
|
|
phone_enabled: bool
|
|
password_enabled: bool
|
|
|
|
|
|
class PublicOperatorConfig(StrictModel):
|
|
call_phone: str = Field(min_length=1, max_length=32)
|
|
|
|
|
|
class PublicMessagesConfig(StrictModel):
|
|
max_text_length: int = Field(strict=True, ge=1, le=CHAT_MESSAGE_TRANSPORT_MAX_LENGTH)
|
|
|
|
|
|
class PublicConsentConfig(StrictModel):
|
|
required: bool
|
|
document_url: HttpUrl
|
|
version: str = Field(min_length=1, max_length=64)
|
|
|
|
|
|
class PublicPersonalDataConsentConfig(PublicConsentConfig):
|
|
privacy_policy_document_url: HttpUrl
|
|
|
|
|
|
class PublicConsentsConfig(StrictModel):
|
|
personal_data: PublicPersonalDataConsentConfig
|
|
user_agreement: PublicConsentConfig
|
|
marketing: PublicConsentConfig
|
|
|
|
|
|
class PublicAttachmentsConfig(StrictModel):
|
|
allowed_extensions: list[str]
|
|
allowed_mime_types: list[str]
|
|
max_size_mb: int = Field(strict=True, gt=0)
|
|
|
|
|
|
class PublicNotificationConfig(StrictModel):
|
|
carousel_autoplay_enabled: bool
|
|
carousel_autoplay_interval_ms: int = Field(strict=True, gt=0)
|
|
|
|
|
|
class PublicUxConfig(StrictModel):
|
|
idle_timeout_minutes: int = Field(strict=True, gt=0)
|
|
|
|
|
|
class MobileStoreUpdatePolicy(StrictModel):
|
|
enabled: bool
|
|
latest_build: int | None = Field(strict=True, ge=1)
|
|
minimum_build: int | None = Field(strict=True, ge=1)
|
|
latest_version: str | None = Field(min_length=1, max_length=64)
|
|
store_url: HttpUrl | None
|
|
release_notes: str | None = Field(max_length=4000)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_policy(self) -> "MobileStoreUpdatePolicy":
|
|
release_fields = (
|
|
self.latest_build,
|
|
self.minimum_build,
|
|
self.latest_version,
|
|
self.store_url,
|
|
)
|
|
if self.enabled and any(value is None for value in release_fields):
|
|
raise ValueError("enabled update policy requires all release fields")
|
|
if not self.enabled and any(value is not None for value in release_fields):
|
|
raise ValueError("disabled update policy must not expose release fields")
|
|
if (
|
|
self.minimum_build is not None
|
|
and self.latest_build is not None
|
|
and self.minimum_build > self.latest_build
|
|
):
|
|
raise ValueError("minimum_build must not exceed latest_build")
|
|
return self
|
|
|
|
|
|
class MobileUpdatePolicy(StrictModel):
|
|
google_play: MobileStoreUpdatePolicy
|
|
rustore: MobileStoreUpdatePolicy
|
|
app_store: MobileStoreUpdatePolicy
|
|
|
|
|
|
class PublicAppConfigResponse(StrictModel):
|
|
auth: PublicAuthConfig
|
|
operator: PublicOperatorConfig
|
|
messages: PublicMessagesConfig
|
|
consents: PublicConsentsConfig
|
|
attachments: PublicAttachmentsConfig
|
|
notification: PublicNotificationConfig
|
|
ux: PublicUxConfig
|
|
mobile_update: MobileUpdatePolicy
|
|
|
|
|
|
class Device(StrictModel):
|
|
platform: Literal["ios", "android", "web"]
|
|
app_version: str = Field(min_length=1, max_length=64)
|
|
device_id: str | None = Field(default=None, max_length=255)
|
|
|
|
|
|
class ConsentChoice(StrictModel):
|
|
accepted: bool
|
|
version: str = Field(min_length=1, max_length=64)
|
|
|
|
|
|
class ConsentSet(StrictModel):
|
|
personal_data: ConsentChoice
|
|
user_agreement: ConsentChoice
|
|
marketing: ConsentChoice
|
|
|
|
|
|
class BootstrapRequest(StrictModel):
|
|
consents: ConsentSet
|
|
device: Device
|
|
|
|
|
|
class ConsentsRequest(StrictModel):
|
|
consents: ConsentSet
|
|
|
|
|
|
class SessionStartRequest(StrictModel):
|
|
start_reason: Literal["first_launch", "cold_start", "idle_timeout"]
|
|
device: Device
|
|
|
|
|
|
class TextMessageRequest(StrictModel):
|
|
content_kind: Literal["text"]
|
|
text: str = Field(min_length=1, max_length=CHAT_MESSAGE_TRANSPORT_MAX_LENGTH)
|
|
|
|
|
|
class FileMessageRequest(StrictModel):
|
|
content_kind: Literal["file"]
|
|
attachment_id: uuid.UUID
|
|
checksum: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
|
|
|
|
|
|
MessageRequest = Annotated[
|
|
TextMessageRequest | FileMessageRequest, Field(discriminator="content_kind")
|
|
]
|
|
|
|
|
|
class AttachmentInitRequest(StrictModel):
|
|
file_name: str = Field(min_length=1, max_length=255)
|
|
mime_type: str = Field(min_length=1, max_length=128)
|
|
size_bytes: int = Field(gt=0)
|
|
|
|
|
|
class AttachmentCompleteRequest(StrictModel):
|
|
checksum: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
|
|
|
|
|
|
class OpenLinesFile(StrictModel):
|
|
name: str = Field(min_length=1, max_length=255)
|
|
mime_type: str = Field(min_length=1, max_length=128)
|
|
size_bytes: int = Field(gt=0)
|
|
download_url: HttpUrl
|
|
|
|
|
|
class OpenLinesMessage(StrictModel):
|
|
text: str = Field(default="", max_length=4000)
|
|
files: list[OpenLinesFile] = Field(default_factory=list, max_length=1)
|
|
|
|
@model_validator(mode="after")
|
|
def non_empty(self) -> "OpenLinesMessage":
|
|
if not self.text.strip() and not self.files:
|
|
raise ValueError("message must contain text or file")
|
|
return self
|
|
|
|
|
|
class OpenLinesInbox(StrictModel):
|
|
event_id: str = Field(min_length=1, max_length=255)
|
|
event_type: Literal["message.new", "dialog.closed"]
|
|
external_chat_id: uuid.UUID
|
|
bitrix_message_id: str | None = Field(default=None, max_length=255)
|
|
occurred_at: datetime
|
|
message: OpenLinesMessage | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def event_shape(self) -> "OpenLinesInbox":
|
|
if self.event_type == "message.new" and (
|
|
not self.bitrix_message_id or self.message is None
|
|
):
|
|
raise ValueError("message.new requires bitrix_message_id and message")
|
|
return self
|
|
|
|
|
|
class DialogStatus(StrEnum):
|
|
OPEN = "open"
|
|
WAITING_COMPANY = "waiting_for_company"
|
|
WAITING_CLIENT = "waiting_for_client"
|
|
CLOSED = "closed"
|
|
|
|
|
|
def canonical_fingerprint(
|
|
method: str, route: str, path_params: dict[str, str], body: object, user_id: uuid.UUID
|
|
) -> str:
|
|
value = {
|
|
"method": method.upper(),
|
|
"route": route,
|
|
"path": dict(sorted(path_params.items())),
|
|
"body": body,
|
|
"user_id": str(user_id),
|
|
}
|
|
return hashlib.sha256(
|
|
json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
|
|
).hexdigest()
|
|
|
|
|
|
def encode_cursor(data: dict[str, str], secret: bytes) -> str:
|
|
payload = json.dumps({"v": 1, **data}, sort_keys=True, separators=(",", ":")).encode()
|
|
signature = hmac.digest(secret, payload, "sha256")
|
|
return urlsafe_b64encode(payload + signature).decode().rstrip("=")
|
|
|
|
|
|
def decode_cursor(value: str, secret: bytes) -> dict[str, str]:
|
|
try:
|
|
raw = urlsafe_b64decode(value + "=" * (-len(value) % 4))
|
|
payload, signature = raw[:-32], raw[-32:]
|
|
if not hmac.compare_digest(signature, hmac.digest(secret, payload, "sha256")):
|
|
raise ValueError("invalid cursor")
|
|
decoded = json.loads(payload)
|
|
if decoded.pop("v") != 1:
|
|
raise ValueError("unsupported cursor")
|
|
return decoded
|
|
except (ValueError, KeyError, json.JSONDecodeError) as exc:
|
|
raise ValueError("invalid cursor") from exc
|