mi
82 lines
2.4 KiB
Python
82 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import datetime
|
|
from typing import Annotated, Literal
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator
|
|
|
|
Checksum = Annotated[str, StringConstraints(pattern=r"^sha256:[0-9a-f]{64}$")]
|
|
|
|
|
|
class StrictModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", strict=True)
|
|
|
|
|
|
class Attachment(StrictModel):
|
|
attachment_id: UUID
|
|
quarantine_object_key: Annotated[str, StringConstraints(min_length=1, max_length=1024)]
|
|
quarantine_version_id: Annotated[str, StringConstraints(min_length=1, max_length=512)]
|
|
quarantine_etag: Annotated[str, StringConstraints(min_length=1, max_length=512)]
|
|
mime_type: Annotated[str, StringConstraints(min_length=1, max_length=127)]
|
|
size_bytes: int = Field(ge=1, le=5_242_880)
|
|
checksum: Checksum
|
|
|
|
@model_validator(mode="after")
|
|
def canonical_key(self) -> Attachment:
|
|
uuid = r"[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}"
|
|
pattern = rf"^quarantine/users/{uuid}/dialogs/{uuid}/{uuid}$"
|
|
if not self.quarantine_object_key.isascii() or not re.fullmatch(
|
|
pattern, self.quarantine_object_key
|
|
):
|
|
raise ValueError("quarantine_object_key is not canonical")
|
|
return self
|
|
|
|
|
|
class TextCheck(StrictModel):
|
|
message_id: UUID
|
|
content_kind: Literal["text"]
|
|
text: Annotated[str, StringConstraints(min_length=1, max_length=10_000)]
|
|
attachment: None = None
|
|
|
|
|
|
class FileCheck(StrictModel):
|
|
message_id: UUID
|
|
content_kind: Literal["file"]
|
|
text: Literal[""]
|
|
attachment: Attachment
|
|
|
|
|
|
CheckRequest = Annotated[TextCheck | FileCheck, Field(discriminator="content_kind")]
|
|
|
|
|
|
class Verdict(StrictModel):
|
|
verdict: Literal["allow", "deny"]
|
|
processing_mode: Literal["standard", "mock"]
|
|
config_version: int
|
|
rule_id: str
|
|
rules_version: str
|
|
reason_code: Literal["message_blocked"] | None = None
|
|
|
|
|
|
class Pending(StrictModel):
|
|
verdict: Literal["pending"] = "pending"
|
|
processing_mode: Literal["standard"] = "standard"
|
|
config_version: int
|
|
task_id: UUID
|
|
poll_after_ms: int = 2000
|
|
expires_at: datetime
|
|
rules_version: str
|
|
|
|
|
|
class ErrorBody(StrictModel):
|
|
code: str
|
|
message: str
|
|
request_id: str
|
|
details: dict[str, object] = Field(default_factory=dict)
|
|
|
|
|
|
class ErrorEnvelope(StrictModel):
|
|
error: ErrorBody
|