aaronrene / gabriel-muse public
transport.py python
2,314 lines 90.8 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
1 """Muse transport layer — HTTP and local-filesystem remote communication.
2
3 The :class:`MuseTransport` Protocol defines the interface between the Muse CLI
4 and a remote host. The CLI calls this Protocol; the implementation is chosen
5 at runtime by :func:`make_transport` based on the URL scheme.
6
7 Transport implementations
8 --------------------------
9
10 :class:`HttpTransport`
11 Synchronous HTTPS transport using stdlib ``urllib.request``. Used for
12 ``https://`` remote URLs (MuseHub server required). Encodes all request
13 bodies as ``msgpack`` (``Content-Type: application/x-msgpack``), which
14 eliminates the 33 % base64 inflation of the old JSON+base64 protocol.
15
16 :class:`LocalFileTransport`
17 Zero-network transport for ``file://`` URLs. Reads and writes directly
18 from the remote's ``.muse/`` directory on the local filesystem (or a
19 shared network mount). No server is required — ideal for local testing,
20 monorepo setups, and offline workflows.
21
22 Use :func:`make_transport` instead of constructing either class directly —
23 it inspects the URL scheme and returns the appropriate implementation.
24
25 MWP — Muse Wire Protocol
26 -------------------------------
27
28 All endpoints live under the remote repository URL. MWP introduces five
29 new phases on top of the original refs/fetch/push trio:
30
31 +-----------------------------------+----------------------------------------+
32 | Endpoint | Purpose |
33 +===================================+========================================+
34 | GET {url}/refs | Pre-flight: branch heads + metadata |
35 +-----------------------------------+----------------------------------------+
36 | POST {url}/filter-objects | Phase 1: dedup — missing object IDs |
37 +-----------------------------------+----------------------------------------+
38 | POST {url}/presign | Phase 3: presigned S3/R2 PUT/GET URLs |
39 +-----------------------------------+----------------------------------------+
40 | POST {url}/push/objects | Pre-upload object chunks (small objs) |
41 +-----------------------------------+----------------------------------------+
42 | POST {url}/push | Phase 5: commit + snapshot push |
43 +-----------------------------------+----------------------------------------+
44 | POST {url}/fetch | Download commit delta pack |
45 +-----------------------------------+----------------------------------------+
46 | POST {url}/negotiate | Phase 5: depth-limited have/ack loop |
47 +-----------------------------------+----------------------------------------+
48
49 Wire encoding
50 ~~~~~~~~~~~~~
51
52 Request bodies are ``msgpack``-encoded dicts (``Content-Type:
53 application/x-msgpack``). The ``Accept`` header advertises both ``msgpack``
54 and ``application/json`` so older servers can respond in JSON. Objects are
55 transmitted as raw ``bytes`` in :class:`~muse.core.pack.ObjectPayload`
56 (``content`` field) — no base64 encoding.
57
58 Authentication
59 --------------
60
61 All endpoints accept an ``Authorization: MSign handle="<handle>" ts=<unix> sig="<b64url>"``
62 header for Ed25519 per-request signing. The signing identity (handle + private key) is
63 read from ``~/.muse/identity.toml`` via :func:`muse.cli.config.get_signing_identity` and
64 is **never** written to any log line.
65
66 :class:`LocalFileTransport` ignores the ``signing`` argument — local repos
67 do not require authentication (access is controlled by filesystem permissions).
68
69 Error codes (HttpTransport)
70 ----------------------------
71
72 401 Unauthorized — invalid or missing signature
73 404 Not found — repo does not exist on the remote
74 409 Conflict — push rejected (non-fast-forward without ``--force``)
75 5xx Server error
76
77 Security model
78 --------------
79
80 HttpTransport:
81 - Refuses all HTTP redirects (prevents credential leakage to other hosts).
82 - Rejects non-HTTPS URLs when a token is present (prevents cleartext exposure).
83 - Caps response bodies at ``MAX_RESPONSE_BYTES`` (64 MiB) to prevent OOM.
84
85 LocalFileTransport:
86 - Calls ``.resolve()`` on all filesystem paths (canonicalises symlinks and
87 ``..`` components before any I/O).
88 - Validates branch names with ``validate_branch_name`` (rejects null bytes,
89 backslashes, consecutive dots, and other path-traversal primitives).
90 - Guards ref-file writes with ``contain_path`` (defence-in-depth: asserts the
91 computed path stays inside ``.muse/refs/heads/`` even after symlink resolution).
92 - Never follows redirects, makes no network calls, and ignores the token arg.
93 """
94
95 from __future__ import annotations
96
97 import base64
98 import collections.abc
99 import contextlib
100 import hashlib
101 import http.client
102 import json
103 import logging
104 import pathlib
105 import signal
106 import threading
107 import time
108 import types
109 import urllib.error
110 import urllib.parse
111 import urllib.request
112 import urllib.response
113 from typing import IO, TYPE_CHECKING, NamedTuple, Protocol, TypedDict, runtime_checkable
114
115 if TYPE_CHECKING:
116 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
117
118
119 class SigningIdentity(NamedTuple):
120 """Ed25519 signing identity for MSign request authentication."""
121
122 handle: str # hub handle, e.g. "gabriel"
123 private_key: Ed25519PrivateKey
124
125
126 def build_msign_header(
127 signing: SigningIdentity,
128 method: str,
129 url: str,
130 body_bytes: bytes | None = None,
131 ) -> str:
132 """Return an ``Authorization: MSign ...`` header value for a request.
133
134 Suitable for use in custom ``urllib.request.Request`` headers when the
135 caller cannot use :class:`HttpTransport` directly.
136
137 Args:
138 signing: The signing identity (handle + Ed25519 private key).
139 method: HTTP method (``"GET"``, ``"POST"``, etc.).
140 url: Full request URL.
141 body_bytes: Request body bytes, or ``None`` for empty body.
142
143 Returns:
144 The complete ``Authorization`` header value, e.g.
145 ``MSign handle="gabriel" ts=1712345678 sig="..."``
146 """
147 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
148
149 parsed = urllib.parse.urlparse(url)
150 path_with_query = parsed.path
151 if parsed.query:
152 path_with_query = f"{path_with_query}?{parsed.query}"
153
154 ts = int(time.time())
155 body = body_bytes or b""
156 body_hash = hashlib.sha256(body).hexdigest()
157 canonical = f"{method}\n{path_with_query}\n{ts}\n{body_hash}".encode()
158
159 private_key = signing.private_key
160 assert isinstance(private_key, Ed25519PrivateKey)
161 sig_bytes = private_key.sign(canonical)
162 sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode("ascii")
163
164 return f'MSign handle="{signing.handle}" ts={ts} sig="{sig_b64}"'
165
166 import msgpack
167
168 from muse.core.compression import apply_delta, compress_zlib, compute_delta, decompress_zlib
169 from muse.core.pack import (
170 FetchRequest,
171 ObjectPayload,
172 ObjectsChunkResponse,
173 PackBundle,
174 PushResult,
175 RemoteInfo,
176 WireTag,
177 )
178 from muse.core.semver import ChangelogEntry, SemVerTag
179 from muse.core.store import (
180 BranchHeads,
181 CommitDict,
182 Manifest,
183 Metadata,
184 MsgpackDict,
185 ReleaseDict,
186 SnapshotDict,
187 safe_unpackb,
188 )
189 from muse.core.validation import (
190 MAX_FETCH_BYTES,
191 MAX_RESPONSE_BYTES,
192 contain_path,
193 sanitize_display,
194 validate_branch_name,
195 )
196 from muse.domain import SemVerBump
197
198 logger = logging.getLogger(__name__)
199
200 _TIMEOUT_SECONDS = 300
201
202 # Recursive type alias for msgpack-serializable values.
203 # Covers every type that msgpack can encode/decode natively — no base64,
204 # no `object`, no `Any`. Python 3.12+ `type` statement creates a proper
205 # TypeAlias that mypy treats as a first-class recursive type.
206 type _MsgVal = (
207 str | int | float | bool | bytes | None
208 | list[_MsgVal]
209 | dict[str, _MsgVal]
210 )
211 type _MsgDict = dict[str, _MsgVal] # top-level msgpack response dict
212 type _WireVal = _MsgVal # wire-protocol value (same shape)
213 type _WireDict = dict[str, _WireVal] # wire-protocol response dict
214 type _HeadersDict = dict[str, str] # HTTP request/response headers
215 type _CommitBundle = dict[str, CommitDict] # commit_id → commit dict
216
217 class FilterObjectsResult(TypedDict):
218 """Structured result from MWP Phase 1 (``POST /filter-objects``).
219
220 ``missing``: IDs the server does not yet have — client must upload these.
221 ``bases``: Suggested delta base for each missing object (oid → base_oid).
222 When present, the client should compute a ``delta+zlib``-encoded
223 object instead of sending the raw bytes.
224 """
225
226 missing: list[str]
227 bases: dict[str, str]
228
229
230 # Maximum number of objects to include in a single POST /push/objects call.
231 # Must stay strictly below the server's MAX_OBJECTS_PER_PUSH limit (1 000).
232 # 100 gives better pipelining than 400: smaller msgpack payloads, faster
233 # MSign computation, shorter server-side processing windows, and lower lock
234 # contention when thousands of agents push concurrently.
235 # 500 objects per chunk keeps HTTP round-trips low (5 000 objects = 10 requests
236 # instead of 50) while staying well within typical server request-body limits.
237 CHUNK_OBJECTS: int = 500
238
239 # Maximum number of commits to include in a single POST /push call.
240 # Large histories are split into sequential chunked pushes so that each
241 # request stays well within Cloudflare's 100 MB upload limit.
242 # Each intermediate chunk uses force=True; the final chunk uses the caller's
243 # force flag.
244 #
245 # 500 commits/chunk means ~2 round-trips for a 1000-commit history vs the
246 # previous 5 at 200/chunk. The server now uses bulk INSERT ON CONFLICT DO
247 # NOTHING so per-chunk processing time is proportional to commit count, not
248 # the number of individual INSERT round-trips.
249 CHUNK_COMMITS: int = 500
250
251 # Objects above this byte threshold are candidates for presigned-URL upload
252 # (MWP Phase 3) — they bypass the API server and go directly to S3/R2.
253 # Objects at or below this size are included inline in the pack body.
254 LARGE_OBJECT_THRESHOLD: int = 64 * 1024 # 64 KiB
255
256 # Depth of the have-list sent per round of commit negotiation (MWP Phase 5).
257 # Caps the negotiation payload at ≤ NEGOTIATE_DEPTH commit IDs per request,
258 # keeping negotiation O(depth) rather than O(history).
259 NEGOTIATE_DEPTH: int = 64
260
261
262 # ---------------------------------------------------------------------------
263 # MWP response TypedDicts
264 # ---------------------------------------------------------------------------
265
266
267
268 class PresignResponse(TypedDict):
269 """Response from ``POST {url}/presign`` — MWP Phase 3."""
270
271 presigned: Manifest # object_id → presigned URL
272 inline: list[str] # IDs whose backend does not support presigned URLs
273
274
275 class ConfirmObjectsResponse(TypedDict):
276 """Response from ``POST {url}/push/objects/confirm`` — MWP Phase 3b."""
277
278 registered: int # objects newly registered in the remote DB
279 skipped: int # objects already registered (idempotent no-op)
280
281
282 class NegotiateResponse(TypedDict):
283 """Response from ``POST {url}/negotiate`` — MWP Phase 5."""
284
285 ack: list[str]
286 common_base: str | None
287 ready: bool
288
289
290 class _PushPayload(TypedDict, total=False):
291 """Msgpack-encoded body for ``POST {url}/push``."""
292
293 bundle: PackBundle
294 branch: str
295 force: bool
296 local_head: str
297
298
299 # ---------------------------------------------------------------------------
300 # Response protocol — typed adapter for urllib response objects
301 # ---------------------------------------------------------------------------
302
303
304 class _HttpHeaders(Protocol):
305 """Minimal interface for HTTP response headers."""
306
307 def get(self, name: str, default: str = "") -> str: ...
308
309
310 @runtime_checkable
311 class _HttpResponse(Protocol):
312 """Structural interface for urllib HTTP response objects.
313
314 Defined as a Protocol so that ``_open_url`` can have a concrete, non-Any
315 return type without importing implementation-specific urllib internals.
316 """
317
318 headers: _HttpHeaders
319
320 def read(self, amt: int | None = None) -> bytes: ...
321
322 def __enter__(self) -> "_HttpResponse": ...
323
324 def __exit__(
325 self,
326 exc_type: type[BaseException] | None,
327 exc_val: BaseException | None,
328 exc_tb: "types.TracebackType | None",
329 ) -> bool | None: ...
330
331
332 # ---------------------------------------------------------------------------
333 # Security — credential-safe HTTP error formatting
334 # ---------------------------------------------------------------------------
335
336
337 def _http_error_message(exc: urllib.error.HTTPError) -> str:
338 """Build a safe, sanitized error message for an HTTP error response.
339
340 HTTP 401 responses are especially dangerous: the server received the
341 ``Authorization: MSign`` header, and a malicious or compromised
342 server may echo credential data verbatim in the response body. We NEVER
343 include the response body for 401 — instead we emit a generic
344 "Authentication failed" string that contains no server-controlled content.
345
346 For all other status codes we include the server error body (capped at
347 200 characters) after stripping terminal control characters (ANSI escape
348 sequences, OSC sequences, etc.) to block terminal injection.
349 """
350 if exc.code == 401:
351 return "Authentication failed (HTTP 401). Run 'muse auth register'."
352 try:
353 raw_body: str = exc.read().decode("utf-8", errors="replace")
354 except Exception: # noqa: BLE001
355 raw_body = ""
356 safe_body = sanitize_display(raw_body[:200])
357 return f"HTTP {exc.code}: {safe_body}" if safe_body else f"HTTP {exc.code}"
358
359
360 # ---------------------------------------------------------------------------
361 # Security — redirect and scheme enforcement
362 # ---------------------------------------------------------------------------
363
364
365 class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
366 """Refuse all HTTP redirects.
367
368 ``urllib.request`` follows redirects by default, including across schemes
369 (HTTPS → HTTP) and across hosts. If a server we contact redirects us:
370
371 - To HTTP: the ``Authorization: MSign`` header would be sent in cleartext.
372 - To a different host: the token would be sent to an unintended recipient.
373
374 We refuse both. The server must use the correct, stable URL. If a
375 redirect is required during operations, it is always better to surface
376 it as a hard error so the operator can update the configured URL than to
377 silently follow it and leak credentials.
378 """
379
380 def redirect_request(
381 self,
382 req: urllib.request.Request,
383 fp: IO[bytes],
384 code: int,
385 msg: str,
386 headers: http.client.HTTPMessage,
387 newurl: str,
388 ) -> urllib.request.Request | None:
389 raise urllib.error.HTTPError(
390 req.full_url,
391 code,
392 (
393 f"Redirect refused ({code}): server tried to redirect to {newurl!r}. "
394 "Update the configured remote URL to the final destination."
395 ),
396 headers,
397 fp,
398 )
399
400
401 # Build one opener that never follows redirects. Used for every request
402 # so that Authorization headers are never sent to an unintended recipient.
403 _STRICT_OPENER = urllib.request.build_opener(_NoRedirectHandler())
404
405
406 @contextlib.contextmanager
407 def _ignore_sigpipe() -> collections.abc.Generator[None, None, None]:
408 """Temporarily ignore SIGPIPE during socket I/O (main thread only).
409
410 ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` at startup so that piping
411 ``muse`` output to ``head``/``grep``/``jq`` exits cleanly. But ``SIG_DFL``
412 also kills the process when a large HTTP request body is still in-flight
413 and the remote closes the connection early (auth failure, 409 conflict,
414 server crash, …). Without this guard the push command dies with exit code
415 141 instead of raising ``TransportError``.
416
417 Signal handlers may only be changed from the main interpreter thread.
418 Worker threads launched by ``ThreadPoolExecutor`` (used for parallel object
419 uploads) are already safe: Python blocks ``SIGPIPE`` in non-main threads,
420 so broken-socket writes raise ``BrokenPipeError`` rather than killing the
421 process. This context manager is therefore a no-op in non-main threads.
422
423 In the main thread the context manager saves the current SIGPIPE
424 disposition, sets it to ``SIG_IGN`` for the duration of the network call
425 (so that broken-pipe conditions surface as ``BrokenPipeError`` → urllib
426 ``URLError`` → ``TransportError``), then restores the original disposition
427 on exit. On platforms without ``SIGPIPE`` (Windows) this is a no-op.
428 """
429 if not hasattr(signal, "SIGPIPE") or not threading.current_thread() is threading.main_thread():
430 yield
431 return
432 old_handler: signal.Handlers = signal.signal(signal.SIGPIPE, signal.SIG_IGN)
433 try:
434 yield
435 finally:
436 signal.signal(signal.SIGPIPE, old_handler)
437
438
439 def _open_url(req: urllib.request.Request, timeout: int) -> _HttpResponse:
440 """Thin wrapper around ``_STRICT_OPENER.open`` — exists purely to give tests
441 a single, importable patch target instead of deep-patching the opener object.
442
443 SIGPIPE is temporarily ignored for the duration of the call so that the
444 process does not die with exit 141 when the server closes the connection
445 while a large request body is still being sent. See ``_ignore_sigpipe``.
446
447 Returns an ``_HttpResponse`` Protocol value so that callers can be fully
448 typed without depending on urllib's concrete ``addinfourl`` class.
449 """
450 with _ignore_sigpipe():
451 resp: _HttpResponse = _STRICT_OPENER.open(req, timeout=timeout)
452 return resp
453
454
455 # ---------------------------------------------------------------------------
456 # Exception
457 # ---------------------------------------------------------------------------
458
459
460 class TransportError(Exception):
461 """Raised when the remote returns a non-2xx response or is unreachable.
462
463 Attributes:
464 status_code: HTTP status code (e.g. ``401``, ``404``, ``409``, ``500``).
465 ``0`` for network-level failures (DNS, connection refused).
466 """
467
468 def __init__(self, message: str, status_code: int) -> None:
469 super().__init__(message)
470 self.status_code = status_code
471
472
473 # ---------------------------------------------------------------------------
474 # Protocol — the seam between CLI commands and the transport implementation
475 # ---------------------------------------------------------------------------
476
477
478 class MuseTransport(Protocol):
479 """Protocol for Muse remote transport implementations.
480
481 All methods are synchronous — the Muse CLI is synchronous by design.
482 """
483
484 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo:
485 """Return repository metadata from ``GET {url}/refs``.
486
487 Args:
488 url: Remote repository URL.
489 token: Ed25519 signing identity, or ``None`` for public repos.
490
491 Raises:
492 :class:`TransportError` on HTTP 4xx/5xx or network failure.
493 """
494 ...
495
496 def fetch_pack(
497 self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str]
498 ) -> PackBundle:
499 """Download a :class:`~muse.core.pack.PackBundle` via ``POST {url}/fetch``.
500
501 Phase 1 of the two-phase fetch protocol. Returns commits, snapshots, and
502 branch_heads only — no object bytes. Call :meth:`fetch_objects` separately
503 with the object IDs needed for checkout.
504
505 Args:
506 url: Remote repository URL.
507 token: Ed25519 signing identity, or ``None``.
508 want: Commit IDs the client wants to receive.
509 have: Commit IDs already present locally.
510
511 Raises:
512 :class:`TransportError` on HTTP 4xx/5xx or network failure.
513 """
514 ...
515
516 def fetch_objects(
517 self,
518 url: str,
519 signing: SigningIdentity | None,
520 object_ids: list[str],
521 ) -> list[ObjectPayload]:
522 """Download object bytes via ``POST {url}/fetch/objects``.
523
524 Phase 2 of the two-phase fetch protocol. The caller passes only the
525 IDs it is missing from its local store; the server streams back the
526 bytes for each one.
527
528 Args:
529 url: Remote repository URL.
530 signing: Ed25519 signing identity, or ``None``.
531 object_ids: Object IDs whose bytes are needed locally.
532
533 Returns:
534 List of :class:`~muse.core.pack.ObjectPayload` with ``object_id``
535 and ``content``. IDs that do not exist on the remote are silently
536 omitted.
537
538 Raises:
539 :class:`TransportError` on HTTP 4xx/5xx or network failure.
540 """
541 ...
542
543 def push_pack(
544 self,
545 url: str,
546 signing: SigningIdentity | None,
547 bundle: PackBundle,
548 branch: str,
549 force: bool,
550 local_head: str | None = None,
551 ) -> PushResult:
552 """Upload a :class:`~muse.core.pack.PackBundle` via ``POST {url}/push``.
553
554 Args:
555 url: Remote repository URL.
556 token: Ed25519 signing identity, or ``None``.
557 bundle: Bundle to upload.
558 branch: Remote branch to update.
559 force: Bypass the server-side fast-forward check.
560
561 Raises:
562 :class:`TransportError` on HTTP 4xx/5xx or network failure.
563 """
564 ...
565
566 def push_objects(
567 self,
568 url: str,
569 signing: SigningIdentity | None,
570 objects: list[ObjectPayload],
571 ) -> ObjectsChunkResponse:
572 """Pre-upload a batch of content-addressed objects via ``POST {url}/push/objects``.
573
574 This is Phase 1 of a chunked push. The caller splits the full object
575 list into batches of at most :data:`CHUNK_OBJECTS` and calls this once
576 per batch. After all batches succeed, the caller issues a single
577 ``push_pack`` (Phase 2) with an empty ``objects`` list — the final push
578 only carries commits and snapshots, which are small.
579
580 Objects are idempotent: the server skips any it already holds and
581 counts them in ``skipped``. Uploading the same object twice is always
582 safe.
583
584 Args:
585 url: Remote repository URL.
586 token: Ed25519 signing identity, or ``None``.
587 objects: Batch of objects to upload (``len ≤ CHUNK_OBJECTS``).
588
589 Returns:
590 :class:`~muse.core.pack.ObjectsChunkResponse` with ``stored`` and
591 ``skipped`` counts.
592
593 Raises:
594 :class:`TransportError` on HTTP 4xx/5xx or network failure.
595 """
596 ...
597
598 def push_object_pack(
599 self,
600 url: str,
601 signing: SigningIdentity | None,
602 objects: list[ObjectPayload],
603 ) -> ObjectsChunkResponse:
604 """Upload a pack of ≤ 1 000 small objects in a single ``POST {url}/push/object-pack``.
605
606 The pack endpoint is the web-scale alternative to the presigned PUT path for
607 small objects (≤ ``LARGE_OBJECT_THRESHOLD``, i.e. 1 MB). Instead of one TLS
608 handshake per object, the client bundles objects into msgpack packs and POSTs
609 them to the API server. Round-trips collapse from O(N) to O(ceil(N/1000)).
610
611 For objects larger than ``LARGE_OBJECT_THRESHOLD``, the presigned URL path
612 (``presign_objects`` + direct PUT) is still preferred — bypassing Cloudflare
613 is worthwhile when the per-object body is large enough to saturate bandwidth.
614
615 Objects are content-addressed and idempotent — re-sending an object that
616 already exists on the remote is safe; it is counted in ``skipped``.
617
618 Args:
619 url: Remote repository URL.
620 signing: Ed25519 signing identity, or ``None``.
621 objects: Batch of objects to upload (``len ≤ PACK_MAX_OBJECTS = 1 000``,
622 total bytes ≤ ``PACK_MAX_BYTES = 50 MB``).
623
624 Returns:
625 :class:`~muse.core.pack.ObjectsChunkResponse` with ``stored`` and
626 ``skipped`` counts.
627
628 Raises:
629 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if the
630 remote server does not support the pack endpoint (404 → caller should
631 fall back to ``push_objects``).
632 """
633 ...
634
635 def filter_objects(
636 self,
637 url: str,
638 signing: SigningIdentity | None,
639 object_ids: list[str],
640 object_hints: dict[str, str] | None = None,
641 ) -> FilterObjectsResult:
642 """Return missing object IDs and delta base suggestions (MWP Phase 1).
643
644 The client sends every candidate object ID plus optional path hints.
645 The server responds with the subset it is missing and, for each missing
646 object whose path already exists in the repo, a suggested base object ID
647 for delta encoding.
648
649 Args:
650 url: Remote repository URL.
651 signing: Ed25519 signing identity, or ``None``.
652 object_ids: All object IDs the client intends to upload.
653 object_hints: ``{oid: path}`` map for delta base lookup (optional).
654
655 Returns:
656 :class:`FilterObjectsResult` with ``missing`` IDs and ``bases`` map.
657
658 Raises:
659 :class:`TransportError` on HTTP 4xx/5xx or network failure.
660 """
661 ...
662
663 def presign_objects(
664 self,
665 url: str,
666 signing: SigningIdentity | None,
667 object_ids: list[str],
668 direction: str,
669 ) -> PresignResponse:
670 """Return presigned S3/R2 URLs for direct-to-storage transfer.
671
672 MWP Phase 3 — when the backend supports presigned URLs (S3/R2), ALL
673 missing objects are presigned and uploaded directly to object storage,
674 bypassing the API server and Cloudflare entirely.
675
676 When the backend is ``local://`` all IDs are returned in ``inline``
677 and the client falls back to the normal ``push_objects`` pack flow.
678
679 Args:
680 url: Remote repository URL.
681 signing: Ed25519 signing identity, or ``None``.
682 object_ids: IDs of all missing objects to presign.
683 direction: ``"put"`` for push, ``"get"`` for pull.
684
685 Returns:
686 :class:`PresignResponse` with ``presigned`` URL map and ``inline``
687 fallback list.
688
689 Raises:
690 :class:`TransportError` on HTTP 4xx/5xx or network failure.
691 """
692 ...
693
694 def confirm_objects(
695 self,
696 url: str,
697 signing: SigningIdentity | None,
698 object_ids: list[str],
699 sizes: dict[str, int],
700 ) -> ConfirmObjectsResponse:
701 """Register objects uploaded via presigned URL in the remote DB.
702
703 MWP Phase 3b — after uploading directly to R2, the client calls this
704 so the server can insert ``musehub_objects`` DB records. Without this
705 step, future ``filter_objects`` calls would report those objects as
706 missing and re-upload them on every push.
707
708 Args:
709 url: Remote repository URL.
710 signing: Ed25519 signing identity, or ``None``.
711 object_ids: IDs of objects the client just uploaded via presign.
712 sizes: ``{object_id: byte_count}`` — supplied by client since
713 the server never saw the bytes.
714
715 Returns:
716 :class:`ConfirmObjectsResponse` with ``registered`` and ``skipped``.
717
718 Raises:
719 :class:`TransportError` on HTTP 4xx/5xx or network failure.
720 """
721 ...
722
723 def negotiate(
724 self,
725 url: str,
726 signing: SigningIdentity | None,
727 want: list[str],
728 have: list[str],
729 ) -> NegotiateResponse:
730 """Depth-limited commit negotiation (MWP Phase 5).
731
732 Replaces sending the full local commit list as ``have``. Each round
733 sends ≤ :data:`NEGOTIATE_DEPTH` recent ancestors. The server responds
734 with which it recognises (``ack``), the common base commit (if found),
735 and whether ``ready`` — i.e. enough context to compute the delta.
736
737 Args:
738 url: Remote repository URL.
739 token: Ed25519 signing identity, or ``None``.
740 want: Branch tips the client wants to receive.
741 have: Recent local commit IDs (≤ NEGOTIATE_DEPTH per round).
742
743 Returns:
744 :class:`NegotiateResponse` with ``ack``, ``common_base``, ``ready``.
745
746 Raises:
747 :class:`TransportError` on HTTP 4xx/5xx or network failure.
748 """
749 ...
750
751 def push_tags(
752 self,
753 url: str,
754 signing: SigningIdentity | None,
755 tags: list[WireTag],
756 ) -> int:
757 """Push local tags to the remote via ``POST {url}/tags``.
758
759 Tags are immutable once created on the remote — the server skips any
760 it already holds. Returns the number of tags newly stored.
761
762 Args:
763 url: Remote repository URL.
764 token: Ed25519 signing identity, or ``None``.
765 tags: Tags to push.
766
767 Raises:
768 :class:`TransportError` on HTTP 4xx/5xx or network failure.
769 """
770 ...
771
772 def create_release(
773 self,
774 url: str,
775 signing: SigningIdentity | None,
776 release: ReleaseDict,
777 ) -> str:
778 """Create a release on the remote via ``POST {url}/releases``.
779
780 Returns the ``release_id`` assigned by the server.
781
782 Args:
783 url: Remote repository URL.
784 token: Ed25519 signing identity.
785 release: Fully-populated :class:`~muse.core.store.ReleaseDict`.
786
787 Raises:
788 :class:`TransportError` on HTTP 4xx/5xx or network failure.
789 """
790 ...
791
792 def list_releases_remote(
793 self,
794 url: str,
795 signing: SigningIdentity | None,
796 channel: str | None = None,
797 include_drafts: bool = False,
798 ) -> list[ReleaseDict]:
799 """Fetch releases from the remote via ``GET {url}/releases``.
800
801 Args:
802 url: Remote repository URL.
803 token: Ed25519 signing identity, or ``None``.
804 channel: Filter by release channel; ``None`` returns all.
805 include_drafts: Include draft releases when ``True``.
806
807 Raises:
808 :class:`TransportError` on HTTP 4xx/5xx or network failure.
809 """
810 ...
811
812 def delete_release_remote(
813 self,
814 url: str,
815 signing: SigningIdentity | None,
816 tag: str,
817 ) -> None:
818 """Retract a release from the remote via ``DELETE {url}/releases/{tag}``.
819
820 Removes only the named label from the remote registry. The underlying
821 commit and snapshot objects are **not** affected.
822
823 Args:
824 url: Remote repository URL.
825 token: Ed25519 signing identity (owner credentials required).
826 tag: Semver tag of the release to retract (e.g. ``"v1.2.0"``).
827
828 Raises:
829 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if
830 the release does not exist on the remote.
831 """
832 ...
833
834 def delete_branch_remote(
835 self,
836 url: str,
837 signing: SigningIdentity | None,
838 branch: str,
839 ) -> None:
840 """Delete a branch on the remote via ``DELETE {url}/branches/{branch}``.
841
842 Equivalent to ``git push origin --delete <branch>``. The branch ref is
843 removed from the server; commits and objects are unaffected.
844
845 Args:
846 url: Remote repository URL.
847 token: Ed25519 signing identity (owner credentials required).
848 branch: Branch name to delete (e.g. ``"feat/my-thing"``).
849
850 Raises:
851 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if
852 the branch does not exist on the remote.
853 """
854 ...
855
856
857 # ---------------------------------------------------------------------------
858 # HTTP/1.1 implementation (stdlib, zero extra dependencies)
859 # ---------------------------------------------------------------------------
860
861
862 def build_msign_header(
863 signing: SigningIdentity,
864 method: str,
865 url: str,
866 body_bytes: bytes | None,
867 ) -> str:
868 """Compute the MSign Authorization header for a request.
869
870 Exported at module level so callers outside :class:`HttpTransport` (e.g.
871 the curl-based Darwin workaround in ``muse push``) can sign requests without
872 instantiating a full transport.
873 """
874 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
875
876 parsed = urllib.parse.urlparse(url)
877 path_with_query = parsed.path
878 if parsed.query:
879 path_with_query = f"{path_with_query}?{parsed.query}"
880
881 ts = int(time.time())
882 body = body_bytes or b""
883 body_hash = hashlib.sha256(body).hexdigest()
884 canonical = f"{method}\n{path_with_query}\n{ts}\n{body_hash}".encode()
885
886 private_key = signing.private_key
887 assert isinstance(private_key, Ed25519PrivateKey)
888 sig_bytes = private_key.sign(canonical)
889 sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode("ascii")
890
891 return f'MSign handle="{signing.handle}" ts={ts} sig="{sig_b64}"'
892
893
894 class HttpTransport:
895 """Synchronous HTTPS transport using stdlib ``urllib.request``.
896
897 One short-lived HTTPS connection per CLI invocation over HTTP/1.1.
898 Signing identity values are **never** written to any log line.
899 """
900
901 @staticmethod
902 def _build_msign_header(
903 signing: SigningIdentity,
904 method: str,
905 url: str,
906 body_bytes: bytes | None,
907 ) -> str:
908 return build_msign_header(signing, method, url, body_bytes)
909
910 def _build_request(
911 self,
912 method: str,
913 url: str,
914 signing: SigningIdentity | None,
915 body_bytes: bytes | None = None,
916 content_type: str = "application/x-msgpack",
917 ) -> urllib.request.Request:
918 # Never send credentials over cleartext HTTP — signatures would be visible
919 # to any network observer.
920 # Localhost (127.0.0.1, ::1, "localhost") is exempt because the traffic
921 # never leaves the machine and TLS would require a self-signed cert.
922 # "host.docker.internal" is the Docker Desktop alias for the host's
923 # loopback interface — traffic never leaves the machine, identical
924 # security posture to "localhost". Exempt it so agent swarms running
925 # inside Docker containers can reach a local MuseHub over plain HTTP.
926 _parsed = urllib.parse.urlparse(url)
927 _is_loopback = _parsed.hostname in {
928 "localhost", "127.0.0.1", "::1", "host.docker.internal",
929 }
930 if signing and _parsed.scheme != "https" and not _is_loopback:
931 raise TransportError(
932 f"Refusing to send credentials to a non-HTTPS URL: {url!r}. "
933 "Ensure the remote URL uses https://.",
934 0,
935 )
936
937 # TOFU hub certificate fingerprint pinning.
938 # Only runs when signing credentials are present — the pinning protects
939 # credentials from being sent to the wrong host. Unauthenticated requests
940 # (public reads, anonymous API calls) are not credential-bearing so the
941 # check is skipped to avoid blocking public access on fingerprint state.
942 if not _is_loopback and signing is not None:
943 try:
944 from muse.core.hub_trust import check_and_pin
945 # Derive the base URL (scheme + netloc only) for pinning.
946 base_url = f"{_parsed.scheme}://{_parsed.netloc}"
947 check_and_pin(base_url)
948 except Exception as _hub_exc: # noqa: BLE001
949 from muse.core.errors import HubFingerprintMismatchError
950 if isinstance(_hub_exc, HubFingerprintMismatchError):
951 raise TransportError(str(_hub_exc), 0) from _hub_exc
952 # Other errors (network, file I/O) are non-fatal: log and continue.
953 logger.warning("⚠️ Hub trust check failed: %s", _hub_exc)
954
955 # Advertise msgpack support so the server can respond in binary.
956 headers: _HeadersDict = {
957 "Accept": "application/x-msgpack, application/json",
958 }
959 if body_bytes is not None:
960 headers["Content-Type"] = content_type
961 if signing:
962 headers["Authorization"] = self._build_msign_header(signing, method, url, body_bytes)
963 return urllib.request.Request(
964 url=url,
965 data=body_bytes,
966 headers=headers,
967 method=method,
968 )
969
970 @staticmethod
971 def _decode(raw: bytes) -> _MsgDict:
972 """Decode a msgpack server response into a plain dict.
973
974 Raises :class:`TransportError` if the payload is not valid msgpack.
975 """
976 if not raw:
977 return {}
978 try:
979 # MAX_RESPONSE_BYTES already caps the network read — safe_unpackb
980 # adds per-field limits as a second layer against billion-laughs payloads.
981 # allow_binary=True because pack/bundle responses carry raw blob content.
982 result: _MsgVal = safe_unpackb(raw, context="server response", allow_binary=True)
983 except Exception as exc:
984 raise TransportError(f"Server returned invalid msgpack: {exc}", 0) from exc
985 if not isinstance(result, dict):
986 return {}
987 return result
988
989 def _execute(self, req: urllib.request.Request) -> bytes:
990 """Send *req* and return raw response bytes.
991
992 Uses :data:`_STRICT_OPENER` which refuses all HTTP redirects, ensuring
993 ``Authorization`` headers are never forwarded to an unintended host or
994 sent over a downgraded cleartext connection.
995
996 Raises:
997 :class:`TransportError` on non-2xx HTTP or any network error.
998 """
999 try:
1000 with _open_url(req, _TIMEOUT_SECONDS) as resp:
1001 # Enforce a hard cap before reading the body to defend against
1002 # a malicious or compromised server sending an unbounded response.
1003 content_length_str = resp.headers.get("Content-Length", "")
1004 if content_length_str:
1005 try:
1006 declared = int(content_length_str)
1007 if declared > MAX_RESPONSE_BYTES:
1008 raise TransportError(
1009 f"Server Content-Length {declared} exceeds the "
1010 f"{MAX_RESPONSE_BYTES // (1024 * 1024)} MiB response cap.",
1011 0,
1012 )
1013 except ValueError:
1014 pass # Unparseable Content-Length — fall through to streaming cap.
1015 # Read one byte more than the cap so we can detect over-limit responses.
1016 body: bytes = resp.read(MAX_RESPONSE_BYTES + 1)
1017 if len(body) > MAX_RESPONSE_BYTES:
1018 raise TransportError(
1019 f"Response body exceeds the {MAX_RESPONSE_BYTES // (1024 * 1024)} MiB "
1020 "cap. The server may be sending unexpected data.",
1021 0,
1022 )
1023 return body
1024 except urllib.error.HTTPError as exc:
1025 raise TransportError(_http_error_message(exc), exc.code) from exc
1026 except urllib.error.URLError as exc:
1027 raise TransportError(sanitize_display(str(exc.reason)), 0) from exc
1028
1029 def _execute_fetch(self, req: urllib.request.Request) -> bytes:
1030 """Send *req* and stream the response body without a Content-Length cap.
1031
1032 This is the correct transport for ``fetch`` and ``clone`` operations,
1033 which use ``Transfer-Encoding: chunked`` (no Content-Length) — the
1034 same pattern as Git's smart HTTP pack protocol. The server never
1035 declares a total size; instead the client reads until EOF, accumulating
1036 chunks.
1037
1038 Security model (identical to ``_execute`` for small endpoints):
1039
1040 - Refuses all HTTP redirects (prevents credential leakage).
1041 - No Content-Length cap — the server intentionally omits that header to
1042 support arbitrarily large repositories.
1043 - Hard total-body cap of ``MAX_FETCH_BYTES`` (2 GiB) — a single fetch
1044 that exceeds this almost certainly indicates a misconfigured server
1045 or an attack, not a legitimate clone.
1046 - 64 KiB read chunks keep the read loop tight and let the OS buffer
1047 incoming data efficiently.
1048 """
1049 import time
1050 _CHUNK = 64 * 1024 # 64 KiB — matches typical TCP window size
1051 t0 = time.perf_counter()
1052 try:
1053 with _open_url(req, _TIMEOUT_SECONDS) as resp:
1054 chunks: list[bytes] = []
1055 total = 0
1056 while True:
1057 chunk: bytes = resp.read(_CHUNK)
1058 if not chunk:
1059 break
1060 total += len(chunk)
1061 if total > MAX_FETCH_BYTES:
1062 gib = MAX_FETCH_BYTES // (1024 ** 3)
1063 raise TransportError(
1064 f"Fetch response exceeds the {gib} GiB cap. "
1065 "The server may be misconfigured.",
1066 0,
1067 )
1068 chunks.append(chunk)
1069 elapsed = time.perf_counter() - t0
1070 logger.debug(
1071 "transport: _execute_fetch %s → %d bytes in %.3fs",
1072 req.full_url, total, elapsed,
1073 )
1074 return b"".join(chunks)
1075 except urllib.error.HTTPError as exc:
1076 raise TransportError(_http_error_message(exc), exc.code) from exc
1077 except urllib.error.URLError as exc:
1078 raise TransportError(str(exc.reason), 0) from exc
1079
1080 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo:
1081 """Fetch repository metadata from ``GET {url}/refs``."""
1082 endpoint = f"{url.rstrip('/')}/refs"
1083 logger.debug("transport: GET %s", endpoint)
1084 req = self._build_request("GET", endpoint, signing)
1085 raw = self._execute(req)
1086 return _parse_remote_info(raw)
1087
1088 def fetch_pack(
1089 self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str]
1090 ) -> PackBundle:
1091 """Download VCS metadata via ``POST {url}/fetch`` (Phase 1 of two-phase fetch).
1092
1093 Returns commits, snapshots, and branch_heads only. Object bytes are
1094 fetched separately via :meth:`fetch_objects`.
1095 """
1096 endpoint = f"{url.rstrip('/')}/fetch"
1097 logger.debug(
1098 "transport: POST %s (want=%d, have=%d)", endpoint, len(want), len(have)
1099 )
1100 payload: FetchRequest = {"want": want, "have": have}
1101 body_bytes: bytes = msgpack.packb(payload, use_bin_type=True)
1102 req = self._build_request("POST", endpoint, signing, body_bytes)
1103 raw = self._execute_fetch(req)
1104 return _parse_bundle(raw)
1105
1106 def fetch_objects(
1107 self,
1108 url: str,
1109 signing: SigningIdentity | None,
1110 object_ids: list[str],
1111 ) -> list[ObjectPayload]:
1112 """Download object bytes via ``POST {url}/fetch/objects`` (Phase 2 of two-phase fetch).
1113
1114 Uses :meth:`_execute_fetch` so the chunked streaming response is read
1115 correctly without hitting the small-endpoint size cap.
1116 """
1117 if not object_ids:
1118 return []
1119 endpoint = f"{url.rstrip('/')}/fetch/objects"
1120 logger.debug("transport: POST %s (objects=%d)", endpoint, len(object_ids))
1121 body_bytes: bytes = msgpack.packb({"object_ids": object_ids}, use_bin_type=True)
1122 req = self._build_request("POST", endpoint, signing, body_bytes)
1123 raw = self._execute_fetch(req)
1124 bundle = _parse_bundle(raw)
1125 return bundle.get("objects") or []
1126
1127 def push_pack(
1128 self,
1129 url: str,
1130 signing: SigningIdentity | None,
1131 bundle: PackBundle,
1132 branch: str,
1133 force: bool,
1134 local_head: str | None = None,
1135 ) -> PushResult:
1136 """Upload a PackBundle via ``POST {url}/push``.
1137
1138 ``local_head`` is included so the server can advance the branch ref even
1139 when the bundle carries zero commits (all already present on the remote,
1140 e.g. the commit was previously pushed under a different branch).
1141 """
1142 endpoint = f"{url.rstrip('/')}/push"
1143 logger.debug(
1144 "transport: POST %s (branch=%s, force=%s, commits=%d, objects=%d)",
1145 endpoint,
1146 branch,
1147 force,
1148 len(bundle.get("commits") or []),
1149 len(bundle.get("objects") or []),
1150 )
1151 payload: _PushPayload = {"bundle": bundle, "branch": branch, "force": force}
1152 if local_head is not None:
1153 payload["local_head"] = local_head
1154 body_bytes: bytes = msgpack.packb(payload, use_bin_type=True)
1155 req = self._build_request("POST", endpoint, signing, body_bytes)
1156 raw = self._execute(req)
1157 return _parse_push_result(raw)
1158
1159 def push_objects(
1160 self,
1161 url: str,
1162 signing: SigningIdentity | None,
1163 objects: list[ObjectPayload],
1164 ) -> ObjectsChunkResponse:
1165 """Pre-upload an object batch via ``POST {url}/push/objects``."""
1166 endpoint = f"{url.rstrip('/')}/push/objects"
1167 logger.debug("transport: POST %s (objects=%d)", endpoint, len(objects))
1168 body_bytes: bytes = msgpack.packb({"objects": objects}, use_bin_type=True)
1169 req = self._build_request("POST", endpoint, signing, body_bytes)
1170 raw = self._execute(req)
1171 return _parse_objects_response(raw)
1172
1173 def push_object_pack(
1174 self,
1175 url: str,
1176 signing: SigningIdentity | None,
1177 objects: list[ObjectPayload],
1178 ) -> ObjectsChunkResponse:
1179 """Upload a pack of small objects via ``POST {url}/push/object-pack``.
1180
1181 Collapses N per-object TLS connections into a single msgpack POST.
1182 The server enforces ≤ 1 000 objects and ≤ 50 MB total per request.
1183 """
1184 endpoint = f"{url.rstrip('/')}/push/object-pack"
1185 logger.debug("transport: POST %s (pack=%d objects)", endpoint, len(objects))
1186 body_bytes: bytes = msgpack.packb({"objects": objects}, use_bin_type=True)
1187 req = self._build_request("POST", endpoint, signing, body_bytes)
1188 raw = self._execute(req)
1189 return _parse_objects_response(raw)
1190
1191 def filter_objects(
1192 self,
1193 url: str,
1194 signing: SigningIdentity | None,
1195 object_ids: list[str],
1196 object_hints: dict[str, str] | None = None,
1197 ) -> FilterObjectsResult:
1198 """Return missing object IDs and delta bases via ``POST {url}/filter-objects``."""
1199 endpoint = f"{url.rstrip('/')}/filter-objects"
1200 logger.debug("transport: POST %s (candidates=%d)", endpoint, len(object_ids))
1201 body: dict[str, list[str] | dict[str, str]] = {"object_ids": object_ids}
1202 if object_hints:
1203 body["object_hints"] = object_hints
1204 body_bytes: bytes = msgpack.packb(body, use_bin_type=True)
1205 req = self._build_request("POST", endpoint, signing, body_bytes)
1206 raw = self._execute(req)
1207 parsed = self._decode(raw)
1208 missing_raw = parsed.get("missing")
1209 missing = [m for m in missing_raw if isinstance(m, str)] if isinstance(missing_raw, list) else []
1210 bases_raw = parsed.get("bases")
1211 bases: dict[str, str] = (
1212 {k: v for k, v in bases_raw.items() if isinstance(k, str) and isinstance(v, str)}
1213 if isinstance(bases_raw, dict)
1214 else {}
1215 )
1216 return FilterObjectsResult(missing=missing, bases=bases)
1217
1218 def presign_objects(
1219 self,
1220 url: str,
1221 signing: SigningIdentity | None,
1222 object_ids: list[str],
1223 direction: str,
1224 ) -> PresignResponse:
1225 """Return presigned S3/R2 URLs via ``POST {url}/presign`` (MWP Phase 3)."""
1226 endpoint = f"{url.rstrip('/')}/presign"
1227 logger.debug(
1228 "transport: POST %s (objects=%d, direction=%s)", endpoint, len(object_ids), direction
1229 )
1230 body_bytes: bytes = msgpack.packb(
1231 {"object_ids": object_ids, "direction": direction}, use_bin_type=True
1232 )
1233 req = self._build_request("POST", endpoint, signing, body_bytes)
1234 raw = self._execute(req)
1235 parsed = self._decode(raw)
1236 presigned_raw = parsed.get("presigned", {})
1237 inline_raw = parsed.get("inline", [])
1238 presigned: Manifest = (
1239 {str(k): str(v) for k, v in presigned_raw.items()}
1240 if isinstance(presigned_raw, dict)
1241 else {}
1242 )
1243 inline: list[str] = (
1244 [str(x) for x in inline_raw] if isinstance(inline_raw, list) else []
1245 )
1246 return PresignResponse(presigned=presigned, inline=inline)
1247
1248 def confirm_objects(
1249 self,
1250 url: str,
1251 signing: SigningIdentity | None,
1252 object_ids: list[str],
1253 sizes: dict[str, int],
1254 ) -> ConfirmObjectsResponse:
1255 """Register presigned-uploaded objects via ``POST {url}/push/objects/confirm``."""
1256 endpoint = f"{url.rstrip('/')}/push/objects/confirm"
1257 logger.debug("transport: POST %s (objects=%d)", endpoint, len(object_ids))
1258 body_bytes: bytes = msgpack.packb(
1259 {"object_ids": object_ids, "sizes": sizes}, use_bin_type=True
1260 )
1261 req = self._build_request("POST", endpoint, signing, body_bytes)
1262 raw = self._execute(req)
1263 parsed = self._decode(raw)
1264 return ConfirmObjectsResponse(
1265 registered=int(parsed.get("registered", 0)),
1266 skipped=int(parsed.get("skipped", 0)),
1267 )
1268
1269 def negotiate(
1270 self,
1271 url: str,
1272 signing: SigningIdentity | None,
1273 want: list[str],
1274 have: list[str],
1275 ) -> NegotiateResponse:
1276 """Depth-limited commit negotiation via ``POST {url}/negotiate`` (MWP Phase 5)."""
1277 endpoint = f"{url.rstrip('/')}/negotiate"
1278 logger.debug(
1279 "transport: POST %s (want=%d, have=%d)", endpoint, len(want), len(have)
1280 )
1281 body_bytes: bytes = msgpack.packb({"want": want, "have": have}, use_bin_type=True)
1282 req = self._build_request("POST", endpoint, signing, body_bytes)
1283 raw = self._execute(req)
1284 parsed = self._decode(raw)
1285 ack_raw = parsed.get("ack", [])
1286 ack = [str(x) for x in ack_raw] if isinstance(ack_raw, list) else []
1287 common_base_raw = parsed.get("common_base")
1288 common_base = str(common_base_raw) if isinstance(common_base_raw, str) else None
1289 ready_raw = parsed.get("ready", False)
1290 return NegotiateResponse(
1291 ack=ack,
1292 common_base=common_base,
1293 ready=bool(ready_raw),
1294 )
1295
1296 def push_tags(
1297 self,
1298 url: str,
1299 signing: SigningIdentity | None,
1300 tags: list[WireTag],
1301 ) -> int:
1302 """Push tags via ``POST {url}/tags``."""
1303 endpoint = f"{url.rstrip('/')}/tags"
1304 logger.debug("transport: POST %s (tags=%d)", endpoint, len(tags))
1305 body_bytes: bytes = msgpack.packb({"tags": list(tags)}, use_bin_type=True)
1306 req = self._build_request("POST", endpoint, signing, body_bytes)
1307 raw = self._execute(req)
1308 parsed = self._decode(raw)
1309 stored_val = parsed.get("stored")
1310 return int(stored_val) if isinstance(stored_val, int) else 0
1311
1312 def create_release(
1313 self,
1314 url: str,
1315 signing: SigningIdentity | None,
1316 release: ReleaseDict,
1317 ) -> str:
1318 """Create a release via ``POST {url}/releases``."""
1319 endpoint = f"{url.rstrip('/')}/releases"
1320 logger.debug("transport: POST %s (tag=%s)", endpoint, release.get("tag", ""))
1321 # ReleaseDict contains no bytes fields so JSON encoding works directly.
1322 body_bytes: bytes = json.dumps(release).encode("utf-8")
1323 req = self._build_request("POST", endpoint, signing, body_bytes, content_type="application/json")
1324 raw = self._execute(req)
1325 parsed = self._decode(raw)
1326 release_id_val = parsed.get("release_id")
1327 return str(release_id_val) if isinstance(release_id_val, str) else ""
1328
1329 def list_releases_remote(
1330 self,
1331 url: str,
1332 signing: SigningIdentity | None,
1333 channel: str | None = None,
1334 include_drafts: bool = False,
1335 ) -> list[ReleaseDict]:
1336 """List releases via ``GET {url}/releases``."""
1337 qs_parts: list[str] = []
1338 if channel:
1339 qs_parts.append(f"channel={urllib.parse.quote(channel)}")
1340 if include_drafts:
1341 qs_parts.append("include_drafts=1")
1342 endpoint = f"{url.rstrip('/')}/releases"
1343 if qs_parts:
1344 endpoint = f"{endpoint}?{'&'.join(qs_parts)}"
1345 logger.debug("transport: GET %s", endpoint)
1346 req = self._build_request("GET", endpoint, signing)
1347 raw = self._execute(req)
1348 parsed = self._decode(raw)
1349 return _parse_releases_list(parsed)
1350
1351 def delete_release_remote(
1352 self,
1353 url: str,
1354 signing: SigningIdentity | None,
1355 tag: str,
1356 ) -> None:
1357 """Retract a release via ``DELETE {url}/releases/{tag}``."""
1358 endpoint = f"{url.rstrip('/')}/releases/{urllib.parse.quote(tag, safe='')}"
1359 logger.debug("transport: DELETE %s", endpoint)
1360 req = self._build_request("DELETE", endpoint, signing)
1361 self._execute(req)
1362
1363 def delete_branch_remote(
1364 self,
1365 url: str,
1366 signing: SigningIdentity | None,
1367 branch: str,
1368 ) -> None:
1369 """Delete a branch via ``DELETE {url}/branches/{branch}``."""
1370 endpoint = f"{url.rstrip('/')}/branches/{urllib.parse.quote(branch, safe='')}"
1371 logger.debug("transport: DELETE %s", endpoint)
1372 req = self._build_request("DELETE", endpoint, signing)
1373 self._execute(req)
1374
1375
1376 # ---------------------------------------------------------------------------
1377 # Response parsers — JSON bytes → typed TypedDicts
1378 # ---------------------------------------------------------------------------
1379 # json.loads() returns Any (per typeshed), so we use isinstance narrowing
1380 # throughout. No explicit Any annotations appear in this file.
1381 # ---------------------------------------------------------------------------
1382
1383
1384
1385 def _parse_remote_info(raw: bytes) -> RemoteInfo:
1386 """Parse ``GET /refs`` response bytes into a :class:`~muse.core.pack.RemoteInfo`."""
1387 parsed = HttpTransport._decode(raw)
1388 repo_id_val = parsed.get("repo_id")
1389 domain_val = parsed.get("domain")
1390 default_branch_val = parsed.get("default_branch")
1391 branch_heads_raw = parsed.get("branch_heads")
1392 branch_heads: BranchHeads = {}
1393 if isinstance(branch_heads_raw, dict):
1394 for k, v in branch_heads_raw.items():
1395 if isinstance(k, str) and isinstance(v, str):
1396 branch_heads[k] = v
1397 info = RemoteInfo(
1398 repo_id=str(repo_id_val) if isinstance(repo_id_val, str) else "",
1399 domain=str(domain_val) if isinstance(domain_val, str) else "midi",
1400 default_branch=(
1401 str(default_branch_val) if isinstance(default_branch_val, str) else "main"
1402 ),
1403 branch_heads=branch_heads,
1404 )
1405 pack_origin_val = parsed.get("pack_origin")
1406 if isinstance(pack_origin_val, str) and pack_origin_val.strip():
1407 info["pack_origin"] = pack_origin_val.strip()
1408 return info
1409
1410
1411 def _parse_bundle(raw: bytes) -> PackBundle:
1412 """Parse ``POST /fetch`` response bytes into a :class:`~muse.core.pack.PackBundle`."""
1413 parsed = HttpTransport._decode(raw)
1414 bundle: PackBundle = {}
1415
1416 # Commits — each item is a raw dict that CommitRecord.from_dict() will validate.
1417 commits_raw = parsed.get("commits")
1418 if isinstance(commits_raw, list):
1419 commits: list[CommitDict] = []
1420 for item in commits_raw:
1421 if isinstance(item, dict):
1422 commits.append(_coerce_commit_dict(item))
1423 bundle["commits"] = commits
1424
1425 # Snapshots
1426 snapshots_raw = parsed.get("snapshots")
1427 if isinstance(snapshots_raw, list):
1428 snapshots: list[SnapshotDict] = []
1429 for item in snapshots_raw:
1430 if isinstance(item, dict):
1431 snapshots.append(_coerce_snapshot_dict(item))
1432 bundle["snapshots"] = snapshots
1433
1434 # Objects — raw bytes in "content" (MWP msgpack wire format)
1435 objects_raw = parsed.get("objects")
1436 if isinstance(objects_raw, list):
1437 objects: list[ObjectPayload] = []
1438 for item in objects_raw:
1439 if not isinstance(item, dict):
1440 continue
1441 oid = item.get("object_id")
1442 if not isinstance(oid, str):
1443 continue
1444 content_raw = item.get("content")
1445 if isinstance(content_raw, (bytes, bytearray)):
1446 objects.append(ObjectPayload(object_id=oid, content=bytes(content_raw)))
1447 bundle["objects"] = objects
1448
1449 # Branch heads
1450 heads_raw = parsed.get("branch_heads")
1451 if isinstance(heads_raw, dict):
1452 branch_heads: BranchHeads = {}
1453 for k, v in heads_raw.items():
1454 if isinstance(k, str) and isinstance(v, str):
1455 branch_heads[k] = v
1456 bundle["branch_heads"] = branch_heads
1457
1458 return bundle
1459
1460
1461 def _parse_push_result(raw: bytes) -> PushResult:
1462 """Parse ``POST /push`` response bytes into a :class:`~muse.core.pack.PushResult`."""
1463 parsed = HttpTransport._decode(raw)
1464 ok_val = parsed.get("ok")
1465 msg_val = parsed.get("message")
1466 heads_raw = parsed.get("branch_heads")
1467 branch_heads: BranchHeads = {}
1468 if isinstance(heads_raw, dict):
1469 for k, v in heads_raw.items():
1470 if isinstance(k, str) and isinstance(v, str):
1471 branch_heads[k] = v
1472 return PushResult(
1473 ok=bool(ok_val) if isinstance(ok_val, bool) else False,
1474 message=str(msg_val) if isinstance(msg_val, str) else "",
1475 branch_heads=branch_heads,
1476 )
1477
1478
1479 def _parse_objects_response(raw: bytes) -> ObjectsChunkResponse:
1480 """Parse ``POST /push/objects`` response into an :class:`~muse.core.pack.ObjectsChunkResponse`."""
1481 parsed = HttpTransport._decode(raw)
1482 stored_val = parsed.get("stored")
1483 skipped_val = parsed.get("skipped")
1484 return ObjectsChunkResponse(
1485 stored=int(stored_val) if isinstance(stored_val, int) else 0,
1486 skipped=int(skipped_val) if isinstance(skipped_val, int) else 0,
1487 )
1488
1489
1490 def _coerce_sem_ver_bump(raw: _MsgVal) -> SemVerBump:
1491 """Safely coerce a raw value to a :class:`~muse.domain.SemVerBump` literal."""
1492 if raw == "major":
1493 return "major"
1494 if raw == "minor":
1495 return "minor"
1496 if raw == "patch":
1497 return "patch"
1498 return "none"
1499
1500
1501 def _parse_releases_list(parsed: _MsgDict) -> list[ReleaseDict]:
1502 """Extract a list of :class:`~muse.core.store.ReleaseDict` from a parsed response."""
1503 releases_raw = parsed.get("releases")
1504 if not isinstance(releases_raw, list):
1505 return []
1506 results: list[ReleaseDict] = []
1507 for item in releases_raw:
1508 if not isinstance(item, dict):
1509 continue
1510 try:
1511 semver_raw = item.get("semver")
1512 if isinstance(semver_raw, dict):
1513 sv_major = semver_raw.get("major")
1514 sv_minor = semver_raw.get("minor")
1515 sv_patch = semver_raw.get("patch")
1516 sv_pre = semver_raw.get("pre")
1517 sv_build = semver_raw.get("build")
1518 semver: SemVerTag = SemVerTag(
1519 major=int(sv_major) if isinstance(sv_major, int) else 0,
1520 minor=int(sv_minor) if isinstance(sv_minor, int) else 0,
1521 patch=int(sv_patch) if isinstance(sv_patch, int) else 0,
1522 pre=sv_pre if isinstance(sv_pre, str) else "",
1523 build=sv_build if isinstance(sv_build, str) else "",
1524 )
1525 else:
1526 semver = SemVerTag(major=0, minor=0, patch=0, pre="", build="")
1527 changelog_raw = item.get("changelog") or []
1528 changelog: list[ChangelogEntry] = []
1529 if isinstance(changelog_raw, list):
1530 for entry in changelog_raw:
1531 if not isinstance(entry, dict):
1532 continue
1533 bc_raw = entry.get("breaking_changes")
1534 bc_list: list[str] = [str(b) for b in bc_raw if isinstance(b, str)] if isinstance(bc_raw, list) else []
1535 changelog.append(ChangelogEntry(
1536 commit_id=str(entry.get("commit_id", "")),
1537 message=str(entry.get("message", "")),
1538 sem_ver_bump=_coerce_sem_ver_bump(entry.get("sem_ver_bump")),
1539 breaking_changes=bc_list,
1540 author=str(entry.get("author", "")),
1541 committed_at=str(entry.get("committed_at", "")),
1542 agent_id=str(entry.get("agent_id", "")),
1543 model_id=str(entry.get("model_id", "")),
1544 ))
1545 results.append(ReleaseDict(
1546 release_id=str(item.get("release_id", "")),
1547 repo_id=str(item.get("repo_id", "")),
1548 tag=str(item.get("tag", "")),
1549 semver=semver,
1550 channel=str(item.get("channel", "stable")),
1551 commit_id=str(item.get("commit_id", "")),
1552 snapshot_id=str(item.get("snapshot_id", "")),
1553 title=str(item.get("title", "")),
1554 body=str(item.get("body", "")),
1555 changelog=changelog,
1556 agent_id=str(item.get("agent_id", "")),
1557 model_id=str(item.get("model_id", "")),
1558 is_draft=bool(item.get("is_draft", False)),
1559 gpg_signature=str(item.get("gpg_signature", "")),
1560 created_at=str(item.get("created_at", "")),
1561 ))
1562 except (KeyError, TypeError, ValueError):
1563 continue
1564 return results
1565
1566
1567 # ---------------------------------------------------------------------------
1568 # TypedDict coercion helpers — extract known string fields from raw JSON dicts
1569 # ---------------------------------------------------------------------------
1570 # CommitDict and SnapshotDict are total=False (all fields optional), so we
1571 # only extract the string/scalar fields we can safely validate here.
1572 # CommitRecord.from_dict() and SnapshotRecord.from_dict() re-validate
1573 # required fields when apply_pack() calls them.
1574 # ---------------------------------------------------------------------------
1575
1576
1577 # _WireVal is now an alias for _MsgVal — kept for readability at call sites.
1578 _WireVal = _MsgVal
1579
1580
1581 def _str(val: _WireVal) -> str:
1582 """Return *val* as str, or empty string if not a str."""
1583 return val if isinstance(val, str) else ""
1584
1585
1586 def _str_or_none(val: _WireVal) -> str | None:
1587 """Return *val* as str, or None if not a str."""
1588 return val if isinstance(val, str) else None
1589
1590
1591 def _int_or(val: _WireVal, default: int) -> int:
1592 """Return *val* as int, or *default* if not an int."""
1593 return val if isinstance(val, int) else default
1594
1595
1596 def _coerce_commit_dict(raw: _WireDict) -> CommitDict:
1597 """Extract typed scalar fields from *raw* into a :class:`~muse.core.store.CommitDict`.
1598
1599 Only primitive fields are validated here; ``structured_delta`` is
1600 preserved as-is because :class:`~muse.core.store.CommitRecord.from_dict`
1601 already handles it gracefully.
1602 """
1603 metadata_raw = raw.get("metadata")
1604 metadata: Metadata = {}
1605 if isinstance(metadata_raw, dict):
1606 for k, v in metadata_raw.items():
1607 if isinstance(k, str) and isinstance(v, str):
1608 metadata[k] = v
1609
1610 reviewed_by_raw = raw.get("reviewed_by")
1611 reviewed_by: list[str] = []
1612 if isinstance(reviewed_by_raw, list):
1613 for item in reviewed_by_raw:
1614 if isinstance(item, str):
1615 reviewed_by.append(item)
1616
1617 breaking_changes_raw = raw.get("breaking_changes")
1618 breaking_changes: list[str] = []
1619 if isinstance(breaking_changes_raw, list):
1620 for item in breaking_changes_raw:
1621 if isinstance(item, str):
1622 breaking_changes.append(item)
1623
1624 sem_ver_raw = raw.get("sem_ver_bump")
1625 sem_ver: SemVerBump
1626 if sem_ver_raw == "major":
1627 sem_ver = "major"
1628 elif sem_ver_raw == "minor":
1629 sem_ver = "minor"
1630 elif sem_ver_raw == "patch":
1631 sem_ver = "patch"
1632 else:
1633 sem_ver = "none"
1634
1635 return CommitDict(
1636 commit_id=_str(raw.get("commit_id")),
1637 repo_id=_str(raw.get("repo_id")),
1638 branch=_str(raw.get("branch")),
1639 snapshot_id=_str(raw.get("snapshot_id")),
1640 message=_str(raw.get("message")),
1641 committed_at=_str(raw.get("committed_at")),
1642 parent_commit_id=_str_or_none(raw.get("parent_commit_id")),
1643 parent2_commit_id=_str_or_none(raw.get("parent2_commit_id")),
1644 author=_str(raw.get("author")),
1645 metadata=metadata,
1646 structured_delta=None,
1647 sem_ver_bump=sem_ver,
1648 breaking_changes=breaking_changes,
1649 agent_id=_str(raw.get("agent_id")),
1650 model_id=_str(raw.get("model_id")),
1651 toolchain_id=_str(raw.get("toolchain_id")),
1652 prompt_hash=_str(raw.get("prompt_hash")),
1653 signature=_str(raw.get("signature")),
1654 signer_key_id=_str(raw.get("signer_key_id")),
1655 format_version=_int_or(raw.get("format_version"), 1),
1656 reviewed_by=reviewed_by,
1657 test_runs=_int_or(raw.get("test_runs"), 0),
1658 )
1659
1660
1661 def _coerce_snapshot_dict(raw: _WireDict) -> SnapshotDict:
1662 """Extract typed fields from *raw* into a :class:`~muse.core.store.SnapshotDict`."""
1663 manifest_raw = raw.get("manifest")
1664 manifest: Manifest = {}
1665 if isinstance(manifest_raw, dict):
1666 for k, v in manifest_raw.items():
1667 if isinstance(k, str) and isinstance(v, str):
1668 manifest[k] = v
1669 # ``directories`` is hashed into snapshot_id — dropping it produces a
1670 # SnapshotRecord whose snapshot_id never matches the recomputed hash,
1671 # making every snapshot with non-empty directories permanently unreadable.
1672 directories_raw = raw.get("directories")
1673 directories: list[str] = []
1674 if isinstance(directories_raw, list):
1675 for item in directories_raw:
1676 if isinstance(item, str):
1677 directories.append(item)
1678 return SnapshotDict(
1679 snapshot_id=_str(raw.get("snapshot_id")),
1680 manifest=manifest,
1681 directories=directories,
1682 created_at=_str(raw.get("created_at")),
1683 note=_str(raw.get("note", "")),
1684 )
1685
1686
1687 # ---------------------------------------------------------------------------
1688 # LocalFileTransport helpers
1689 # ---------------------------------------------------------------------------
1690
1691
1692 def _is_ancestor(
1693 candidate: str,
1694 from_commit: str,
1695 bundle_by_id: _CommitBundle,
1696 remote_root: pathlib.Path,
1697 max_depth: int = 100_000,
1698 ) -> bool:
1699 """Return True if *candidate* is an ancestor of (or equal to) *from_commit*.
1700
1701 Walks the commit graph BFS-style starting from *from_commit*, consulting
1702 *bundle_by_id* first (commits included in the push bundle) and falling back
1703 to the existing commits on disk in *remote_root* (commits already present).
1704
1705 This two-source walk is necessary because ``build_pack()`` excludes commits
1706 in the caller's ``have`` set from the bundle — those commits are already on
1707 disk at the remote and must be consulted directly.
1708
1709 Args:
1710 candidate: The commit ID to search for (typically the remote's current HEAD).
1711 from_commit: Starting point of the BFS walk (typically the new tip being pushed).
1712 bundle_by_id: Commits included in the push bundle, keyed by commit_id.
1713 remote_root: Root of the remote Muse repo (for reading pre-existing commits).
1714 max_depth: BFS depth cap — prevents unbounded walks on corrupt graphs.
1715
1716 Returns:
1717 ``True`` if *candidate* is reachable from *from_commit*, ``False`` otherwise.
1718 """
1719 from muse.core.store import read_commit as _rc
1720
1721 seen: set[str] = set()
1722 queue: list[str] = [from_commit]
1723 depth = 0
1724 while queue and depth < max_depth:
1725 cid = queue.pop(0)
1726 if cid in seen:
1727 continue
1728 seen.add(cid)
1729 if cid == candidate:
1730 return True
1731 # Prefer bundle for unwritten commits; fall back to remote store.
1732 parent1: str | None
1733 parent2: str | None
1734 if cid in bundle_by_id:
1735 bc = bundle_by_id[cid]
1736 p1_raw = bc.get("parent_commit_id")
1737 p2_raw = bc.get("parent2_commit_id")
1738 parent1 = p1_raw if isinstance(p1_raw, str) else None
1739 parent2 = p2_raw if isinstance(p2_raw, str) else None
1740 else:
1741 rec = _rc(remote_root, cid)
1742 if rec is None:
1743 depth += 1
1744 continue
1745 parent1 = rec.parent_commit_id
1746 parent2 = rec.parent2_commit_id
1747 if parent1 and parent1 not in seen:
1748 queue.append(parent1)
1749 if parent2 and parent2 not in seen:
1750 queue.append(parent2)
1751 depth += 1
1752 return False
1753
1754
1755 # ---------------------------------------------------------------------------
1756 # LocalFileTransport — push/pull between two repos on the same filesystem
1757 # ---------------------------------------------------------------------------
1758
1759
1760 class LocalFileTransport:
1761 """Transport implementation for ``file://`` URLs.
1762
1763 Allows ``muse push file:///path/to/repo`` and ``muse pull`` between two
1764 Muse repositories on the same filesystem (or a shared network mount)
1765 without requiring a MuseHub server.
1766
1767 The remote path must be the root of an initialised Muse repository — it
1768 must contain a ``.muse/`` directory. No separate "bare repo" format is
1769 required; Muse repositories are self-describing.
1770
1771 This transport never makes network calls. All operations are synchronous
1772 filesystem reads and writes, consistent with the rest of the Muse CLI.
1773 """
1774
1775 @staticmethod
1776 def _repo_root(url: str) -> pathlib.Path:
1777 """Extract and validate the filesystem path from a ``file://`` URL.
1778
1779 Security guarantees:
1780 - Rejects non-``file://`` schemes unconditionally.
1781 - Calls ``.resolve()`` to canonicalize the path, dereferencing all
1782 symlinks before any filesystem operations. A symlink at the URL
1783 target that points to a directory without a ``.muse/`` subdirectory
1784 is rejected — the check is on the resolved, canonical path, not the
1785 symlink itself.
1786 - Verifies that ``.muse/`` exists at the resolved root, preventing
1787 accidental pushes to arbitrary directories.
1788 """
1789 parsed = urllib.parse.urlparse(url)
1790 if parsed.scheme != "file":
1791 raise TransportError(
1792 f"LocalFileTransport requires a file:// URL, got: {url!r}", 0
1793 )
1794 # urllib.parse.urlparse on file:///abs/path gives netloc="" path="/abs/path".
1795 # On Windows file://C:/path gives netloc="" path="/C:/path" — strip leading
1796 # slash for Windows compatibility via pathlib.
1797 path_str = parsed.netloc + parsed.path
1798 # resolve() dereferences all symlinks and normalises ".." components.
1799 # This is the defence against symlink-based path escape attempts.
1800 root = pathlib.Path(path_str).resolve()
1801 if not (root / ".muse").is_dir():
1802 raise TransportError(
1803 f"Remote path {root!r} does not contain a .muse/ directory. "
1804 "Run 'muse init' in the target directory first.",
1805 404,
1806 )
1807 return root
1808
1809 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo: # noqa: ARG002
1810 """Read branch heads directly from the remote's ref files."""
1811 from muse.core.store import get_all_branch_heads
1812
1813 remote_root = self._repo_root(url)
1814 repo_json_path = remote_root / ".muse" / "repo.json"
1815 try:
1816 repo_data = json.loads(repo_json_path.read_text(encoding="utf-8"))
1817 except (OSError, json.JSONDecodeError) as exc:
1818 raise TransportError(f"Cannot read remote repo.json: {exc}", 0) from exc
1819
1820 repo_id = str(repo_data.get("repo_id", ""))
1821 domain = str(repo_data.get("domain", "midi"))
1822 default_branch = str(repo_data.get("default_branch", "main"))
1823
1824 branch_heads = get_all_branch_heads(remote_root)
1825
1826 return RemoteInfo(
1827 repo_id=repo_id,
1828 domain=domain,
1829 default_branch=default_branch,
1830 branch_heads=branch_heads,
1831 )
1832
1833 def fetch_pack(
1834 self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str] # noqa: ARG002
1835 ) -> PackBundle:
1836 """Build a PackBundle (metadata only) from the remote's local store."""
1837 from muse.core.pack import build_pack
1838
1839 remote_root = self._repo_root(url)
1840 have_set = set(have)
1841 bundle = build_pack(remote_root, commit_ids=want, have=list(have_set))
1842 # Two-phase protocol: strip objects from the metadata bundle.
1843 # LocalFileTransport.fetch_objects() reads them directly from the remote store.
1844 bundle_no_objects: PackBundle = {}
1845 if "commits" in bundle:
1846 bundle_no_objects["commits"] = bundle["commits"]
1847 if "snapshots" in bundle:
1848 bundle_no_objects["snapshots"] = bundle["snapshots"]
1849 if "tags" in bundle:
1850 bundle_no_objects["tags"] = bundle["tags"]
1851 if "branch_heads" in bundle:
1852 bundle_no_objects["branch_heads"] = bundle["branch_heads"]
1853 return bundle_no_objects
1854
1855 def fetch_objects(
1856 self,
1857 url: str,
1858 signing: SigningIdentity | None, # noqa: ARG002
1859 object_ids: list[str],
1860 ) -> list[ObjectPayload]:
1861 """Read object bytes directly from the remote's local object store."""
1862 from muse.core.object_store import read_object
1863
1864 remote_root = self._repo_root(url)
1865 result: list[ObjectPayload] = []
1866 for oid in object_ids:
1867 data = read_object(remote_root, oid)
1868 if data is not None:
1869 result.append(ObjectPayload(object_id=oid, content=data))
1870 return result
1871
1872 def push_pack(
1873 self,
1874 url: str,
1875 signing: SigningIdentity | None, # noqa: ARG002
1876 bundle: PackBundle,
1877 branch: str,
1878 force: bool,
1879 local_head: str | None = None, # noqa: ARG002
1880 ) -> PushResult:
1881 """Write a PackBundle directly into the remote's local store.
1882
1883 Security guarantees:
1884 - ``branch`` is validated with :func:`~muse.core.validation.validate_branch_name`
1885 before any I/O. Names containing path traversal components (`..`),
1886 null bytes, backslashes, or other forbidden characters are rejected.
1887 - The ref file path is further hardened with
1888 :func:`~muse.core.validation.contain_path`, which resolves symlinks
1889 and asserts the result stays inside ``.muse/refs/heads/``. A branch
1890 name that ``validate_branch_name`` would allow but that resolves
1891 outside the expected directory (e.g. via a pre-placed symlink) is
1892 rejected before any write occurs.
1893 - A fast-forward check prevents overwriting diverged remote history
1894 unless ``force=True`` is explicitly passed.
1895 """
1896 from muse.core.pack import apply_pack
1897 from muse.core.store import get_all_branch_heads, get_head_commit_id, write_branch_ref as _write_branch_ref
1898
1899 remote_root = self._repo_root(url)
1900
1901 try:
1902 validate_branch_name(branch)
1903 except ValueError as exc:
1904 return PushResult(ok=False, message=str(exc), branch_heads={})
1905
1906 # Determine the new tip for the branch.
1907 # Prefer an explicit branch_heads entry in the bundle (set by the push
1908 # command when it knows the local HEAD). Fall back to computing the
1909 # leaf commit — the commit in the bundle that is not referenced as a
1910 # parent of any other commit in the bundle, filtered to the branch.
1911 # This handles bundles produced by build_pack(), which does not
1912 # populate branch_heads.
1913 bundle_heads = bundle.get("branch_heads") or {}
1914 new_tip: str | None = bundle_heads.get(branch)
1915 if new_tip is None:
1916 bundle_commits_list = bundle.get("commits") or []
1917 all_parent_ids: set[str] = set()
1918 for bc in bundle_commits_list:
1919 pid = bc.get("parent_commit_id")
1920 if isinstance(pid, str):
1921 all_parent_ids.add(pid)
1922 pid2 = bc.get("parent2_commit_id")
1923 if isinstance(pid2, str):
1924 all_parent_ids.add(pid2)
1925 # Leaf = commit whose ID is not a parent of any other bundle commit.
1926 # Prefer commits whose branch field matches; otherwise take any leaf.
1927 leaves_for_branch = [
1928 bc["commit_id"]
1929 for bc in bundle_commits_list
1930 if bc.get("commit_id") not in all_parent_ids
1931 and bc.get("branch") == branch
1932 and isinstance(bc.get("commit_id"), str)
1933 ]
1934 any_leaves = [
1935 bc["commit_id"]
1936 for bc in bundle_commits_list
1937 if bc.get("commit_id") not in all_parent_ids
1938 and isinstance(bc.get("commit_id"), str)
1939 ]
1940 fallback: list[str | None] = [None]
1941 new_tip = (leaves_for_branch or any_leaves or fallback)[0]
1942
1943 # Fast-forward check: the remote's current HEAD for this branch must be
1944 # an ancestor of the tip commit in the bundle, unless --force is passed.
1945 if not force and new_tip:
1946 remote_tip = get_head_commit_id(remote_root, branch)
1947 if remote_tip and remote_tip != new_tip:
1948 # BFS from new_tip through bundle commits *and* existing remote
1949 # commits to find whether remote_tip is a reachable ancestor.
1950 # We cannot rely on bundle commits alone because build_pack()
1951 # excludes commits the receiver already has (the "have" set).
1952 bundle_by_id: _CommitBundle = {
1953 c["commit_id"]: c
1954 for c in (bundle.get("commits") or [])
1955 if isinstance(c.get("commit_id"), str)
1956 }
1957 if not _is_ancestor(remote_tip, new_tip, bundle_by_id, remote_root):
1958 return PushResult(
1959 ok=False,
1960 message=(
1961 f"Push rejected: remote branch '{branch}' has diverged. "
1962 "Pull and merge first, or use --force."
1963 ),
1964 branch_heads={},
1965 )
1966
1967 try:
1968 apply_pack(remote_root, bundle)
1969 except Exception as exc: # noqa: BLE001
1970 return PushResult(ok=False, message=f"Failed to apply pack: {exc}", branch_heads={})
1971
1972 # Update the remote branch ref to the new tip.
1973 # contain_path() resolves symlinks and asserts the result stays inside
1974 # .muse/refs/heads/ — defence-in-depth beyond validate_branch_name.
1975 if new_tip:
1976 heads_base = remote_root / ".muse" / "refs" / "heads"
1977 try:
1978 ref_path = contain_path(heads_base, branch)
1979 except ValueError as exc:
1980 return PushResult(
1981 ok=False,
1982 message=f"Rejected: branch ref path is unsafe — {exc}",
1983 branch_heads={},
1984 )
1985 _write_branch_ref(remote_root, branch, new_tip)
1986 logger.info("✅ local-transport: updated %s → %s", branch, new_tip[:8])
1987
1988 return PushResult(
1989 ok=True,
1990 message=f"local push to {url!r} succeeded",
1991 branch_heads=get_all_branch_heads(remote_root),
1992 )
1993
1994 def push_objects(
1995 self,
1996 url: str,
1997 signing: SigningIdentity | None, # noqa: ARG002
1998 objects: list[ObjectPayload],
1999 ) -> ObjectsChunkResponse:
2000 """Write objects directly into the remote's local object store.
2001
2002 Mirrors the server-side ``POST /push/objects`` behaviour: objects are
2003 content-addressed and idempotent — already-present objects are skipped.
2004 No branch refs are touched; only blob bytes are written.
2005 """
2006 from muse.core.object_store import write_object
2007
2008 remote_root = self._repo_root(url)
2009 stored = 0
2010 skipped = 0
2011 for obj in objects:
2012 oid = obj.get("object_id", "")
2013 raw = obj.get("content", b"")
2014 if not oid or not raw:
2015 continue
2016 if write_object(remote_root, oid, raw):
2017 stored += 1
2018 else:
2019 skipped += 1
2020 return ObjectsChunkResponse(stored=stored, skipped=skipped)
2021
2022 def push_object_pack(
2023 self,
2024 url: str,
2025 signing: SigningIdentity | None, # noqa: ARG002
2026 objects: list[ObjectPayload],
2027 ) -> ObjectsChunkResponse:
2028 """Write a pack of small objects directly into the remote's local object store.
2029
2030 Mirrors the server-side ``POST /push/object-pack`` behaviour. Objects
2031 are content-addressed and idempotent — already-present objects are skipped.
2032 No branch refs are touched; only blob bytes are written.
2033
2034 The ``local://`` backend has no TLS overhead so the pack/presign split
2035 does not matter here — both paths write to the same object store. This
2036 implementation is identical to ``push_objects`` and exists solely to
2037 satisfy the :class:`MuseTransport` protocol.
2038 """
2039 from muse.core.object_store import read_object, write_object
2040
2041 remote_root = self._repo_root(url)
2042 stored = 0
2043 skipped = 0
2044 for obj in objects:
2045 oid = obj.get("object_id", "")
2046 raw = obj.get("content", b"")
2047 if not oid or not raw:
2048 continue
2049 encoding = obj.get("encoding", "raw")
2050 if encoding == "zlib":
2051 raw = decompress_zlib(raw)
2052 elif encoding == "delta+zlib":
2053 base_id_val: str = obj.get("base_id", "") # type: ignore[assignment]
2054 base_bytes = read_object(remote_root, base_id_val) or b""
2055 raw = apply_delta(base_bytes, raw)
2056 if write_object(remote_root, oid, raw):
2057 stored += 1
2058 else:
2059 skipped += 1
2060 return ObjectsChunkResponse(stored=stored, skipped=skipped)
2061
2062 def filter_objects(
2063 self,
2064 url: str,
2065 signing: SigningIdentity | None, # noqa: ARG002
2066 object_ids: list[str],
2067 object_hints: dict[str, str] | None = None, # noqa: ARG002
2068 ) -> FilterObjectsResult:
2069 """Return object IDs missing from the remote's local store (MWP Phase 1).
2070
2071 The local-file transport does not perform delta base lookup — bases is
2072 always empty. Delta encoding is only profitable over a real network.
2073 """
2074 from muse.core.object_store import read_object
2075
2076 remote_root = self._repo_root(url)
2077 missing: list[str] = []
2078 for oid in object_ids:
2079 if read_object(remote_root, oid) is None:
2080 missing.append(oid)
2081 return FilterObjectsResult(missing=missing, bases={})
2082
2083 def presign_objects(
2084 self,
2085 url: str,
2086 signing: SigningIdentity | None, # noqa: ARG002
2087 object_ids: list[str],
2088 direction: str, # noqa: ARG002
2089 ) -> PresignResponse:
2090 """Local transport has no object storage backend — return all as inline."""
2091 return PresignResponse(presigned={}, inline=list(object_ids))
2092
2093 def confirm_objects(
2094 self,
2095 url: str, # noqa: ARG002
2096 signing: SigningIdentity | None, # noqa: ARG002
2097 object_ids: list[str], # noqa: ARG002
2098 sizes: dict[str, int], # noqa: ARG002
2099 ) -> ConfirmObjectsResponse:
2100 """No-op for local transport — local push_objects handles DB registration."""
2101 return ConfirmObjectsResponse(registered=0, skipped=0)
2102
2103 def negotiate(
2104 self,
2105 url: str,
2106 signing: SigningIdentity | None, # noqa: ARG002
2107 want: list[str],
2108 have: list[str],
2109 ) -> NegotiateResponse:
2110 """Commit negotiation against a local repo (MWP Phase 5)."""
2111 from muse.core.store import read_commit as _rc
2112
2113 remote_root = self._repo_root(url)
2114 have_set = set(have)
2115
2116 # Which of the client's have-IDs exist on the remote?
2117 ack = [cid for cid in have if _rc(remote_root, cid) is not None]
2118 ack_set = set(ack)
2119
2120 common_base: str | None = None
2121 for cid in want:
2122 # Walk parents looking for an acked ancestor.
2123 commit = _rc(remote_root, cid)
2124 if commit is None:
2125 continue
2126 for pid in filter(None, [commit.parent_commit_id, commit.parent2_commit_id]):
2127 if pid in ack_set:
2128 common_base = pid
2129 break
2130 if common_base:
2131 break
2132
2133 ready = common_base is not None or not have_set
2134 return NegotiateResponse(ack=ack, common_base=common_base, ready=ready)
2135
2136 def push_tags(
2137 self,
2138 url: str,
2139 signing: SigningIdentity | None, # noqa: ARG002
2140 tags: list[WireTag],
2141 ) -> int:
2142 """Write tags directly into the remote's local tag store."""
2143 from muse.core.store import TagDict, TagRecord, write_tag
2144
2145 remote_root = self._repo_root(url)
2146 stored = 0
2147 for wire_tag in tags:
2148 try:
2149 tag_record = TagRecord.from_dict(TagDict(
2150 tag_id=wire_tag["tag_id"],
2151 repo_id=wire_tag["repo_id"],
2152 commit_id=wire_tag["commit_id"],
2153 tag=wire_tag["tag"],
2154 created_at=wire_tag["created_at"],
2155 ))
2156 write_tag(remote_root, tag_record)
2157 stored += 1
2158 except (KeyError, ValueError) as exc:
2159 logger.warning("⚠️ local-transport push_tags: bad tag — %s", exc)
2160 return stored
2161
2162 def create_release(
2163 self,
2164 url: str,
2165 signing: SigningIdentity | None, # noqa: ARG002
2166 release: ReleaseDict,
2167 ) -> str:
2168 """Write a release directly into the remote's local release store."""
2169 from muse.core.store import ReleaseRecord, write_release
2170
2171 remote_root = self._repo_root(url)
2172 try:
2173 release_record = ReleaseRecord.from_dict(release)
2174 write_release(remote_root, release_record)
2175 return release_record.release_id
2176 except (KeyError, ValueError) as exc:
2177 raise TransportError(f"create_release: invalid release data — {exc}", 0) from exc
2178
2179 def list_releases_remote(
2180 self,
2181 url: str,
2182 signing: SigningIdentity | None, # noqa: ARG002
2183 channel: str | None = None,
2184 include_drafts: bool = False,
2185 ) -> list[ReleaseDict]:
2186 """Read releases from the remote's local release store."""
2187 from muse.core.semver import ReleaseChannel
2188 from muse.core.store import list_releases
2189
2190 remote_root = self._repo_root(url)
2191 repo_json_path = remote_root / ".muse" / "repo.json"
2192 try:
2193 repo_data = json.loads(repo_json_path.read_text(encoding="utf-8"))
2194 except (OSError, json.JSONDecodeError) as exc:
2195 raise TransportError(f"Cannot read remote repo.json: {exc}", 0) from exc
2196 repo_id = str(repo_data.get("repo_id", ""))
2197 _ChannelMap = dict[str, ReleaseChannel]
2198 _channel_map: _ChannelMap = {
2199 "stable": "stable", "beta": "beta", "alpha": "alpha", "nightly": "nightly",
2200 }
2201 channel_arg: ReleaseChannel | None = _channel_map.get(channel, None) if channel else None
2202 records = list_releases(remote_root, repo_id, channel=channel_arg, include_drafts=include_drafts)
2203 return [r.to_dict() for r in records]
2204
2205 def delete_release_remote(
2206 self,
2207 url: str,
2208 signing: SigningIdentity | None, # noqa: ARG002
2209 tag: str,
2210 ) -> None:
2211 """Delete a release record from the remote's local release store."""
2212 from muse.core.store import delete_release, get_release_for_tag
2213
2214 remote_root = self._repo_root(url)
2215 repo_json_path = remote_root / ".muse" / "repo.json"
2216 try:
2217 repo_data = json.loads(repo_json_path.read_text(encoding="utf-8"))
2218 except (OSError, json.JSONDecodeError) as exc:
2219 raise TransportError(f"Cannot read remote repo.json: {exc}", 0) from exc
2220 repo_id = str(repo_data.get("repo_id", ""))
2221 release = get_release_for_tag(remote_root, repo_id, tag)
2222 if release is None:
2223 raise TransportError(f"Release '{tag}' not found on remote.", 404)
2224 if not delete_release(remote_root, repo_id, release.release_id):
2225 raise TransportError(f"Failed to delete release '{tag}' on remote.", 0)
2226
2227 def delete_branch_remote(
2228 self,
2229 url: str,
2230 signing: SigningIdentity | None, # noqa: ARG002
2231 branch: str,
2232 ) -> None:
2233 """Delete a branch ref from the remote's local ref store."""
2234 remote_root = self._repo_root(url)
2235 ref_file = remote_root / ".muse" / "refs" / "heads" / branch
2236 if not ref_file.is_file():
2237 raise TransportError(f"Branch '{branch}' not found on remote.", 404)
2238 heads_dir = remote_root / ".muse" / "refs" / "heads"
2239 ref_file.unlink()
2240 # Prune empty parent directories (e.g. feat/ left behind after feat/my-thing).
2241 for parent in ref_file.parents:
2242 if parent == heads_dir:
2243 break
2244 try:
2245 parent.rmdir()
2246 except OSError:
2247 break
2248
2249
2250 # ---------------------------------------------------------------------------
2251 # Factory — select transport based on URL scheme
2252 # ---------------------------------------------------------------------------
2253
2254
2255 def make_transport(url: str) -> "HttpTransport | LocalFileTransport":
2256 """Return the appropriate transport for *url*.
2257
2258 - ``file://`` URLs → :class:`LocalFileTransport` (no server required)
2259 - All other URLs → :class:`HttpTransport` (requires MuseHub server)
2260
2261 Args:
2262 url: Remote repository URL.
2263
2264 Returns:
2265 A transport instance implementing :class:`MuseTransport`.
2266 """
2267 if urllib.parse.urlparse(url).scheme == "file":
2268 return LocalFileTransport()
2269 return HttpTransport()
2270
2271
2272 def negotiate_have(
2273 transport: "HttpTransport | LocalFileTransport",
2274 url: str,
2275 signing: SigningIdentity | None,
2276 want: list[str],
2277 all_local: list[str],
2278 ) -> list[str]:
2279 """Return the minimal have-list via MWP depth-limited negotiation.
2280
2281 Sends at most :data:`NEGOTIATE_DEPTH` recent ancestors per round. The
2282 server responds with which commits it already has and whether a common base
2283 was found (``ready``). Repeats until the server signals ready or all
2284 local commits are exhausted, then falls back to the full list so the
2285 subsequent fetch always succeeds.
2286
2287 This is the transport-layer half of the Muse Wire Protocol (MWP) commit
2288 negotiation step. Both ``muse fetch`` and ``muse pull`` delegate here to
2289 avoid duplicating the same negotiation loop.
2290
2291 Args:
2292 transport: An active transport (HTTP or local file).
2293 url: Remote repository URL.
2294 token: Optional auth token.
2295 want: Commit IDs the client wants from the remote.
2296 all_local: All local commit IDs, most-recent first.
2297
2298 Returns:
2299 A minimal ``have`` list for use in ``fetch_pack``.
2300 """
2301 if not all_local:
2302 return []
2303
2304 offset = 0
2305 while offset < len(all_local):
2306 batch = all_local[offset : offset + NEGOTIATE_DEPTH]
2307 resp = transport.negotiate(url, signing, want=want, have=batch)
2308 ack = resp["ack"]
2309 if resp["ready"]:
2310 return ack or batch
2311 offset += NEGOTIATE_DEPTH
2312
2313 # Exhausted all local commits without finding a common base — send full list.
2314 return all_local
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 32 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 32 days ago