gabriel / musehub public
wire.py python
399 lines 17.0 KB
Raw
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago
1 """Wire protocol Pydantic models — Muse CLI native format (MWP — Muse Wire Protocol).
2
3 These models match the Muse CLI ``HttpTransport`` wire format exactly.
4 All fields are snake_case to match Muse's internal CommitDict/SnapshotDict/
5 ObjectPayload TypedDicts.
6
7 The wire protocol is intentionally separate from the REST API's CamelModel:
8 Wire protocol /wire/repos/{repo_id}/ ← Muse CLI speaks here (MWP, msgpack)
9 REST API /api/repos/{id}/ ← agents and integrations speak here
10 MCP /mcp ← agents speak here too
11
12 Encoding
13 --------
14 All wire endpoints accept and return ``application/x-msgpack`` binary.
15 Objects are transported as raw ``bytes`` under the ``content`` key — no
16 base64 encoding overhead.
17
18 Denial-of-Service limits
19 ------------------------
20 All list fields that arrive over the network are capped so a single large
21 request cannot exhaust memory or DB connections:
22
23 MAX_COMMITS_PER_PUSH = 10 000 — one push should carry at most 10k commits
24 MAX_OBJECTS_PER_PUSH = 1 000 — ditto for binary blobs per chunk
25 MAX_SNAPSHOTS_PER_PUSH = 10 000 — ditto for snapshot manifests
26 MAX_WANT_PER_FETCH = 1 000 — fetch want/have lists
27 MAX_OBJECT_BYTES = 38_000_000 — ~38 MB raw; objects above this limit are rejected
28 """
29
30 import re
31
32 from pydantic import BaseModel, Field, field_validator, model_validator
33
34 from musehub.types.json_types import JSONObject, StrDict
35 from musehub.types.pydantic_types import PydanticJson
36
37 type _SizeMap = dict[str, int]
38
39 # ── Per-request DoS limits ────────────────────────────────────────────────────
40 MAX_COMMITS_PER_PUSH: int = 10_000
41 MAX_OBJECTS_PER_PUSH: int = 1_000
42
43 # ── Object ID validation ──────────────────────────────────────────────────────
44 # object_id values arrive from untrusted clients and are used to construct
45 # storage keys (S3/R2 object paths). A malicious value containing '/' or '..'
46 # could escape the objects/ key namespace and overwrite arbitrary R2 keys.
47 #
48 # Valid format: <algo>:<lowercase hex digest>
49 # - algo : lower-case alphanumeric only (e.g. "sha256", "blake3") — no slashes
50 # - digest: lowercase hex, at least 32 chars (128-bit minimum)
51 # Raw hex (no prefix) is rejected — the algo: prefix is mandatory everywhere.
52 # The pattern is intentionally algo-agnostic so future hash upgrades (blake3,
53 # sha3-256, …) require no validator change. The hex-only digest ensures no
54 # path-traversal characters ('.', '/') can appear in the storage key.
55 _OBJECT_ID_RE: re.Pattern[str] = re.compile(r"^[a-z][a-z0-9]*:[0-9a-f]{32,}$")
56 _OBJECT_ID_MAX_LEN: int = 200 # generous cap; algo(<=16) + ":" + digest(<=128) = 145
57
58 def _validate_object_ids(ids: list[str]) -> list[str]:
59 """Raise ValueError for any object_id that contains unsafe characters."""
60 for oid in ids:
61 if not _OBJECT_ID_RE.match(oid):
62 raise ValueError(
63 f"invalid object_id {oid!r}: only [a-zA-Z0-9:_-] characters are allowed"
64 )
65 if len(oid) > _OBJECT_ID_MAX_LEN:
66 raise ValueError(
67 f"object_id exceeds maximum length ({_OBJECT_ID_MAX_LEN}): {oid[:40]!r}…"
68 )
69 return ids
70 MAX_SNAPSHOTS_PER_PUSH: int = 10_000
71 MAX_WANT_PER_FETCH: int = 1_000
72 # Raw bytes limit per object — objects above this are rejected at the wire layer.
73 MAX_OBJECT_BYTES: int = 38_000_000
74
75 class WireCommit(BaseModel):
76 """Muse native commit record — mirrors CommitDict from muse.core.store.
77
78 Field names match CommitDict exactly so both sides of the wire use the
79 same vocabulary. ``branch`` is the branch where the author made the
80 commit; it is distinct from the push-target branch carried in the H frame.
81 """
82
83 commit_id: str
84 repo_id: str = ""
85 branch: str = "" # author's branch (CommitDict.branch)
86 snapshot_id: str | None = None
87 message: str = ""
88 committed_at: str = "" # ISO-8601 UTC string
89 parent_commit_id: str | None = None # first parent (linear history)
90 parent2_commit_id: str | None = None # second parent (merge commits)
91 author: str = ""
92 metadata: StrDict = Field(default_factory=dict)
93 structured_delta: PydanticJson | None = None # domain-specific delta blob
94 sem_ver_bump: str = "none" # "none" | "patch" | "minor" | "major"
95 breaking_changes: list[str] = Field(default_factory=list)
96 agent_id: str = ""
97 model_id: str = ""
98 toolchain_id: str = ""
99 prompt_hash: str = ""
100 signature: str = ""
101 signer_public_key: str = ""
102 signer_key_id: str = ""
103 format_version: int = 7
104 reviewed_by: list[str] = Field(default_factory=list)
105 test_runs: int = 0
106
107 model_config = {"extra": "ignore"} # tolerate future Muse fields gracefully
108
109 @field_validator("commit_id")
110 @classmethod
111 def _check_commit_id(cls, v: str) -> str:
112 if not _OBJECT_ID_RE.match(v):
113 raise ValueError(
114 f"invalid commit_id {v!r}: must be 'sha256:<64 lowercase hex chars>'"
115 )
116 return v
117
118 @field_validator("snapshot_id")
119 @classmethod
120 def _check_snapshot_id(cls, v: str | None) -> str | None:
121 if v is not None and not _OBJECT_ID_RE.match(v):
122 raise ValueError(
123 f"invalid snapshot_id {v!r}: must be 'sha256:<64 lowercase hex chars>'"
124 )
125 return v
126
127 @field_validator("parent_commit_id")
128 @classmethod
129 def _check_parent_commit_id(cls, v: str | None) -> str | None:
130 if v is not None and not _OBJECT_ID_RE.match(v):
131 raise ValueError(
132 f"invalid parent_commit_id {v!r}: must be 'sha256:<64 lowercase hex chars>'"
133 )
134 return v
135
136 @field_validator("prompt_hash")
137 @classmethod
138 def _check_prompt_hash(cls, v: str) -> str:
139 if v and not _OBJECT_ID_RE.match(v):
140 raise ValueError(
141 f"invalid prompt_hash {v!r}: must be empty or 'sha256:<64 lowercase hex chars>'"
142 )
143 return v
144
145 class WireSnapshot(BaseModel):
146 """Muse native snapshot — mirrors SnapshotDict from muse.core.store.
147
148 The manifest maps file paths to content-addressed object IDs,
149 e.g. ``{"muse/core/pack.py": "sha256:abc123..."}``.
150
151 ``directories`` is the sorted list of workspace-relative directory paths
152 that were explicitly tracked at snapshot time. It is included in the
153 snapshot ID hash — omitting it causes every snapshot with non-empty
154 directories to fail content-hash verification on clone.
155 """
156
157 snapshot_id: str
158 # max_length caps the number of manifest entries — a 10 000-file snapshot
159 # would already be pathologically large; prevent unbounded dict parsing.
160 manifest: StrDict = Field(default_factory=dict, max_length=10_000)
161 directories: list[str] = Field(default_factory=list, max_length=10_000)
162 created_at: str = ""
163
164 model_config = {"extra": "ignore"}
165
166 @field_validator("snapshot_id")
167 @classmethod
168 def _check_snapshot_id(cls, v: str) -> str:
169 if not _OBJECT_ID_RE.match(v):
170 raise ValueError(
171 f"invalid snapshot_id {v!r}: must be 'sha256:<64 lowercase hex chars>'"
172 )
173 return v
174
175 @field_validator("manifest")
176 @classmethod
177 def _check_manifest_values(cls, v: StrDict) -> StrDict:
178 for path, oid in v.items():
179 if not _OBJECT_ID_RE.match(oid):
180 raise ValueError(
181 f"manifest entry {path!r} has invalid object_id {oid!r}: "
182 "must be 'sha256:<64 lowercase hex chars>'"
183 )
184 if len(oid) > _OBJECT_ID_MAX_LEN:
185 raise ValueError(
186 f"manifest entry {path!r} object_id exceeds maximum length: {oid[:40]!r}…"
187 )
188 return v
189
190 class WireSnapshotDelta(BaseModel):
191 """Delta-encoded snapshot — reconstructed by applying added/removed to a base manifest.
192
193 Sent in C frames as ``snapshot_deltas`` alongside (or instead of) full
194 ``snapshots``. The server reconstructs the full manifest by looking up
195 ``base_id`` in its stream-local cache and applying the delta.
196
197 Wire format::
198
199 {
200 "snapshot_id": "sha256:<64hex>",
201 "base_id": "sha256:<64hex>", # snapshot already seen in this stream
202 "added": {"path": "sha256:<64hex>", ...}, # new + modified files
203 "removed": ["path", ...], # deleted files
204 "directories": [...],
205 "created_at": "...",
206 }
207 """
208
209 snapshot_id: str
210 base_id: str
211 added: StrDict = Field(default_factory=dict, max_length=10_000)
212 removed: list[str] = Field(default_factory=list, max_length=10_000)
213 directories: list[str] = Field(default_factory=list, max_length=10_000)
214 created_at: str = ""
215
216 model_config = {"extra": "ignore"}
217
218 @field_validator("snapshot_id", "base_id")
219 @classmethod
220 def _check_id(cls, v: str) -> str:
221 if not _OBJECT_ID_RE.match(v):
222 raise ValueError(
223 f"invalid snapshot id {v!r}: must be 'sha256:<64 lowercase hex chars>'"
224 )
225 return v
226
227 @field_validator("added")
228 @classmethod
229 def _check_added_values(cls, v: StrDict) -> StrDict:
230 for path, oid in v.items():
231 if not _OBJECT_ID_RE.match(oid):
232 raise ValueError(
233 f"delta added entry {path!r} has invalid object_id {oid!r}: "
234 "must be 'sha256:<64 lowercase hex chars>'"
235 )
236 if len(oid) > _OBJECT_ID_MAX_LEN:
237 raise ValueError(
238 f"delta added entry {path!r} object_id exceeds maximum length: {oid[:40]!r}…"
239 )
240 return v
241
242 class WireObject(BaseModel):
243 """Content-addressed object payload — mirrors ObjectPayload from muse.core.pack.
244
245 MWP encodes ``content`` as raw bytes (msgpack bin type) — no base64 overhead.
246
247 Encoding field controls how the server interprets ``content``:
248 ``"raw"`` — plain bytes; store as-is after hash verification.
249 ``"zlib"`` — zlib-compressed; decompress then verify hash.
250 ``"delta+zlib"`` — delta-encoded relative to ``base_id``, then zlib-compressed;
251 fetch base, apply delta, then verify hash.
252 """
253
254 object_id: str
255 content: bytes = Field(max_length=MAX_OBJECT_BYTES)
256 path: str = Field(default="", max_length=4096)
257 encoding: str = Field(default="raw")
258 base_id: str | None = Field(default=None)
259
260 model_config = {"extra": "ignore"}
261
262 @field_validator("object_id")
263 @classmethod
264 def _check_object_id(cls, v: str) -> str:
265 if not _OBJECT_ID_RE.match(v):
266 raise ValueError(
267 f"invalid object_id {v!r}: must be 'sha256:<64 lowercase hex chars>'"
268 )
269 return v
270
271 @field_validator("content")
272 @classmethod
273 def _check_content_size(cls, v: bytes) -> bytes:
274 if len(v) > MAX_OBJECT_BYTES:
275 raise ValueError(
276 f"content exceeds maximum size ({MAX_OBJECT_BYTES} bytes)."
277 )
278 return v
279
280 class WireBundle(BaseModel):
281 """A pack bundle sent in a push request.
282
283 Mirrors PackBundle from muse.core.pack. All fields are optional because
284 a minimal push may only contain commits (no new objects).
285
286 List lengths are capped to prevent DoS via an oversized single request.
287 See the module-level ``MAX_*`` constants for the exact limits.
288 """
289
290 commits: list[WireCommit] = Field(default_factory=list, max_length=MAX_COMMITS_PER_PUSH)
291 snapshots: list[WireSnapshot] = Field(default_factory=list, max_length=MAX_SNAPSHOTS_PER_PUSH)
292 objects: list[WireObject] = Field(default_factory=list, max_length=MAX_OBJECTS_PER_PUSH)
293 branch_heads: StrDict = Field(default_factory=dict)
294
295 class WireFetchRequest(BaseModel):
296 """Body for ``POST /wire/repos/{repo_id}/fetch``.
297
298 Matches HttpTransport.fetch_pack() payload:
299 ``{"want": [...sha...], "have": [...sha...]}``
300
301 ``want`` — commit SHAs the client wants.
302 ``have`` — commit SHAs the client already has (exclusion list).
303 """
304
305 want: list[str] = Field(default_factory=list, max_length=MAX_WANT_PER_FETCH)
306 have: list[str] = Field(default_factory=list, max_length=MAX_WANT_PER_FETCH)
307 depth: int | None = Field(default=None, ge=1)
308
309 @field_validator("want", "have")
310 @classmethod
311 def _check_commit_ids(cls, v: list[str]) -> list[str]:
312 return _validate_object_ids(v)
313
314 class WireRefsResponse(BaseModel):
315 """Response for ``GET /wire/repos/{repo_id}/refs``.
316
317 Parsed by HttpTransport._parse_remote_info() into RemoteInfo.
318 """
319
320 repo_id: str
321 domain: str
322 default_branch: str
323 branch_heads: StrDict
324
325 class WireNegotiateRequest(BaseModel):
326 """Body for ``POST /{owner}/{slug}/negotiate`` — commit negotiation."""
327 have: list[str] = []
328 want: list[str] = []
329
330 @field_validator("have", "want")
331 @classmethod
332 def _check_commit_ids(cls, v: list[str]) -> list[str]:
333 return _validate_object_ids(v)
334
335 class WireNegotiateResponse(BaseModel):
336 """Response for ``POST /{owner}/{slug}/negotiate`` — commit negotiation."""
337 ack: list[str]
338 common_base: str | None
339 ready: bool
340
341 # ─────────────────────────────────────────────────────────────────────────────
342 # MWP — Streaming Push Protocol
343 # ─────────────────────────────────────────────────────────────────────────────
344 #
345 # Two push paths: stream (small) and presigned R2 (large).
346 # This is the server-side equivalent of
347 # git-receive-pack's smart HTTP protocol:
348 #
349 # git smart HTTP: Muse Wire Protocol (MWP):
350 # ───────────────────── ──────────────────────────────
351 # GET /info/refs GET /{owner}/{slug}/refs (unchanged)
352 # POST /git-receive-pack POST /{owner}/{slug}/push/stream (single streaming request)
353 #
354 # Request body — concatenated self-delimiting msgpack values (no length prefix;
355 # msgpack.Unpacker handles framing exactly like the existing /fetch/objects stream):
356 #
357 # frame 1: HEADER frame — branch, force, have-list, object/commit counts
358 # frame 2…N: OBJECT frames — content-addressed binary objects (zlib-compressed)
359 # frame N+1: COMMIT_PACK — commits and snapshot manifests
360 # frame N+2: END — stream terminator
361 #
362 # Response body — same wire format (concatenated msgpack dicts):
363 #
364 # frame(s): PROGRESS — human-readable progress (print to stderr)
365 # frame: ERROR — fatal; client aborts, push failed
366 # frame: RESULT — final push result; stream ends
367 #
368 # Content-Type: application/x-muse-wire (both request and response)
369 #
370 # ── Frame type tag constants ──────────────────────────────────────────────────
371
372 # Request frame types (client → server) — single-char tags keep wire overhead minimal.
373 SFRAME_HEADER: str = "H" # opens stream; branch, force, have, object/commit count
374 SFRAME_OBJECT: str = "O" # one content-addressed binary object
375 SFRAME_OBJECT_CHUNK: str = "OC" # one chunk of a large object (Phase 14B)
376 SFRAME_COMMIT_PACK: str = "C" # final batch of commits + snapshot manifests
377 SFRAME_END: str = "E" # stream terminator
378
379 # Response frame types (server → client)
380 SFRAME_PROGRESS: str = "P" # human-readable progress (sideband 2 — stderr)
381 SFRAME_ERROR: str = "X" # fatal error — client aborts; no RESULT follows
382 SFRAME_RESULT: str = "R" # final push result — always the last frame on success
383
384 # ── Per-stream DoS limits ─────────────────────────────────────────────────────
385 #
386 # These are deliberately generous: a single streaming push replaces what used to
387 # be dozens of individual HTTP requests. The server enforces hard caps so an
388 # adversarial client cannot exhaust memory or storage by sending an unbounded stream.
389
390 STREAM_MAX_OBJECTS: int = 500_000 # 500k objects per single push stream
391 STREAM_MAX_COMMITS: int = 1_000_000 # 1M commits per single push stream
392
393 # Maximum wire bytes the server accepts per OBJECT frame's content field.
394 # Applies to the *compressed* wire bytes; the server decompresses before storing.
395 STREAM_MAX_OBJECT_WIRE_BYTES: int = 50_000_000 # 50 MB compressed per object
396
397 # R2 PUT batch size during stream ingestion — flush every N incoming OBJECT frames.
398 # Must not exceed _R2_PUT_SEM capacity (100) in musehub_wire.py.
399 STREAM_OBJECT_BATCH_SIZE: int = 64
File History 1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef debug(push/stream): instrument O-frame decode path with INF… Sonnet 4.6 patch 123 days ago