"""Wire protocol Pydantic models — Muse CLI native format (msgpack). These models match the Muse CLI ``HttpTransport`` wire format exactly. All fields are snake_case to match Muse's internal CommitDict/SnapshotDict/ ObjectPayload TypedDicts. The wire protocol is intentionally separate from the REST API's CamelModel: Wire protocol /{owner}/{slug}/push|fetch|refs ← Muse CLI speaks here (msgpack) REST API /api/repos/{id}/ ← agents and integrations speak here MCP /mcp ← agents speak here too Encoding -------- All wire endpoints accept and return ``application/x-msgpack`` binary. Objects are transported as raw ``bytes`` under the ``content`` key — no base64 encoding overhead. Denial-of-Service limits ------------------------ All list fields that arrive over the network are capped so a single large request cannot exhaust memory or DB connections: MAX_COMMITS_PER_PUSH = 10 000 — one push should carry at most 10k commits MAX_OBJECTS_PER_PUSH = 1 000 — ditto for binary blobs per chunk MAX_SNAPSHOTS_PER_PUSH = 10 000 — ditto for snapshot manifests MAX_WANT_PER_FETCH = 1 000 — fetch want/have lists MAX_OBJECT_BYTES = 38_000_000 — ~38 MB raw; objects above this limit are rejected """ import re from pydantic import BaseModel, Field, field_validator, model_validator from musehub.types.json_types import JSONObject, StrDict from musehub.types.pydantic_types import PydanticJson type _SizeMap = dict[str, int] # ── Per-request DoS limits ──────────────────────────────────────────────────── MAX_COMMITS_PER_PUSH: int = 10_000 MAX_OBJECTS_PER_PUSH: int = 1_000 # ── Object ID validation ────────────────────────────────────────────────────── # object_id values arrive from untrusted clients and are used to construct # storage keys (S3/R2 object paths). A malicious value containing '/' or '..' # could escape the objects/ key namespace and overwrite arbitrary R2 keys. # # Valid format: : # - algo : lower-case alphanumeric only (e.g. "sha256", "blake3") — no slashes # - digest: lowercase hex, at least 32 chars (128-bit minimum) # Raw hex (no prefix) is rejected — the algo: prefix is mandatory everywhere. # The pattern is intentionally algo-agnostic so future hash upgrades (blake3, # sha3-256, …) require no validator change. The hex-only digest ensures no # path-traversal characters ('.', '/') can appear in the storage key. _OBJECT_ID_RE: re.Pattern[str] = re.compile(r"^[a-z][a-z0-9]*:[0-9a-f]{32,}$") _OBJECT_ID_MAX_LEN: int = 200 # generous cap; algo(<=16) + ":" + digest(<=128) = 145 def _validate_object_ids(ids: list[str]) -> list[str]: """Raise ValueError for any object_id that contains unsafe characters.""" for oid in ids: if not _OBJECT_ID_RE.match(oid): raise ValueError( f"invalid object_id {oid!r}: only [a-zA-Z0-9:_-] characters are allowed" ) if len(oid) > _OBJECT_ID_MAX_LEN: raise ValueError( f"object_id exceeds maximum length ({_OBJECT_ID_MAX_LEN}): {oid[:40]!r}…" ) return ids MAX_SNAPSHOTS_PER_PUSH: int = 10_000 MAX_WANT_PER_FETCH: int = 1_000 # Raw bytes limit per object — objects above this are rejected at the wire layer. MAX_OBJECT_BYTES: int = 38_000_000 class WireCommit(BaseModel): """Muse native commit record — mirrors CommitDict from muse.core.store. Field names match CommitDict exactly so both sides of the wire use the same vocabulary. ``branch`` is the branch where the author made the commit; it is distinct from the push-target branch in the push request body. """ commit_id: str repo_id: str = "" branch: str = "" # author's branch (CommitDict.branch) snapshot_id: str | None = None message: str = "" committed_at: str = "" # ISO-8601 UTC string parent_commit_id: str | None = None # first parent (linear history) parent2_commit_id: str | None = None # second parent (merge commits) author: str = "" metadata: StrDict = Field(default_factory=dict) structured_delta: PydanticJson | None = None # domain-specific delta blob sem_ver_bump: str = "none" # "none" | "patch" | "minor" | "major" breaking_changes: list[str] = Field(default_factory=list) agent_id: str = "" model_id: str = "" toolchain_id: str = "" prompt_hash: str = "" signature: str = "" signer_public_key: str = "" signer_key_id: str = "" format_version: int = 7 reviewed_by: list[str] = Field(default_factory=list) test_runs: int = 0 model_config = {"extra": "ignore"} # tolerate future Muse fields gracefully @field_validator("commit_id") @classmethod def _check_commit_id(cls, v: str) -> str: if not _OBJECT_ID_RE.match(v): raise ValueError( f"invalid commit_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" ) return v @field_validator("snapshot_id") @classmethod def _check_snapshot_id(cls, v: str | None) -> str | None: if v is not None and not _OBJECT_ID_RE.match(v): raise ValueError( f"invalid snapshot_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" ) return v @field_validator("parent_commit_id") @classmethod def _check_parent_commit_id(cls, v: str | None) -> str | None: if v is not None and not _OBJECT_ID_RE.match(v): raise ValueError( f"invalid parent_commit_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" ) return v @field_validator("prompt_hash") @classmethod def _check_prompt_hash(cls, v: str) -> str: if v and not _OBJECT_ID_RE.match(v): raise ValueError( f"invalid prompt_hash {v!r}: must be empty or 'sha256:<64 lowercase hex chars>'" ) return v class WireSnapshot(BaseModel): """Unified snapshot wire format — same shape in both push and fetch directions. Both the client (push) and server (fetch) use delta encoding: - ``delta_upsert`` — files added or changed relative to parent ({path: oid}) - ``delta_remove`` — paths removed relative to parent - ``parent_snapshot_id`` — None for the root snapshot of a push chain The root snapshot of a new repo has no parent; its ``delta_upsert`` equals the full manifest. All other snapshots carry only the diff. ``directories`` is the sorted list of workspace-relative directory paths tracked at snapshot time. It is included in the snapshot_id hash. The client's ``apply_mpack`` already handles this format. ``manifest`` is accepted for backward compatibility but never produced by the server. """ snapshot_id: str parent_snapshot_id: str | None = None delta_upsert: StrDict = Field(default_factory=dict, max_length=10_000) delta_remove: list[str] = Field(default_factory=list, max_length=10_000) directories: list[str] = Field(default_factory=list, max_length=10_000) created_at: str = "" model_config = {"extra": "ignore"} @field_validator("snapshot_id") @classmethod def _check_snapshot_id(cls, v: str) -> str: if not _OBJECT_ID_RE.match(v): raise ValueError( f"invalid snapshot_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" ) return v @field_validator("delta_upsert") @classmethod def _check_delta_upsert_values(cls, v: StrDict) -> StrDict: for path, oid in v.items(): if not _OBJECT_ID_RE.match(oid): raise ValueError( f"delta_upsert entry {path!r} has invalid object_id {oid!r}: " "must be 'sha256:<64 lowercase hex chars>'" ) if len(oid) > _OBJECT_ID_MAX_LEN: raise ValueError( f"delta_upsert entry {path!r} object_id exceeds maximum length: {oid[:40]!r}…" ) return v class WireObject(BaseModel): """Content-addressed object payload — mirrors ObjectPayload from muse.core.mpack. ``content`` is raw bytes (msgpack bin type) — no base64 overhead. Encoding field controls how the server interprets ``content``: ``"raw"`` — plain bytes; store as-is after hash verification. ``"zlib"`` — zlib-compressed; decompress then verify hash. ``"delta+zlib"`` — delta-encoded relative to ``base_id``, then zlib-compressed; fetch base, apply delta, then verify hash. """ object_id: str content: bytes = Field(max_length=MAX_OBJECT_BYTES) path: str = Field(default="", max_length=4096) encoding: str = Field(default="raw") base_id: str | None = Field(default=None) model_config = {"extra": "ignore"} @field_validator("object_id") @classmethod def _check_object_id(cls, v: str) -> str: if not _OBJECT_ID_RE.match(v): raise ValueError( f"invalid object_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" ) return v @field_validator("content") @classmethod def _check_content_size(cls, v: bytes) -> bytes: if len(v) > MAX_OBJECT_BYTES: raise ValueError( f"content exceeds maximum size ({MAX_OBJECT_BYTES} bytes)." ) return v class WireMPack(BaseModel): """An mpack sent in a push request. Mirrors MPack from muse.core.mpack. All fields are optional because a minimal push may only contain commits (no new objects). List lengths are capped to prevent DoS via an oversized single request. See the module-level ``MAX_*`` constants for the exact limits. """ commits: list[WireCommit] = Field(default_factory=list, max_length=MAX_COMMITS_PER_PUSH) snapshots: list[WireSnapshot] = Field(default_factory=list, max_length=MAX_SNAPSHOTS_PER_PUSH) objects: list[WireObject] = Field(default_factory=list, max_length=MAX_OBJECTS_PER_PUSH) branch_heads: StrDict = Field(default_factory=dict) class WireFetchRequest(BaseModel): """Body for ``POST /wire/repos/{repo_id}/fetch``. Matches HttpTransport.fetch_mpack() payload: ``{"want": [...sha...], "have": [...sha...]}`` ``want`` — commit SHAs the client wants. ``have`` — commit SHAs the client already has (exclusion list). """ want: list[str] = Field(default_factory=list, max_length=MAX_WANT_PER_FETCH) have: list[str] = Field(default_factory=list, max_length=MAX_WANT_PER_FETCH) depth: int | None = Field(default=None, ge=1) @field_validator("want", "have") @classmethod def _check_commit_ids(cls, v: list[str]) -> list[str]: return _validate_object_ids(v) class WireRefsResponse(BaseModel): """Response for ``GET /wire/repos/{repo_id}/refs``. Parsed by HttpTransport._parse_remote_info() into RemoteInfo. """ repo_id: str domain: str default_branch: str branch_heads: StrDict