"""Muse transport layer — HTTP and local-filesystem remote communication. The :class:`MuseTransport` Protocol defines the interface between the Muse CLI and a remote host. The CLI calls this Protocol; the implementation is chosen at runtime by :func:`make_transport` based on the URL scheme. Transport implementations -------------------------- :class:`HttpTransport` Synchronous HTTPS transport using stdlib ``urllib.request``. Used for ``https://`` remote URLs (MuseHub server required). Encodes all request bodies as ``msgpack`` (``Content-Type: application/x-msgpack``), which eliminates the 33 % base64 inflation of the old JSON+base64 protocol. :class:`LocalFileTransport` Zero-network transport for ``file://`` URLs. Reads and writes directly from the remote's ``.muse/`` directory on the local filesystem (or a shared network mount). No server is required — ideal for local testing, monorepo setups, and offline workflows. Use :func:`make_transport` instead of constructing either class directly — it inspects the URL scheme and returns the appropriate implementation. MWP — Muse Wire Protocol ------------------------------- All endpoints live under the remote repository URL. MWP introduces five new phases on top of the original refs/fetch/push trio: +-----------------------------------+----------------------------------------+ | Endpoint | Purpose | +===================================+========================================+ | GET {url}/refs | Pre-flight: branch heads + metadata | +-----------------------------------+----------------------------------------+ | POST {url}/filter-objects | Phase 1: dedup — missing object IDs | +-----------------------------------+----------------------------------------+ | POST {url}/presign | Phase 3: presigned S3/R2 PUT/GET URLs | +-----------------------------------+----------------------------------------+ | POST {url}/push/objects | Pre-upload object chunks (small objs) | +-----------------------------------+----------------------------------------+ | POST {url}/push | Phase 5: commit + snapshot push | +-----------------------------------+----------------------------------------+ | POST {url}/fetch | Download commit delta pack | +-----------------------------------+----------------------------------------+ | POST {url}/negotiate | Phase 5: depth-limited have/ack loop | +-----------------------------------+----------------------------------------+ Wire encoding ~~~~~~~~~~~~~ Request bodies are ``msgpack``-encoded dicts (``Content-Type: application/x-msgpack``). The ``Accept`` header advertises both ``msgpack`` and ``application/json`` so older servers can respond in JSON. Objects are transmitted as raw ``bytes`` in :class:`~muse.core.pack.ObjectPayload` (``content`` field) — no base64 encoding. Authentication -------------- All endpoints accept an ``Authorization: MSign handle="" ts= sig=""`` header for Ed25519 per-request signing. The signing identity (handle + private key) is read from ``~/.muse/identity.toml`` via :func:`muse.cli.config.get_signing_identity` and is **never** written to any log line. :class:`LocalFileTransport` ignores the ``signing`` argument — local repos do not require authentication (access is controlled by filesystem permissions). Error codes (HttpTransport) ---------------------------- 401 Unauthorized — invalid or missing signature 404 Not found — repo does not exist on the remote 409 Conflict — push rejected (non-fast-forward without ``--force``) 5xx Server error Security model -------------- HttpTransport: - Refuses all HTTP redirects (prevents credential leakage to other hosts). - Rejects non-HTTPS URLs when a token is present (prevents cleartext exposure). - Caps response bodies at ``MAX_RESPONSE_BYTES`` (64 MiB) to prevent OOM. LocalFileTransport: - Calls ``.resolve()`` on all filesystem paths (canonicalises symlinks and ``..`` components before any I/O). - Validates branch names with ``validate_branch_name`` (rejects null bytes, backslashes, consecutive dots, and other path-traversal primitives). - Guards ref-file writes with ``contain_path`` (defence-in-depth: asserts the computed path stays inside ``.muse/refs/heads/`` even after symlink resolution). - Never follows redirects, makes no network calls, and ignores the token arg. """ from __future__ import annotations import base64 import collections.abc import contextlib import hashlib import http.client import json import logging import pathlib import signal import threading import time import types import urllib.error import urllib.parse import urllib.request import urllib.response from typing import IO, TYPE_CHECKING, NamedTuple, Protocol, TypedDict, runtime_checkable if TYPE_CHECKING: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey class SigningIdentity(NamedTuple): """Ed25519 signing identity for MSign request authentication.""" handle: str # hub handle, e.g. "gabriel" private_key: Ed25519PrivateKey def build_msign_header( signing: SigningIdentity, method: str, url: str, body_bytes: bytes | None = None, ) -> str: """Return an ``Authorization: MSign ...`` header value for a request. Suitable for use in custom ``urllib.request.Request`` headers when the caller cannot use :class:`HttpTransport` directly. Args: signing: The signing identity (handle + Ed25519 private key). method: HTTP method (``"GET"``, ``"POST"``, etc.). url: Full request URL. body_bytes: Request body bytes, or ``None`` for empty body. Returns: The complete ``Authorization`` header value, e.g. ``MSign handle="gabriel" ts=1712345678 sig="..."`` """ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey parsed = urllib.parse.urlparse(url) path_with_query = parsed.path if parsed.query: path_with_query = f"{path_with_query}?{parsed.query}" ts = int(time.time()) body = body_bytes or b"" body_hash = hashlib.sha256(body).hexdigest() canonical = f"{method}\n{path_with_query}\n{ts}\n{body_hash}".encode() private_key = signing.private_key assert isinstance(private_key, Ed25519PrivateKey) sig_bytes = private_key.sign(canonical) sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode("ascii") return f'MSign handle="{signing.handle}" ts={ts} sig="{sig_b64}"' import msgpack from muse.core.compression import apply_delta, compress_zlib, compute_delta, decompress_zlib from muse.core.pack import ( FetchRequest, ObjectPayload, ObjectsChunkResponse, PackBundle, PushResult, RemoteInfo, WireTag, ) from muse.core.semver import ChangelogEntry, SemVerTag from muse.core.store import ( BranchHeads, CommitDict, Manifest, Metadata, MsgpackDict, ReleaseDict, SnapshotDict, safe_unpackb, ) from muse.core.validation import ( MAX_FETCH_BYTES, MAX_RESPONSE_BYTES, contain_path, sanitize_display, validate_branch_name, ) from muse.domain import SemVerBump logger = logging.getLogger(__name__) _TIMEOUT_SECONDS = 300 # Recursive type alias for msgpack-serializable values. # Covers every type that msgpack can encode/decode natively — no base64, # no `object`, no `Any`. Python 3.12+ `type` statement creates a proper # TypeAlias that mypy treats as a first-class recursive type. type _MsgVal = ( str | int | float | bool | bytes | None | list[_MsgVal] | dict[str, _MsgVal] ) type _MsgDict = dict[str, _MsgVal] # top-level msgpack response dict type _WireVal = _MsgVal # wire-protocol value (same shape) type _WireDict = dict[str, _WireVal] # wire-protocol response dict type _HeadersDict = dict[str, str] # HTTP request/response headers type _CommitBundle = dict[str, CommitDict] # commit_id → commit dict class FilterObjectsResult(TypedDict): """Structured result from MWP Phase 1 (``POST /filter-objects``). ``missing``: IDs the server does not yet have — client must upload these. ``bases``: Suggested delta base for each missing object (oid → base_oid). When present, the client should compute a ``delta+zlib``-encoded object instead of sending the raw bytes. """ missing: list[str] bases: dict[str, str] # Maximum number of objects to include in a single POST /push/objects call. # Must stay strictly below the server's MAX_OBJECTS_PER_PUSH limit (1 000). # 100 gives better pipelining than 400: smaller msgpack payloads, faster # MSign computation, shorter server-side processing windows, and lower lock # contention when thousands of agents push concurrently. # 500 objects per chunk keeps HTTP round-trips low (5 000 objects = 10 requests # instead of 50) while staying well within typical server request-body limits. CHUNK_OBJECTS: int = 500 # Maximum number of commits to include in a single POST /push call. # Large histories are split into sequential chunked pushes so that each # request stays well within Cloudflare's 100 MB upload limit. # Each intermediate chunk uses force=True; the final chunk uses the caller's # force flag. # # 500 commits/chunk means ~2 round-trips for a 1000-commit history vs the # previous 5 at 200/chunk. The server now uses bulk INSERT ON CONFLICT DO # NOTHING so per-chunk processing time is proportional to commit count, not # the number of individual INSERT round-trips. CHUNK_COMMITS: int = 500 # Objects above this byte threshold are candidates for presigned-URL upload # (MWP Phase 3) — they bypass the API server and go directly to S3/R2. # Objects at or below this size are included inline in the pack body. LARGE_OBJECT_THRESHOLD: int = 64 * 1024 # 64 KiB # Depth of the have-list sent per round of commit negotiation (MWP Phase 5). # Caps the negotiation payload at ≤ NEGOTIATE_DEPTH commit IDs per request, # keeping negotiation O(depth) rather than O(history). NEGOTIATE_DEPTH: int = 64 # --------------------------------------------------------------------------- # MWP response TypedDicts # --------------------------------------------------------------------------- class PresignResponse(TypedDict): """Response from ``POST {url}/presign`` — MWP Phase 3.""" presigned: Manifest # object_id → presigned URL inline: list[str] # IDs whose backend does not support presigned URLs class ConfirmObjectsResponse(TypedDict): """Response from ``POST {url}/push/objects/confirm`` — MWP Phase 3b.""" registered: int # objects newly registered in the remote DB skipped: int # objects already registered (idempotent no-op) class NegotiateResponse(TypedDict): """Response from ``POST {url}/negotiate`` — MWP Phase 5.""" ack: list[str] common_base: str | None ready: bool class _PushPayload(TypedDict, total=False): """Msgpack-encoded body for ``POST {url}/push``.""" bundle: PackBundle branch: str force: bool local_head: str # --------------------------------------------------------------------------- # Response protocol — typed adapter for urllib response objects # --------------------------------------------------------------------------- class _HttpHeaders(Protocol): """Minimal interface for HTTP response headers.""" def get(self, name: str, default: str = "") -> str: ... @runtime_checkable class _HttpResponse(Protocol): """Structural interface for urllib HTTP response objects. Defined as a Protocol so that ``_open_url`` can have a concrete, non-Any return type without importing implementation-specific urllib internals. """ headers: _HttpHeaders def read(self, amt: int | None = None) -> bytes: ... def __enter__(self) -> "_HttpResponse": ... def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: "types.TracebackType | None", ) -> bool | None: ... # --------------------------------------------------------------------------- # Security — credential-safe HTTP error formatting # --------------------------------------------------------------------------- def _http_error_message(exc: urllib.error.HTTPError) -> str: """Build a safe, sanitized error message for an HTTP error response. HTTP 401 responses are especially dangerous: the server received the ``Authorization: MSign`` header, and a malicious or compromised server may echo credential data verbatim in the response body. We NEVER include the response body for 401 — instead we emit a generic "Authentication failed" string that contains no server-controlled content. For all other status codes we include the server error body (capped at 200 characters) after stripping terminal control characters (ANSI escape sequences, OSC sequences, etc.) to block terminal injection. """ if exc.code == 401: return "Authentication failed (HTTP 401). Run 'muse auth register'." try: raw_body: str = exc.read().decode("utf-8", errors="replace") except Exception: # noqa: BLE001 raw_body = "" safe_body = sanitize_display(raw_body[:200]) return f"HTTP {exc.code}: {safe_body}" if safe_body else f"HTTP {exc.code}" # --------------------------------------------------------------------------- # Security — redirect and scheme enforcement # --------------------------------------------------------------------------- class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): """Refuse all HTTP redirects. ``urllib.request`` follows redirects by default, including across schemes (HTTPS → HTTP) and across hosts. If a server we contact redirects us: - To HTTP: the ``Authorization: MSign`` header would be sent in cleartext. - To a different host: the token would be sent to an unintended recipient. We refuse both. The server must use the correct, stable URL. If a redirect is required during operations, it is always better to surface it as a hard error so the operator can update the configured URL than to silently follow it and leak credentials. """ def redirect_request( self, req: urllib.request.Request, fp: IO[bytes], code: int, msg: str, headers: http.client.HTTPMessage, newurl: str, ) -> urllib.request.Request | None: raise urllib.error.HTTPError( req.full_url, code, ( f"Redirect refused ({code}): server tried to redirect to {newurl!r}. " "Update the configured remote URL to the final destination." ), headers, fp, ) # Build one opener that never follows redirects. Used for every request # so that Authorization headers are never sent to an unintended recipient. _STRICT_OPENER = urllib.request.build_opener(_NoRedirectHandler()) @contextlib.contextmanager def _ignore_sigpipe() -> collections.abc.Generator[None, None, None]: """Temporarily ignore SIGPIPE during socket I/O (main thread only). ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` at startup so that piping ``muse`` output to ``head``/``grep``/``jq`` exits cleanly. But ``SIG_DFL`` also kills the process when a large HTTP request body is still in-flight and the remote closes the connection early (auth failure, 409 conflict, server crash, …). Without this guard the push command dies with exit code 141 instead of raising ``TransportError``. Signal handlers may only be changed from the main interpreter thread. Worker threads launched by ``ThreadPoolExecutor`` (used for parallel object uploads) are already safe: Python blocks ``SIGPIPE`` in non-main threads, so broken-socket writes raise ``BrokenPipeError`` rather than killing the process. This context manager is therefore a no-op in non-main threads. In the main thread the context manager saves the current SIGPIPE disposition, sets it to ``SIG_IGN`` for the duration of the network call (so that broken-pipe conditions surface as ``BrokenPipeError`` → urllib ``URLError`` → ``TransportError``), then restores the original disposition on exit. On platforms without ``SIGPIPE`` (Windows) this is a no-op. """ if not hasattr(signal, "SIGPIPE") or not threading.current_thread() is threading.main_thread(): yield return old_handler: signal.Handlers = signal.signal(signal.SIGPIPE, signal.SIG_IGN) try: yield finally: signal.signal(signal.SIGPIPE, old_handler) def _open_url(req: urllib.request.Request, timeout: int) -> _HttpResponse: """Thin wrapper around ``_STRICT_OPENER.open`` — exists purely to give tests a single, importable patch target instead of deep-patching the opener object. SIGPIPE is temporarily ignored for the duration of the call so that the process does not die with exit 141 when the server closes the connection while a large request body is still being sent. See ``_ignore_sigpipe``. Returns an ``_HttpResponse`` Protocol value so that callers can be fully typed without depending on urllib's concrete ``addinfourl`` class. """ with _ignore_sigpipe(): resp: _HttpResponse = _STRICT_OPENER.open(req, timeout=timeout) return resp # --------------------------------------------------------------------------- # Exception # --------------------------------------------------------------------------- class TransportError(Exception): """Raised when the remote returns a non-2xx response or is unreachable. Attributes: status_code: HTTP status code (e.g. ``401``, ``404``, ``409``, ``500``). ``0`` for network-level failures (DNS, connection refused). """ def __init__(self, message: str, status_code: int) -> None: super().__init__(message) self.status_code = status_code # --------------------------------------------------------------------------- # Protocol — the seam between CLI commands and the transport implementation # --------------------------------------------------------------------------- class MuseTransport(Protocol): """Protocol for Muse remote transport implementations. All methods are synchronous — the Muse CLI is synchronous by design. """ def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo: """Return repository metadata from ``GET {url}/refs``. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None`` for public repos. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def fetch_pack( self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str] ) -> PackBundle: """Download a :class:`~muse.core.pack.PackBundle` via ``POST {url}/fetch``. Phase 1 of the two-phase fetch protocol. Returns commits, snapshots, and branch_heads only — no object bytes. Call :meth:`fetch_objects` separately with the object IDs needed for checkout. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. want: Commit IDs the client wants to receive. have: Commit IDs already present locally. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def fetch_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], ) -> list[ObjectPayload]: """Download object bytes via ``POST {url}/fetch/objects``. Phase 2 of the two-phase fetch protocol. The caller passes only the IDs it is missing from its local store; the server streams back the bytes for each one. Args: url: Remote repository URL. signing: Ed25519 signing identity, or ``None``. object_ids: Object IDs whose bytes are needed locally. Returns: List of :class:`~muse.core.pack.ObjectPayload` with ``object_id`` and ``content``. IDs that do not exist on the remote are silently omitted. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def push_pack( self, url: str, signing: SigningIdentity | None, bundle: PackBundle, branch: str, force: bool, local_head: str | None = None, ) -> PushResult: """Upload a :class:`~muse.core.pack.PackBundle` via ``POST {url}/push``. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. bundle: Bundle to upload. branch: Remote branch to update. force: Bypass the server-side fast-forward check. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def push_objects( self, url: str, signing: SigningIdentity | None, objects: list[ObjectPayload], ) -> ObjectsChunkResponse: """Pre-upload a batch of content-addressed objects via ``POST {url}/push/objects``. This is Phase 1 of a chunked push. The caller splits the full object list into batches of at most :data:`CHUNK_OBJECTS` and calls this once per batch. After all batches succeed, the caller issues a single ``push_pack`` (Phase 2) with an empty ``objects`` list — the final push only carries commits and snapshots, which are small. Objects are idempotent: the server skips any it already holds and counts them in ``skipped``. Uploading the same object twice is always safe. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. objects: Batch of objects to upload (``len ≤ CHUNK_OBJECTS``). Returns: :class:`~muse.core.pack.ObjectsChunkResponse` with ``stored`` and ``skipped`` counts. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def push_object_pack( self, url: str, signing: SigningIdentity | None, objects: list[ObjectPayload], ) -> ObjectsChunkResponse: """Upload a pack of ≤ 1 000 small objects in a single ``POST {url}/push/object-pack``. The pack endpoint is the web-scale alternative to the presigned PUT path for small objects (≤ ``LARGE_OBJECT_THRESHOLD``, i.e. 1 MB). Instead of one TLS handshake per object, the client bundles objects into msgpack packs and POSTs them to the API server. Round-trips collapse from O(N) to O(ceil(N/1000)). For objects larger than ``LARGE_OBJECT_THRESHOLD``, the presigned URL path (``presign_objects`` + direct PUT) is still preferred — bypassing Cloudflare is worthwhile when the per-object body is large enough to saturate bandwidth. Objects are content-addressed and idempotent — re-sending an object that already exists on the remote is safe; it is counted in ``skipped``. Args: url: Remote repository URL. signing: Ed25519 signing identity, or ``None``. objects: Batch of objects to upload (``len ≤ PACK_MAX_OBJECTS = 1 000``, total bytes ≤ ``PACK_MAX_BYTES = 50 MB``). Returns: :class:`~muse.core.pack.ObjectsChunkResponse` with ``stored`` and ``skipped`` counts. Raises: :class:`TransportError` on HTTP 4xx/5xx, network failure, or if the remote server does not support the pack endpoint (404 → caller should fall back to ``push_objects``). """ ... def filter_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], object_hints: dict[str, str] | None = None, ) -> FilterObjectsResult: """Return missing object IDs and delta base suggestions (MWP Phase 1). The client sends every candidate object ID plus optional path hints. The server responds with the subset it is missing and, for each missing object whose path already exists in the repo, a suggested base object ID for delta encoding. Args: url: Remote repository URL. signing: Ed25519 signing identity, or ``None``. object_ids: All object IDs the client intends to upload. object_hints: ``{oid: path}`` map for delta base lookup (optional). Returns: :class:`FilterObjectsResult` with ``missing`` IDs and ``bases`` map. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def presign_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], direction: str, ) -> PresignResponse: """Return presigned S3/R2 URLs for direct-to-storage transfer. MWP Phase 3 — when the backend supports presigned URLs (S3/R2), ALL missing objects are presigned and uploaded directly to object storage, bypassing the API server and Cloudflare entirely. When the backend is ``local://`` all IDs are returned in ``inline`` and the client falls back to the normal ``push_objects`` pack flow. Args: url: Remote repository URL. signing: Ed25519 signing identity, or ``None``. object_ids: IDs of all missing objects to presign. direction: ``"put"`` for push, ``"get"`` for pull. Returns: :class:`PresignResponse` with ``presigned`` URL map and ``inline`` fallback list. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def confirm_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], sizes: dict[str, int], ) -> ConfirmObjectsResponse: """Register objects uploaded via presigned URL in the remote DB. MWP Phase 3b — after uploading directly to R2, the client calls this so the server can insert ``musehub_objects`` DB records. Without this step, future ``filter_objects`` calls would report those objects as missing and re-upload them on every push. Args: url: Remote repository URL. signing: Ed25519 signing identity, or ``None``. object_ids: IDs of objects the client just uploaded via presign. sizes: ``{object_id: byte_count}`` — supplied by client since the server never saw the bytes. Returns: :class:`ConfirmObjectsResponse` with ``registered`` and ``skipped``. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def negotiate( self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str], ) -> NegotiateResponse: """Depth-limited commit negotiation (MWP Phase 5). Replaces sending the full local commit list as ``have``. Each round sends ≤ :data:`NEGOTIATE_DEPTH` recent ancestors. The server responds with which it recognises (``ack``), the common base commit (if found), and whether ``ready`` — i.e. enough context to compute the delta. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. want: Branch tips the client wants to receive. have: Recent local commit IDs (≤ NEGOTIATE_DEPTH per round). Returns: :class:`NegotiateResponse` with ``ack``, ``common_base``, ``ready``. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def push_tags( self, url: str, signing: SigningIdentity | None, tags: list[WireTag], ) -> int: """Push local tags to the remote via ``POST {url}/tags``. Tags are immutable once created on the remote — the server skips any it already holds. Returns the number of tags newly stored. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. tags: Tags to push. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def create_release( self, url: str, signing: SigningIdentity | None, release: ReleaseDict, ) -> str: """Create a release on the remote via ``POST {url}/releases``. Returns the ``release_id`` assigned by the server. Args: url: Remote repository URL. token: Ed25519 signing identity. release: Fully-populated :class:`~muse.core.store.ReleaseDict`. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def list_releases_remote( self, url: str, signing: SigningIdentity | None, channel: str | None = None, include_drafts: bool = False, ) -> list[ReleaseDict]: """Fetch releases from the remote via ``GET {url}/releases``. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. channel: Filter by release channel; ``None`` returns all. include_drafts: Include draft releases when ``True``. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def delete_release_remote( self, url: str, signing: SigningIdentity | None, tag: str, ) -> None: """Retract a release from the remote via ``DELETE {url}/releases/{tag}``. Removes only the named label from the remote registry. The underlying commit and snapshot objects are **not** affected. Args: url: Remote repository URL. token: Ed25519 signing identity (owner credentials required). tag: Semver tag of the release to retract (e.g. ``"v1.2.0"``). Raises: :class:`TransportError` on HTTP 4xx/5xx, network failure, or if the release does not exist on the remote. """ ... def delete_branch_remote( self, url: str, signing: SigningIdentity | None, branch: str, ) -> None: """Delete a branch on the remote via ``DELETE {url}/branches/{branch}``. Equivalent to ``git push origin --delete ``. The branch ref is removed from the server; commits and objects are unaffected. Args: url: Remote repository URL. token: Ed25519 signing identity (owner credentials required). branch: Branch name to delete (e.g. ``"feat/my-thing"``). Raises: :class:`TransportError` on HTTP 4xx/5xx, network failure, or if the branch does not exist on the remote. """ ... # --------------------------------------------------------------------------- # HTTP/1.1 implementation (stdlib, zero extra dependencies) # --------------------------------------------------------------------------- def build_msign_header( signing: SigningIdentity, method: str, url: str, body_bytes: bytes | None, ) -> str: """Compute the MSign Authorization header for a request. Exported at module level so callers outside :class:`HttpTransport` (e.g. the curl-based Darwin workaround in ``muse push``) can sign requests without instantiating a full transport. """ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey parsed = urllib.parse.urlparse(url) path_with_query = parsed.path if parsed.query: path_with_query = f"{path_with_query}?{parsed.query}" ts = int(time.time()) body = body_bytes or b"" body_hash = hashlib.sha256(body).hexdigest() canonical = f"{method}\n{path_with_query}\n{ts}\n{body_hash}".encode() private_key = signing.private_key assert isinstance(private_key, Ed25519PrivateKey) sig_bytes = private_key.sign(canonical) sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode("ascii") return f'MSign handle="{signing.handle}" ts={ts} sig="{sig_b64}"' class HttpTransport: """Synchronous HTTPS transport using stdlib ``urllib.request``. One short-lived HTTPS connection per CLI invocation over HTTP/1.1. Signing identity values are **never** written to any log line. """ @staticmethod def _build_msign_header( signing: SigningIdentity, method: str, url: str, body_bytes: bytes | None, ) -> str: return build_msign_header(signing, method, url, body_bytes) def _build_request( self, method: str, url: str, signing: SigningIdentity | None, body_bytes: bytes | None = None, content_type: str = "application/x-msgpack", ) -> urllib.request.Request: # Never send credentials over cleartext HTTP — signatures would be visible # to any network observer. # Localhost (127.0.0.1, ::1, "localhost") is exempt because the traffic # never leaves the machine and TLS would require a self-signed cert. # "host.docker.internal" is the Docker Desktop alias for the host's # loopback interface — traffic never leaves the machine, identical # security posture to "localhost". Exempt it so agent swarms running # inside Docker containers can reach a local MuseHub over plain HTTP. _parsed = urllib.parse.urlparse(url) _is_loopback = _parsed.hostname in { "localhost", "127.0.0.1", "::1", "host.docker.internal", } if signing and _parsed.scheme != "https" and not _is_loopback: raise TransportError( f"Refusing to send credentials to a non-HTTPS URL: {url!r}. " "Ensure the remote URL uses https://.", 0, ) # TOFU hub certificate fingerprint pinning. # Only runs when signing credentials are present — the pinning protects # credentials from being sent to the wrong host. Unauthenticated requests # (public reads, anonymous API calls) are not credential-bearing so the # check is skipped to avoid blocking public access on fingerprint state. if not _is_loopback and signing is not None: try: from muse.core.hub_trust import check_and_pin # Derive the base URL (scheme + netloc only) for pinning. base_url = f"{_parsed.scheme}://{_parsed.netloc}" check_and_pin(base_url) except Exception as _hub_exc: # noqa: BLE001 from muse.core.errors import HubFingerprintMismatchError if isinstance(_hub_exc, HubFingerprintMismatchError): raise TransportError(str(_hub_exc), 0) from _hub_exc # Other errors (network, file I/O) are non-fatal: log and continue. logger.warning("⚠️ Hub trust check failed: %s", _hub_exc) # Advertise msgpack support so the server can respond in binary. headers: _HeadersDict = { "Accept": "application/x-msgpack, application/json", } if body_bytes is not None: headers["Content-Type"] = content_type if signing: headers["Authorization"] = self._build_msign_header(signing, method, url, body_bytes) return urllib.request.Request( url=url, data=body_bytes, headers=headers, method=method, ) @staticmethod def _decode(raw: bytes) -> _MsgDict: """Decode a msgpack server response into a plain dict. Raises :class:`TransportError` if the payload is not valid msgpack. """ if not raw: return {} try: # MAX_RESPONSE_BYTES already caps the network read — safe_unpackb # adds per-field limits as a second layer against billion-laughs payloads. # allow_binary=True because pack/bundle responses carry raw blob content. result: _MsgVal = safe_unpackb(raw, context="server response", allow_binary=True) except Exception as exc: raise TransportError(f"Server returned invalid msgpack: {exc}", 0) from exc if not isinstance(result, dict): return {} return result def _execute(self, req: urllib.request.Request) -> bytes: """Send *req* and return raw response bytes. Uses :data:`_STRICT_OPENER` which refuses all HTTP redirects, ensuring ``Authorization`` headers are never forwarded to an unintended host or sent over a downgraded cleartext connection. Raises: :class:`TransportError` on non-2xx HTTP or any network error. """ try: with _open_url(req, _TIMEOUT_SECONDS) as resp: # Enforce a hard cap before reading the body to defend against # a malicious or compromised server sending an unbounded response. content_length_str = resp.headers.get("Content-Length", "") if content_length_str: try: declared = int(content_length_str) if declared > MAX_RESPONSE_BYTES: raise TransportError( f"Server Content-Length {declared} exceeds the " f"{MAX_RESPONSE_BYTES // (1024 * 1024)} MiB response cap.", 0, ) except ValueError: pass # Unparseable Content-Length — fall through to streaming cap. # Read one byte more than the cap so we can detect over-limit responses. body: bytes = resp.read(MAX_RESPONSE_BYTES + 1) if len(body) > MAX_RESPONSE_BYTES: raise TransportError( f"Response body exceeds the {MAX_RESPONSE_BYTES // (1024 * 1024)} MiB " "cap. The server may be sending unexpected data.", 0, ) return body except urllib.error.HTTPError as exc: raise TransportError(_http_error_message(exc), exc.code) from exc except urllib.error.URLError as exc: raise TransportError(sanitize_display(str(exc.reason)), 0) from exc def _execute_fetch(self, req: urllib.request.Request) -> bytes: """Send *req* and stream the response body without a Content-Length cap. This is the correct transport for ``fetch`` and ``clone`` operations, which use ``Transfer-Encoding: chunked`` (no Content-Length) — the same pattern as Git's smart HTTP pack protocol. The server never declares a total size; instead the client reads until EOF, accumulating chunks. Security model (identical to ``_execute`` for small endpoints): - Refuses all HTTP redirects (prevents credential leakage). - No Content-Length cap — the server intentionally omits that header to support arbitrarily large repositories. - Hard total-body cap of ``MAX_FETCH_BYTES`` (2 GiB) — a single fetch that exceeds this almost certainly indicates a misconfigured server or an attack, not a legitimate clone. - 64 KiB read chunks keep the read loop tight and let the OS buffer incoming data efficiently. """ import time _CHUNK = 64 * 1024 # 64 KiB — matches typical TCP window size t0 = time.perf_counter() try: with _open_url(req, _TIMEOUT_SECONDS) as resp: chunks: list[bytes] = [] total = 0 while True: chunk: bytes = resp.read(_CHUNK) if not chunk: break total += len(chunk) if total > MAX_FETCH_BYTES: gib = MAX_FETCH_BYTES // (1024 ** 3) raise TransportError( f"Fetch response exceeds the {gib} GiB cap. " "The server may be misconfigured.", 0, ) chunks.append(chunk) elapsed = time.perf_counter() - t0 logger.debug( "transport: _execute_fetch %s → %d bytes in %.3fs", req.full_url, total, elapsed, ) return b"".join(chunks) except urllib.error.HTTPError as exc: raise TransportError(_http_error_message(exc), exc.code) from exc except urllib.error.URLError as exc: raise TransportError(str(exc.reason), 0) from exc def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo: """Fetch repository metadata from ``GET {url}/refs``.""" endpoint = f"{url.rstrip('/')}/refs" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) return _parse_remote_info(raw) def fetch_pack( self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str] ) -> PackBundle: """Download VCS metadata via ``POST {url}/fetch`` (Phase 1 of two-phase fetch). Returns commits, snapshots, and branch_heads only. Object bytes are fetched separately via :meth:`fetch_objects`. """ endpoint = f"{url.rstrip('/')}/fetch" logger.debug( "transport: POST %s (want=%d, have=%d)", endpoint, len(want), len(have) ) payload: FetchRequest = {"want": want, "have": have} body_bytes: bytes = msgpack.packb(payload, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute_fetch(req) return _parse_bundle(raw) def fetch_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], ) -> list[ObjectPayload]: """Download object bytes via ``POST {url}/fetch/objects`` (Phase 2 of two-phase fetch). Uses :meth:`_execute_fetch` so the chunked streaming response is read correctly without hitting the small-endpoint size cap. """ if not object_ids: return [] endpoint = f"{url.rstrip('/')}/fetch/objects" logger.debug("transport: POST %s (objects=%d)", endpoint, len(object_ids)) body_bytes: bytes = msgpack.packb({"object_ids": object_ids}, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute_fetch(req) bundle = _parse_bundle(raw) return bundle.get("objects") or [] def push_pack( self, url: str, signing: SigningIdentity | None, bundle: PackBundle, branch: str, force: bool, local_head: str | None = None, ) -> PushResult: """Upload a PackBundle via ``POST {url}/push``. ``local_head`` is included so the server can advance the branch ref even when the bundle carries zero commits (all already present on the remote, e.g. the commit was previously pushed under a different branch). """ endpoint = f"{url.rstrip('/')}/push" logger.debug( "transport: POST %s (branch=%s, force=%s, commits=%d, objects=%d)", endpoint, branch, force, len(bundle.get("commits") or []), len(bundle.get("objects") or []), ) payload: _PushPayload = {"bundle": bundle, "branch": branch, "force": force} if local_head is not None: payload["local_head"] = local_head body_bytes: bytes = msgpack.packb(payload, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) return _parse_push_result(raw) def push_objects( self, url: str, signing: SigningIdentity | None, objects: list[ObjectPayload], ) -> ObjectsChunkResponse: """Pre-upload an object batch via ``POST {url}/push/objects``.""" endpoint = f"{url.rstrip('/')}/push/objects" logger.debug("transport: POST %s (objects=%d)", endpoint, len(objects)) body_bytes: bytes = msgpack.packb({"objects": objects}, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) return _parse_objects_response(raw) def push_object_pack( self, url: str, signing: SigningIdentity | None, objects: list[ObjectPayload], ) -> ObjectsChunkResponse: """Upload a pack of small objects via ``POST {url}/push/object-pack``. Collapses N per-object TLS connections into a single msgpack POST. The server enforces ≤ 1 000 objects and ≤ 50 MB total per request. """ endpoint = f"{url.rstrip('/')}/push/object-pack" logger.debug("transport: POST %s (pack=%d objects)", endpoint, len(objects)) body_bytes: bytes = msgpack.packb({"objects": objects}, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) return _parse_objects_response(raw) def filter_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], object_hints: dict[str, str] | None = None, ) -> FilterObjectsResult: """Return missing object IDs and delta bases via ``POST {url}/filter-objects``.""" endpoint = f"{url.rstrip('/')}/filter-objects" logger.debug("transport: POST %s (candidates=%d)", endpoint, len(object_ids)) body: dict[str, list[str] | dict[str, str]] = {"object_ids": object_ids} if object_hints: body["object_hints"] = object_hints body_bytes: bytes = msgpack.packb(body, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) parsed = self._decode(raw) missing_raw = parsed.get("missing") missing = [m for m in missing_raw if isinstance(m, str)] if isinstance(missing_raw, list) else [] bases_raw = parsed.get("bases") bases: dict[str, str] = ( {k: v for k, v in bases_raw.items() if isinstance(k, str) and isinstance(v, str)} if isinstance(bases_raw, dict) else {} ) return FilterObjectsResult(missing=missing, bases=bases) def presign_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], direction: str, ) -> PresignResponse: """Return presigned S3/R2 URLs via ``POST {url}/presign`` (MWP Phase 3).""" endpoint = f"{url.rstrip('/')}/presign" logger.debug( "transport: POST %s (objects=%d, direction=%s)", endpoint, len(object_ids), direction ) body_bytes: bytes = msgpack.packb( {"object_ids": object_ids, "direction": direction}, use_bin_type=True ) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) parsed = self._decode(raw) presigned_raw = parsed.get("presigned", {}) inline_raw = parsed.get("inline", []) presigned: Manifest = ( {str(k): str(v) for k, v in presigned_raw.items()} if isinstance(presigned_raw, dict) else {} ) inline: list[str] = ( [str(x) for x in inline_raw] if isinstance(inline_raw, list) else [] ) return PresignResponse(presigned=presigned, inline=inline) def confirm_objects( self, url: str, signing: SigningIdentity | None, object_ids: list[str], sizes: dict[str, int], ) -> ConfirmObjectsResponse: """Register presigned-uploaded objects via ``POST {url}/push/objects/confirm``.""" endpoint = f"{url.rstrip('/')}/push/objects/confirm" logger.debug("transport: POST %s (objects=%d)", endpoint, len(object_ids)) body_bytes: bytes = msgpack.packb( {"object_ids": object_ids, "sizes": sizes}, use_bin_type=True ) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) parsed = self._decode(raw) return ConfirmObjectsResponse( registered=int(parsed.get("registered", 0)), skipped=int(parsed.get("skipped", 0)), ) def negotiate( self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str], ) -> NegotiateResponse: """Depth-limited commit negotiation via ``POST {url}/negotiate`` (MWP Phase 5).""" endpoint = f"{url.rstrip('/')}/negotiate" logger.debug( "transport: POST %s (want=%d, have=%d)", endpoint, len(want), len(have) ) body_bytes: bytes = msgpack.packb({"want": want, "have": have}, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) parsed = self._decode(raw) ack_raw = parsed.get("ack", []) ack = [str(x) for x in ack_raw] if isinstance(ack_raw, list) else [] common_base_raw = parsed.get("common_base") common_base = str(common_base_raw) if isinstance(common_base_raw, str) else None ready_raw = parsed.get("ready", False) return NegotiateResponse( ack=ack, common_base=common_base, ready=bool(ready_raw), ) def push_tags( self, url: str, signing: SigningIdentity | None, tags: list[WireTag], ) -> int: """Push tags via ``POST {url}/tags``.""" endpoint = f"{url.rstrip('/')}/tags" logger.debug("transport: POST %s (tags=%d)", endpoint, len(tags)) body_bytes: bytes = msgpack.packb({"tags": list(tags)}, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) parsed = self._decode(raw) stored_val = parsed.get("stored") return int(stored_val) if isinstance(stored_val, int) else 0 def create_release( self, url: str, signing: SigningIdentity | None, release: ReleaseDict, ) -> str: """Create a release via ``POST {url}/releases``.""" endpoint = f"{url.rstrip('/')}/releases" logger.debug("transport: POST %s (tag=%s)", endpoint, release.get("tag", "")) # ReleaseDict contains no bytes fields so JSON encoding works directly. body_bytes: bytes = json.dumps(release).encode("utf-8") req = self._build_request("POST", endpoint, signing, body_bytes, content_type="application/json") raw = self._execute(req) parsed = self._decode(raw) release_id_val = parsed.get("release_id") return str(release_id_val) if isinstance(release_id_val, str) else "" def list_releases_remote( self, url: str, signing: SigningIdentity | None, channel: str | None = None, include_drafts: bool = False, ) -> list[ReleaseDict]: """List releases via ``GET {url}/releases``.""" qs_parts: list[str] = [] if channel: qs_parts.append(f"channel={urllib.parse.quote(channel)}") if include_drafts: qs_parts.append("include_drafts=1") endpoint = f"{url.rstrip('/')}/releases" if qs_parts: endpoint = f"{endpoint}?{'&'.join(qs_parts)}" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) parsed = self._decode(raw) return _parse_releases_list(parsed) def delete_release_remote( self, url: str, signing: SigningIdentity | None, tag: str, ) -> None: """Retract a release via ``DELETE {url}/releases/{tag}``.""" endpoint = f"{url.rstrip('/')}/releases/{urllib.parse.quote(tag, safe='')}" logger.debug("transport: DELETE %s", endpoint) req = self._build_request("DELETE", endpoint, signing) self._execute(req) def delete_branch_remote( self, url: str, signing: SigningIdentity | None, branch: str, ) -> None: """Delete a branch via ``DELETE {url}/branches/{branch}``.""" endpoint = f"{url.rstrip('/')}/branches/{urllib.parse.quote(branch, safe='')}" logger.debug("transport: DELETE %s", endpoint) req = self._build_request("DELETE", endpoint, signing) self._execute(req) # --------------------------------------------------------------------------- # Response parsers — JSON bytes → typed TypedDicts # --------------------------------------------------------------------------- # json.loads() returns Any (per typeshed), so we use isinstance narrowing # throughout. No explicit Any annotations appear in this file. # --------------------------------------------------------------------------- def _parse_remote_info(raw: bytes) -> RemoteInfo: """Parse ``GET /refs`` response bytes into a :class:`~muse.core.pack.RemoteInfo`.""" parsed = HttpTransport._decode(raw) repo_id_val = parsed.get("repo_id") domain_val = parsed.get("domain") default_branch_val = parsed.get("default_branch") branch_heads_raw = parsed.get("branch_heads") branch_heads: BranchHeads = {} if isinstance(branch_heads_raw, dict): for k, v in branch_heads_raw.items(): if isinstance(k, str) and isinstance(v, str): branch_heads[k] = v info = RemoteInfo( repo_id=str(repo_id_val) if isinstance(repo_id_val, str) else "", domain=str(domain_val) if isinstance(domain_val, str) else "midi", default_branch=( str(default_branch_val) if isinstance(default_branch_val, str) else "main" ), branch_heads=branch_heads, ) pack_origin_val = parsed.get("pack_origin") if isinstance(pack_origin_val, str) and pack_origin_val.strip(): info["pack_origin"] = pack_origin_val.strip() return info def _parse_bundle(raw: bytes) -> PackBundle: """Parse ``POST /fetch`` response bytes into a :class:`~muse.core.pack.PackBundle`.""" parsed = HttpTransport._decode(raw) bundle: PackBundle = {} # Commits — each item is a raw dict that CommitRecord.from_dict() will validate. commits_raw = parsed.get("commits") if isinstance(commits_raw, list): commits: list[CommitDict] = [] for item in commits_raw: if isinstance(item, dict): commits.append(_coerce_commit_dict(item)) bundle["commits"] = commits # Snapshots snapshots_raw = parsed.get("snapshots") if isinstance(snapshots_raw, list): snapshots: list[SnapshotDict] = [] for item in snapshots_raw: if isinstance(item, dict): snapshots.append(_coerce_snapshot_dict(item)) bundle["snapshots"] = snapshots # Objects — raw bytes in "content" (MWP msgpack wire format) objects_raw = parsed.get("objects") if isinstance(objects_raw, list): objects: list[ObjectPayload] = [] for item in objects_raw: if not isinstance(item, dict): continue oid = item.get("object_id") if not isinstance(oid, str): continue content_raw = item.get("content") if isinstance(content_raw, (bytes, bytearray)): objects.append(ObjectPayload(object_id=oid, content=bytes(content_raw))) bundle["objects"] = objects # Branch heads heads_raw = parsed.get("branch_heads") if isinstance(heads_raw, dict): branch_heads: BranchHeads = {} for k, v in heads_raw.items(): if isinstance(k, str) and isinstance(v, str): branch_heads[k] = v bundle["branch_heads"] = branch_heads return bundle def _parse_push_result(raw: bytes) -> PushResult: """Parse ``POST /push`` response bytes into a :class:`~muse.core.pack.PushResult`.""" parsed = HttpTransport._decode(raw) ok_val = parsed.get("ok") msg_val = parsed.get("message") heads_raw = parsed.get("branch_heads") branch_heads: BranchHeads = {} if isinstance(heads_raw, dict): for k, v in heads_raw.items(): if isinstance(k, str) and isinstance(v, str): branch_heads[k] = v return PushResult( ok=bool(ok_val) if isinstance(ok_val, bool) else False, message=str(msg_val) if isinstance(msg_val, str) else "", branch_heads=branch_heads, ) def _parse_objects_response(raw: bytes) -> ObjectsChunkResponse: """Parse ``POST /push/objects`` response into an :class:`~muse.core.pack.ObjectsChunkResponse`.""" parsed = HttpTransport._decode(raw) stored_val = parsed.get("stored") skipped_val = parsed.get("skipped") return ObjectsChunkResponse( stored=int(stored_val) if isinstance(stored_val, int) else 0, skipped=int(skipped_val) if isinstance(skipped_val, int) else 0, ) def _coerce_sem_ver_bump(raw: _MsgVal) -> SemVerBump: """Safely coerce a raw value to a :class:`~muse.domain.SemVerBump` literal.""" if raw == "major": return "major" if raw == "minor": return "minor" if raw == "patch": return "patch" return "none" def _parse_releases_list(parsed: _MsgDict) -> list[ReleaseDict]: """Extract a list of :class:`~muse.core.store.ReleaseDict` from a parsed response.""" releases_raw = parsed.get("releases") if not isinstance(releases_raw, list): return [] results: list[ReleaseDict] = [] for item in releases_raw: if not isinstance(item, dict): continue try: semver_raw = item.get("semver") if isinstance(semver_raw, dict): sv_major = semver_raw.get("major") sv_minor = semver_raw.get("minor") sv_patch = semver_raw.get("patch") sv_pre = semver_raw.get("pre") sv_build = semver_raw.get("build") semver: SemVerTag = SemVerTag( major=int(sv_major) if isinstance(sv_major, int) else 0, minor=int(sv_minor) if isinstance(sv_minor, int) else 0, patch=int(sv_patch) if isinstance(sv_patch, int) else 0, pre=sv_pre if isinstance(sv_pre, str) else "", build=sv_build if isinstance(sv_build, str) else "", ) else: semver = SemVerTag(major=0, minor=0, patch=0, pre="", build="") changelog_raw = item.get("changelog") or [] changelog: list[ChangelogEntry] = [] if isinstance(changelog_raw, list): for entry in changelog_raw: if not isinstance(entry, dict): continue bc_raw = entry.get("breaking_changes") bc_list: list[str] = [str(b) for b in bc_raw if isinstance(b, str)] if isinstance(bc_raw, list) else [] changelog.append(ChangelogEntry( commit_id=str(entry.get("commit_id", "")), message=str(entry.get("message", "")), sem_ver_bump=_coerce_sem_ver_bump(entry.get("sem_ver_bump")), breaking_changes=bc_list, author=str(entry.get("author", "")), committed_at=str(entry.get("committed_at", "")), agent_id=str(entry.get("agent_id", "")), model_id=str(entry.get("model_id", "")), )) results.append(ReleaseDict( release_id=str(item.get("release_id", "")), repo_id=str(item.get("repo_id", "")), tag=str(item.get("tag", "")), semver=semver, channel=str(item.get("channel", "stable")), commit_id=str(item.get("commit_id", "")), snapshot_id=str(item.get("snapshot_id", "")), title=str(item.get("title", "")), body=str(item.get("body", "")), changelog=changelog, agent_id=str(item.get("agent_id", "")), model_id=str(item.get("model_id", "")), is_draft=bool(item.get("is_draft", False)), gpg_signature=str(item.get("gpg_signature", "")), created_at=str(item.get("created_at", "")), )) except (KeyError, TypeError, ValueError): continue return results # --------------------------------------------------------------------------- # TypedDict coercion helpers — extract known string fields from raw JSON dicts # --------------------------------------------------------------------------- # CommitDict and SnapshotDict are total=False (all fields optional), so we # only extract the string/scalar fields we can safely validate here. # CommitRecord.from_dict() and SnapshotRecord.from_dict() re-validate # required fields when apply_pack() calls them. # --------------------------------------------------------------------------- # _WireVal is now an alias for _MsgVal — kept for readability at call sites. _WireVal = _MsgVal def _str(val: _WireVal) -> str: """Return *val* as str, or empty string if not a str.""" return val if isinstance(val, str) else "" def _str_or_none(val: _WireVal) -> str | None: """Return *val* as str, or None if not a str.""" return val if isinstance(val, str) else None def _int_or(val: _WireVal, default: int) -> int: """Return *val* as int, or *default* if not an int.""" return val if isinstance(val, int) else default def _coerce_commit_dict(raw: _WireDict) -> CommitDict: """Extract typed scalar fields from *raw* into a :class:`~muse.core.store.CommitDict`. Only primitive fields are validated here; ``structured_delta`` is preserved as-is because :class:`~muse.core.store.CommitRecord.from_dict` already handles it gracefully. """ metadata_raw = raw.get("metadata") metadata: Metadata = {} if isinstance(metadata_raw, dict): for k, v in metadata_raw.items(): if isinstance(k, str) and isinstance(v, str): metadata[k] = v reviewed_by_raw = raw.get("reviewed_by") reviewed_by: list[str] = [] if isinstance(reviewed_by_raw, list): for item in reviewed_by_raw: if isinstance(item, str): reviewed_by.append(item) breaking_changes_raw = raw.get("breaking_changes") breaking_changes: list[str] = [] if isinstance(breaking_changes_raw, list): for item in breaking_changes_raw: if isinstance(item, str): breaking_changes.append(item) sem_ver_raw = raw.get("sem_ver_bump") sem_ver: SemVerBump if sem_ver_raw == "major": sem_ver = "major" elif sem_ver_raw == "minor": sem_ver = "minor" elif sem_ver_raw == "patch": sem_ver = "patch" else: sem_ver = "none" return CommitDict( commit_id=_str(raw.get("commit_id")), repo_id=_str(raw.get("repo_id")), branch=_str(raw.get("branch")), snapshot_id=_str(raw.get("snapshot_id")), message=_str(raw.get("message")), committed_at=_str(raw.get("committed_at")), parent_commit_id=_str_or_none(raw.get("parent_commit_id")), parent2_commit_id=_str_or_none(raw.get("parent2_commit_id")), author=_str(raw.get("author")), metadata=metadata, structured_delta=None, sem_ver_bump=sem_ver, breaking_changes=breaking_changes, agent_id=_str(raw.get("agent_id")), model_id=_str(raw.get("model_id")), toolchain_id=_str(raw.get("toolchain_id")), prompt_hash=_str(raw.get("prompt_hash")), signature=_str(raw.get("signature")), signer_key_id=_str(raw.get("signer_key_id")), format_version=_int_or(raw.get("format_version"), 1), reviewed_by=reviewed_by, test_runs=_int_or(raw.get("test_runs"), 0), ) def _coerce_snapshot_dict(raw: _WireDict) -> SnapshotDict: """Extract typed fields from *raw* into a :class:`~muse.core.store.SnapshotDict`.""" manifest_raw = raw.get("manifest") manifest: Manifest = {} if isinstance(manifest_raw, dict): for k, v in manifest_raw.items(): if isinstance(k, str) and isinstance(v, str): manifest[k] = v # ``directories`` is hashed into snapshot_id — dropping it produces a # SnapshotRecord whose snapshot_id never matches the recomputed hash, # making every snapshot with non-empty directories permanently unreadable. directories_raw = raw.get("directories") directories: list[str] = [] if isinstance(directories_raw, list): for item in directories_raw: if isinstance(item, str): directories.append(item) return SnapshotDict( snapshot_id=_str(raw.get("snapshot_id")), manifest=manifest, directories=directories, created_at=_str(raw.get("created_at")), note=_str(raw.get("note", "")), ) # --------------------------------------------------------------------------- # LocalFileTransport helpers # --------------------------------------------------------------------------- def _is_ancestor( candidate: str, from_commit: str, bundle_by_id: _CommitBundle, remote_root: pathlib.Path, max_depth: int = 100_000, ) -> bool: """Return True if *candidate* is an ancestor of (or equal to) *from_commit*. Walks the commit graph BFS-style starting from *from_commit*, consulting *bundle_by_id* first (commits included in the push bundle) and falling back to the existing commits on disk in *remote_root* (commits already present). This two-source walk is necessary because ``build_pack()`` excludes commits in the caller's ``have`` set from the bundle — those commits are already on disk at the remote and must be consulted directly. Args: candidate: The commit ID to search for (typically the remote's current HEAD). from_commit: Starting point of the BFS walk (typically the new tip being pushed). bundle_by_id: Commits included in the push bundle, keyed by commit_id. remote_root: Root of the remote Muse repo (for reading pre-existing commits). max_depth: BFS depth cap — prevents unbounded walks on corrupt graphs. Returns: ``True`` if *candidate* is reachable from *from_commit*, ``False`` otherwise. """ from muse.core.store import read_commit as _rc seen: set[str] = set() queue: list[str] = [from_commit] depth = 0 while queue and depth < max_depth: cid = queue.pop(0) if cid in seen: continue seen.add(cid) if cid == candidate: return True # Prefer bundle for unwritten commits; fall back to remote store. parent1: str | None parent2: str | None if cid in bundle_by_id: bc = bundle_by_id[cid] p1_raw = bc.get("parent_commit_id") p2_raw = bc.get("parent2_commit_id") parent1 = p1_raw if isinstance(p1_raw, str) else None parent2 = p2_raw if isinstance(p2_raw, str) else None else: rec = _rc(remote_root, cid) if rec is None: depth += 1 continue parent1 = rec.parent_commit_id parent2 = rec.parent2_commit_id if parent1 and parent1 not in seen: queue.append(parent1) if parent2 and parent2 not in seen: queue.append(parent2) depth += 1 return False # --------------------------------------------------------------------------- # LocalFileTransport — push/pull between two repos on the same filesystem # --------------------------------------------------------------------------- class LocalFileTransport: """Transport implementation for ``file://`` URLs. Allows ``muse push file:///path/to/repo`` and ``muse pull`` between two Muse repositories on the same filesystem (or a shared network mount) without requiring a MuseHub server. The remote path must be the root of an initialised Muse repository — it must contain a ``.muse/`` directory. No separate "bare repo" format is required; Muse repositories are self-describing. This transport never makes network calls. All operations are synchronous filesystem reads and writes, consistent with the rest of the Muse CLI. """ @staticmethod def _repo_root(url: str) -> pathlib.Path: """Extract and validate the filesystem path from a ``file://`` URL. Security guarantees: - Rejects non-``file://`` schemes unconditionally. - Calls ``.resolve()`` to canonicalize the path, dereferencing all symlinks before any filesystem operations. A symlink at the URL target that points to a directory without a ``.muse/`` subdirectory is rejected — the check is on the resolved, canonical path, not the symlink itself. - Verifies that ``.muse/`` exists at the resolved root, preventing accidental pushes to arbitrary directories. """ parsed = urllib.parse.urlparse(url) if parsed.scheme != "file": raise TransportError( f"LocalFileTransport requires a file:// URL, got: {url!r}", 0 ) # urllib.parse.urlparse on file:///abs/path gives netloc="" path="/abs/path". # On Windows file://C:/path gives netloc="" path="/C:/path" — strip leading # slash for Windows compatibility via pathlib. path_str = parsed.netloc + parsed.path # resolve() dereferences all symlinks and normalises ".." components. # This is the defence against symlink-based path escape attempts. root = pathlib.Path(path_str).resolve() if not (root / ".muse").is_dir(): raise TransportError( f"Remote path {root!r} does not contain a .muse/ directory. " "Run 'muse init' in the target directory first.", 404, ) return root def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo: # noqa: ARG002 """Read branch heads directly from the remote's ref files.""" from muse.core.store import get_all_branch_heads remote_root = self._repo_root(url) repo_json_path = remote_root / ".muse" / "repo.json" try: repo_data = json.loads(repo_json_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise TransportError(f"Cannot read remote repo.json: {exc}", 0) from exc repo_id = str(repo_data.get("repo_id", "")) domain = str(repo_data.get("domain", "midi")) default_branch = str(repo_data.get("default_branch", "main")) branch_heads = get_all_branch_heads(remote_root) return RemoteInfo( repo_id=repo_id, domain=domain, default_branch=default_branch, branch_heads=branch_heads, ) def fetch_pack( self, url: str, signing: SigningIdentity | None, want: list[str], have: list[str] # noqa: ARG002 ) -> PackBundle: """Build a PackBundle (metadata only) from the remote's local store.""" from muse.core.pack import build_pack remote_root = self._repo_root(url) have_set = set(have) bundle = build_pack(remote_root, commit_ids=want, have=list(have_set)) # Two-phase protocol: strip objects from the metadata bundle. # LocalFileTransport.fetch_objects() reads them directly from the remote store. bundle_no_objects: PackBundle = {} if "commits" in bundle: bundle_no_objects["commits"] = bundle["commits"] if "snapshots" in bundle: bundle_no_objects["snapshots"] = bundle["snapshots"] if "tags" in bundle: bundle_no_objects["tags"] = bundle["tags"] if "branch_heads" in bundle: bundle_no_objects["branch_heads"] = bundle["branch_heads"] return bundle_no_objects def fetch_objects( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 object_ids: list[str], ) -> list[ObjectPayload]: """Read object bytes directly from the remote's local object store.""" from muse.core.object_store import read_object remote_root = self._repo_root(url) result: list[ObjectPayload] = [] for oid in object_ids: data = read_object(remote_root, oid) if data is not None: result.append(ObjectPayload(object_id=oid, content=data)) return result def push_pack( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 bundle: PackBundle, branch: str, force: bool, local_head: str | None = None, # noqa: ARG002 ) -> PushResult: """Write a PackBundle directly into the remote's local store. Security guarantees: - ``branch`` is validated with :func:`~muse.core.validation.validate_branch_name` before any I/O. Names containing path traversal components (`..`), null bytes, backslashes, or other forbidden characters are rejected. - The ref file path is further hardened with :func:`~muse.core.validation.contain_path`, which resolves symlinks and asserts the result stays inside ``.muse/refs/heads/``. A branch name that ``validate_branch_name`` would allow but that resolves outside the expected directory (e.g. via a pre-placed symlink) is rejected before any write occurs. - A fast-forward check prevents overwriting diverged remote history unless ``force=True`` is explicitly passed. """ from muse.core.pack import apply_pack from muse.core.store import get_all_branch_heads, get_head_commit_id, write_branch_ref as _write_branch_ref remote_root = self._repo_root(url) try: validate_branch_name(branch) except ValueError as exc: return PushResult(ok=False, message=str(exc), branch_heads={}) # Determine the new tip for the branch. # Prefer an explicit branch_heads entry in the bundle (set by the push # command when it knows the local HEAD). Fall back to computing the # leaf commit — the commit in the bundle that is not referenced as a # parent of any other commit in the bundle, filtered to the branch. # This handles bundles produced by build_pack(), which does not # populate branch_heads. bundle_heads = bundle.get("branch_heads") or {} new_tip: str | None = bundle_heads.get(branch) if new_tip is None: bundle_commits_list = bundle.get("commits") or [] all_parent_ids: set[str] = set() for bc in bundle_commits_list: pid = bc.get("parent_commit_id") if isinstance(pid, str): all_parent_ids.add(pid) pid2 = bc.get("parent2_commit_id") if isinstance(pid2, str): all_parent_ids.add(pid2) # Leaf = commit whose ID is not a parent of any other bundle commit. # Prefer commits whose branch field matches; otherwise take any leaf. leaves_for_branch = [ bc["commit_id"] for bc in bundle_commits_list if bc.get("commit_id") not in all_parent_ids and bc.get("branch") == branch and isinstance(bc.get("commit_id"), str) ] any_leaves = [ bc["commit_id"] for bc in bundle_commits_list if bc.get("commit_id") not in all_parent_ids and isinstance(bc.get("commit_id"), str) ] fallback: list[str | None] = [None] new_tip = (leaves_for_branch or any_leaves or fallback)[0] # Fast-forward check: the remote's current HEAD for this branch must be # an ancestor of the tip commit in the bundle, unless --force is passed. if not force and new_tip: remote_tip = get_head_commit_id(remote_root, branch) if remote_tip and remote_tip != new_tip: # BFS from new_tip through bundle commits *and* existing remote # commits to find whether remote_tip is a reachable ancestor. # We cannot rely on bundle commits alone because build_pack() # excludes commits the receiver already has (the "have" set). bundle_by_id: _CommitBundle = { c["commit_id"]: c for c in (bundle.get("commits") or []) if isinstance(c.get("commit_id"), str) } if not _is_ancestor(remote_tip, new_tip, bundle_by_id, remote_root): return PushResult( ok=False, message=( f"Push rejected: remote branch '{branch}' has diverged. " "Pull and merge first, or use --force." ), branch_heads={}, ) try: apply_pack(remote_root, bundle) except Exception as exc: # noqa: BLE001 return PushResult(ok=False, message=f"Failed to apply pack: {exc}", branch_heads={}) # Update the remote branch ref to the new tip. # contain_path() resolves symlinks and asserts the result stays inside # .muse/refs/heads/ — defence-in-depth beyond validate_branch_name. if new_tip: heads_base = remote_root / ".muse" / "refs" / "heads" try: ref_path = contain_path(heads_base, branch) except ValueError as exc: return PushResult( ok=False, message=f"Rejected: branch ref path is unsafe — {exc}", branch_heads={}, ) _write_branch_ref(remote_root, branch, new_tip) logger.info("✅ local-transport: updated %s → %s", branch, new_tip[:8]) return PushResult( ok=True, message=f"local push to {url!r} succeeded", branch_heads=get_all_branch_heads(remote_root), ) def push_objects( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 objects: list[ObjectPayload], ) -> ObjectsChunkResponse: """Write objects directly into the remote's local object store. Mirrors the server-side ``POST /push/objects`` behaviour: objects are content-addressed and idempotent — already-present objects are skipped. No branch refs are touched; only blob bytes are written. """ from muse.core.object_store import write_object remote_root = self._repo_root(url) stored = 0 skipped = 0 for obj in objects: oid = obj.get("object_id", "") raw = obj.get("content", b"") if not oid or not raw: continue if write_object(remote_root, oid, raw): stored += 1 else: skipped += 1 return ObjectsChunkResponse(stored=stored, skipped=skipped) def push_object_pack( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 objects: list[ObjectPayload], ) -> ObjectsChunkResponse: """Write a pack of small objects directly into the remote's local object store. Mirrors the server-side ``POST /push/object-pack`` behaviour. Objects are content-addressed and idempotent — already-present objects are skipped. No branch refs are touched; only blob bytes are written. The ``local://`` backend has no TLS overhead so the pack/presign split does not matter here — both paths write to the same object store. This implementation is identical to ``push_objects`` and exists solely to satisfy the :class:`MuseTransport` protocol. """ from muse.core.object_store import read_object, write_object remote_root = self._repo_root(url) stored = 0 skipped = 0 for obj in objects: oid = obj.get("object_id", "") raw = obj.get("content", b"") if not oid or not raw: continue encoding = obj.get("encoding", "raw") if encoding == "zlib": raw = decompress_zlib(raw) elif encoding == "delta+zlib": base_id_val: str = obj.get("base_id", "") # type: ignore[assignment] base_bytes = read_object(remote_root, base_id_val) or b"" raw = apply_delta(base_bytes, raw) if write_object(remote_root, oid, raw): stored += 1 else: skipped += 1 return ObjectsChunkResponse(stored=stored, skipped=skipped) def filter_objects( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 object_ids: list[str], object_hints: dict[str, str] | None = None, # noqa: ARG002 ) -> FilterObjectsResult: """Return object IDs missing from the remote's local store (MWP Phase 1). The local-file transport does not perform delta base lookup — bases is always empty. Delta encoding is only profitable over a real network. """ from muse.core.object_store import read_object remote_root = self._repo_root(url) missing: list[str] = [] for oid in object_ids: if read_object(remote_root, oid) is None: missing.append(oid) return FilterObjectsResult(missing=missing, bases={}) def presign_objects( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 object_ids: list[str], direction: str, # noqa: ARG002 ) -> PresignResponse: """Local transport has no object storage backend — return all as inline.""" return PresignResponse(presigned={}, inline=list(object_ids)) def confirm_objects( self, url: str, # noqa: ARG002 signing: SigningIdentity | None, # noqa: ARG002 object_ids: list[str], # noqa: ARG002 sizes: dict[str, int], # noqa: ARG002 ) -> ConfirmObjectsResponse: """No-op for local transport — local push_objects handles DB registration.""" return ConfirmObjectsResponse(registered=0, skipped=0) def negotiate( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 want: list[str], have: list[str], ) -> NegotiateResponse: """Commit negotiation against a local repo (MWP Phase 5).""" from muse.core.store import read_commit as _rc remote_root = self._repo_root(url) have_set = set(have) # Which of the client's have-IDs exist on the remote? ack = [cid for cid in have if _rc(remote_root, cid) is not None] ack_set = set(ack) common_base: str | None = None for cid in want: # Walk parents looking for an acked ancestor. commit = _rc(remote_root, cid) if commit is None: continue for pid in filter(None, [commit.parent_commit_id, commit.parent2_commit_id]): if pid in ack_set: common_base = pid break if common_base: break ready = common_base is not None or not have_set return NegotiateResponse(ack=ack, common_base=common_base, ready=ready) def push_tags( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 tags: list[WireTag], ) -> int: """Write tags directly into the remote's local tag store.""" from muse.core.store import TagDict, TagRecord, write_tag remote_root = self._repo_root(url) stored = 0 for wire_tag in tags: try: tag_record = TagRecord.from_dict(TagDict( tag_id=wire_tag["tag_id"], repo_id=wire_tag["repo_id"], commit_id=wire_tag["commit_id"], tag=wire_tag["tag"], created_at=wire_tag["created_at"], )) write_tag(remote_root, tag_record) stored += 1 except (KeyError, ValueError) as exc: logger.warning("⚠️ local-transport push_tags: bad tag — %s", exc) return stored def create_release( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 release: ReleaseDict, ) -> str: """Write a release directly into the remote's local release store.""" from muse.core.store import ReleaseRecord, write_release remote_root = self._repo_root(url) try: release_record = ReleaseRecord.from_dict(release) write_release(remote_root, release_record) return release_record.release_id except (KeyError, ValueError) as exc: raise TransportError(f"create_release: invalid release data — {exc}", 0) from exc def list_releases_remote( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 channel: str | None = None, include_drafts: bool = False, ) -> list[ReleaseDict]: """Read releases from the remote's local release store.""" from muse.core.semver import ReleaseChannel from muse.core.store import list_releases remote_root = self._repo_root(url) repo_json_path = remote_root / ".muse" / "repo.json" try: repo_data = json.loads(repo_json_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise TransportError(f"Cannot read remote repo.json: {exc}", 0) from exc repo_id = str(repo_data.get("repo_id", "")) _ChannelMap = dict[str, ReleaseChannel] _channel_map: _ChannelMap = { "stable": "stable", "beta": "beta", "alpha": "alpha", "nightly": "nightly", } channel_arg: ReleaseChannel | None = _channel_map.get(channel, None) if channel else None records = list_releases(remote_root, repo_id, channel=channel_arg, include_drafts=include_drafts) return [r.to_dict() for r in records] def delete_release_remote( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 tag: str, ) -> None: """Delete a release record from the remote's local release store.""" from muse.core.store import delete_release, get_release_for_tag remote_root = self._repo_root(url) repo_json_path = remote_root / ".muse" / "repo.json" try: repo_data = json.loads(repo_json_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise TransportError(f"Cannot read remote repo.json: {exc}", 0) from exc repo_id = str(repo_data.get("repo_id", "")) release = get_release_for_tag(remote_root, repo_id, tag) if release is None: raise TransportError(f"Release '{tag}' not found on remote.", 404) if not delete_release(remote_root, repo_id, release.release_id): raise TransportError(f"Failed to delete release '{tag}' on remote.", 0) def delete_branch_remote( self, url: str, signing: SigningIdentity | None, # noqa: ARG002 branch: str, ) -> None: """Delete a branch ref from the remote's local ref store.""" remote_root = self._repo_root(url) ref_file = remote_root / ".muse" / "refs" / "heads" / branch if not ref_file.is_file(): raise TransportError(f"Branch '{branch}' not found on remote.", 404) heads_dir = remote_root / ".muse" / "refs" / "heads" ref_file.unlink() # Prune empty parent directories (e.g. feat/ left behind after feat/my-thing). for parent in ref_file.parents: if parent == heads_dir: break try: parent.rmdir() except OSError: break # --------------------------------------------------------------------------- # Factory — select transport based on URL scheme # --------------------------------------------------------------------------- def make_transport(url: str) -> "HttpTransport | LocalFileTransport": """Return the appropriate transport for *url*. - ``file://`` URLs → :class:`LocalFileTransport` (no server required) - All other URLs → :class:`HttpTransport` (requires MuseHub server) Args: url: Remote repository URL. Returns: A transport instance implementing :class:`MuseTransport`. """ if urllib.parse.urlparse(url).scheme == "file": return LocalFileTransport() return HttpTransport() def negotiate_have( transport: "HttpTransport | LocalFileTransport", url: str, signing: SigningIdentity | None, want: list[str], all_local: list[str], ) -> list[str]: """Return the minimal have-list via MWP depth-limited negotiation. Sends at most :data:`NEGOTIATE_DEPTH` recent ancestors per round. The server responds with which commits it already has and whether a common base was found (``ready``). Repeats until the server signals ready or all local commits are exhausted, then falls back to the full list so the subsequent fetch always succeeds. This is the transport-layer half of the Muse Wire Protocol (MWP) commit negotiation step. Both ``muse fetch`` and ``muse pull`` delegate here to avoid duplicating the same negotiation loop. Args: transport: An active transport (HTTP or local file). url: Remote repository URL. token: Optional auth token. want: Commit IDs the client wants from the remote. all_local: All local commit IDs, most-recent first. Returns: A minimal ``have`` list for use in ``fetch_pack``. """ if not all_local: return [] offset = 0 while offset < len(all_local): batch = all_local[offset : offset + NEGOTIATE_DEPTH] resp = transport.negotiate(url, signing, want=want, have=batch) ack = resp["ack"] if resp["ready"]: return ack or batch offset += NEGOTIATE_DEPTH # Exhausted all local commits without finding a common base — send full list. return all_local