houseplan-card/scripts/support-relay/hp_relay/multipart.py

"""Строгий разбор multipart/form-data.

Строгий — значит «принимаем ровно то, что описано в §8.3 ТЗ, всё остальное
отвергаем». Публичный эндпоинт без общего секрета защищается схемой, размером и
частотой; парсер здесь — первая из трёх защит, поэтому он не прощает ничего:
ни лишних частей, ни повторов, ни отсутствующей границы, ни вложенного
multipart.
"""

from __future__ import annotations

from dataclasses import dataclass


class MultipartError(ValueError):
    """Тело не соответствует объявленной схеме."""


@dataclass(frozen=True)
class Part:
    name: str
    filename: str | None
    content_type: str
    body: bytes


def parse_content_type(header: str) -> str:
    """Возвращает boundary или бросает MultipartError."""
    if not header:
        raise MultipartError("missing content-type")
    pieces = [piece.strip() for piece in header.split(";")]
    if pieces[0].lower() != "multipart/form-data":
        raise MultipartError("content-type must be multipart/form-data")
    for piece in pieces[1:]:
        key, _, value = piece.partition("=")
        if key.strip().lower() != "boundary":
            continue
        value = value.strip()
        if value.startswith('"') and value.endswith('"') and len(value) >= 2:
            value = value[1:-1]
        if not value or len(value) > 70:
            raise MultipartError("bad boundary")
        return value
    raise MultipartError("missing boundary")


def _split_headers(chunk: bytes) -> tuple[dict[str, str], bytes]:
    head, sep, body = chunk.partition(b"\r\n\r\n")
    if not sep:
        raise MultipartError("part without headers")
    headers: dict[str, str] = {}
    for raw in head.split(b"\r\n"):
        if not raw:
            continue
        try:
            line = raw.decode("ascii")
        except UnicodeDecodeError as exc:
            raise MultipartError("non-ascii header") from exc
        key, _, value = line.partition(":")
        if not _:
            raise MultipartError("malformed header")
        key = key.strip().lower()
        if key in headers:
            raise MultipartError("duplicate header")
        headers[key] = value.strip()
    return headers, body


def _disposition(value: str) -> tuple[str, str | None]:
    pieces = [piece.strip() for piece in value.split(";")]
    if not pieces or pieces[0].lower() != "form-data":
        raise MultipartError("bad content-disposition")
    name: str | None = None
    filename: str | None = None
    for piece in pieces[1:]:
        key, _, raw = piece.partition("=")
        raw = raw.strip()
        if raw.startswith('"') and raw.endswith('"') and len(raw) >= 2:
            raw = raw[1:-1]
        key = key.strip().lower()
        if key == "name":
            name = raw
        elif key == "filename":
            filename = raw
    if not name:
        raise MultipartError("part without name")
    return name, filename


def parse(body: bytes, boundary: str, allowed: frozenset[str]) -> dict[str, Part]:
    """Разбирает тело и возвращает части по именам.

    Имя, которого нет в `allowed`, — ошибка, а не игнорируемое поле: клиент,
    приславший лишнюю часть, разговаривает не по этому контракту, и молча
    принять его запрос значит принять неизвестно что.
    """
    marker = b"--" + boundary.encode("ascii")
    if not body.startswith(marker):
        raise MultipartError("body does not start with boundary")
    rest = body[len(marker):]
    if rest.startswith(b"--"):
        raise MultipartError("empty body")
    if not rest.startswith(b"\r\n"):
        raise MultipartError("malformed preamble")
    # Открывающая граница снимается ДО разбиения: иначе первая часть вбирает
    # в себя весь остаток тела вместе с чужими заголовками.
    segments = rest[2:].split(b"\r\n" + marker)
    parts: dict[str, Part] = {}
    closed = False
    for index, segment in enumerate(segments):
        if index:
            if segment.startswith(b"--"):
                closed = True
                break
            if not segment.startswith(b"\r\n"):
                raise MultipartError("malformed boundary")
            segment = segment[2:]
        headers, raw = _split_headers(segment)
        name, filename = _disposition(headers.get("content-disposition", ""))
        if name not in allowed:
            raise MultipartError(f"unexpected part: {name}")
        if name in parts:
            raise MultipartError(f"duplicate part: {name}")
        content_type = headers.get("content-type", "").split(";")[0].strip().lower()
        if content_type.startswith("multipart/"):
            raise MultipartError("nested multipart is not accepted")
        parts[name] = Part(name=name, filename=filename, content_type=content_type, body=raw)
    if not closed:
        raise MultipartError("missing closing boundary")
    return parts