"""muse hub — MuseHub fabric connection management. The hub is not just a remote. It is the shared fabric where versioned multidimensional state flows across agents, humans, and repositories. Connecting a repo to a hub anchors it to the synchronisation layer that enables push/pull, plugin discovery, and multi-agent coordination. Separation of concerns ----------------------- - ``muse remote`` manages generic push/pull endpoints (any Muse server). - ``muse hub`` manages the *primary identity fabric* — the one hub this repo belongs to for authentication, discovery, and coordination. A repo has at most **one** hub. It may have many remotes. Security model -------------- - ``_normalise_url`` enforces HTTPS for non-loopback hosts at connect time. - ``_hub_api`` validates the URL scheme before every network request, preventing SSRF from a tampered ``config.toml``. - ``_hub_api`` caps the response body at ``_MAX_API_RESPONSE_BYTES`` to prevent OOM from a hostile hub. - All user-controlled values (proposal titles, branch names, hub URLs) are passed through ``sanitize_display()`` before being printed. - All diagnostic messages go to **stderr**; **stdout** is reserved for JSON and machine-readable data. Subcommands ----------- :: muse hub connect Attach this repo to a MuseHub instance. muse hub status [--json] Show connection and identity information. muse hub disconnect [--json] Remove the hub association from this repo. muse hub ping [--json] Test HTTP connectivity to the hub. muse hub proposal list [--json] List proposals on MuseHub. muse hub proposal read [--json] Show a single proposal. muse hub proposal create [--json] Open a new proposal. muse hub proposal merge [--json] Merge a proposal. JSON schemas ------------ ``muse hub connect --json``:: {"status": "ok", "hub_url": "", "hostname": "", "authenticated": true|false, "identity_name": "", "identity_type": ""} ``muse hub status --json``:: {"hub_url": "", "hostname": "", "authenticated": true|false, "identity_type": "", "identity_name": "", "identity_id": ""} ``muse hub disconnect --json``:: {"status": "ok"|"nothing_to_do", "hostname": ""} ``muse hub ping --json``:: {"status": "ok"|"error", "hub_url": "", "hostname": "", "reachable": true|false, "message": ""} Agent workflow examples ----------------------- :: # Verify hub auth before starting work muse hub status --json | python3 -c "import json,sys; d=json.load(sys.stdin); sys.exit(0 if d['authenticated'] else 1)" # Create a proposal and capture the ID muse hub proposal create --title "feat: x" --json | python3 -c "import json,sys; print(json.load(sys.stdin)['proposalId'])" # List open proposals (machine-readable) muse hub proposal list --json # Merge with squash strategy muse hub proposal merge af54753d --strategy squash --json """ import argparse import http.client import json import logging import re import sys import textwrap import urllib.error import urllib.parse import urllib.request from collections.abc import Mapping from typing import IO, TypedDict from muse.cli.config import ( clear_hub_url, get_hub_url, get_remote, list_remotes, set_hub_url, ) from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.identity import IdentityEntry, load_identity from muse.core.repo import find_repo_root from muse.core.refs import read_current_branch from muse.core.timing import start_timer from muse.core.types import Metadata, short_id, split_id from muse.core.validation import clamp_int, sanitize_display logger = logging.getLogger(__name__) type _HubPayload = dict[str, str | bool | list[str]] type _ProposalPayload = dict[str, str | list[str]] _CONNECT_TIMEOUT = 8 # seconds for ping/status health check _MAX_API_RESPONSE_BYTES = 4 * 1024 * 1024 # 4 MiB cap on API responses # Only allow http and https — no file://, ftp://, data://, etc. _ALLOWED_API_SCHEMES = frozenset({"http", "https"}) # Maximum proposals fetched when resolving an 8-char prefix — repos with more proposals # than this will not find older ones by prefix. Intentionally conservative; # raise if real-world repos hit the limit. _PROPOSAL_PREFIX_RESOLVE_LIMIT = 100 _MAX_PROPOSAL_BODY_LINES = 20 # body lines shown in text mode before truncation hint _MAX_PROPOSAL_TITLE_LEN = 512 # client-side guard: titles longer than this are rejected _MAX_ISSUE_TITLE_LEN = 512 # client-side guard: issue titles longer than this are rejected _MAX_REPO_NAME_LEN = 255 # mirrors CreateRepoRequest.name max_length on the server _MAX_REPO_DESC_LEN = 1024 # reasonable cap on description before sending to server _MAX_ISSUE_LABEL_LEN = 255 # mirrors label name max_length on the server _MAX_ISSUE_COMMENT_LEN = 50_000 # mirrors IssueCommentCreate.body max_length on the server _MAX_LABEL_NAME_LEN = 50 # mirrors LabelCreate.name max_length on the server _MAX_LABEL_DESC_LEN = 200 # mirrors LabelCreate.description max_length on the server _MAX_HANDLE_LEN = 64 # mirrors hub identity handle max_length on the server # Handle format: starts with letter/digit; continues with letters, digits, # hyphens, underscores. No leading/trailing whitespace; no shell metacharacters; # no Unicode confusables. Enforced before any network I/O. _HANDLE_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]*$") # ── TypedDicts ──────────────────────────────────────────────────────────────── class _ConnectJson(EnvelopeJson): """JSON output for ``muse hub connect --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ status Always ``"ok"`` on success — all error paths raise SystemExit before any JSON is emitted. hub_url Normalised hub URL with no trailing slash (e.g. ``https://musehub.ai``). hostname Host and optional port in display form (e.g. ``musehub.ai``). authenticated ``True`` when an Ed25519 identity is stored for this hub in ``~/.muse/identity.toml``. identity_name Handle of the stored identity, or ``""`` when not authenticated. identity_type ``"human"`` | ``"agent"`` | ``""`` when not authenticated. """ status: str hub_url: str hostname: str authenticated: bool identity_name: str identity_type: str class _StatusJson(EnvelopeJson): """JSON output for ``muse hub status --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. All domain keys are always present so agents can access them without guard checks. Fields ------ hub_url URL stored in ``.muse/config.toml``. hostname Host and optional port in display form. authenticated ``True`` when an Ed25519 identity is stored for this hub. identity_type ``"human"`` | ``"agent"`` | ``""`` when not authenticated. identity_name Handle of the stored identity, or ``""`` when not authenticated. identity_id Fingerprint of the stored key, or ``""`` when not authenticated. capabilities List of capability strings (e.g. ``["read:*"]``); ``[]`` for humans or when not authenticated. """ hub_url: str hostname: str authenticated: bool identity_type: str identity_name: str identity_id: str capabilities: list[str] class _DisconnectJson(EnvelopeJson): """JSON output for ``muse hub disconnect --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ status ``"ok"`` when a hub was disconnected; ``"nothing_to_do"`` when no hub was configured (idempotent — always exits 0). hub_url Full normalised URL that was removed, or ``""`` on nothing_to_do. hostname Host[:port] display form, or ``""`` on nothing_to_do. """ status: str hub_url: str hostname: str class _PingJson(EnvelopeJson): """JSON output for ``muse hub ping --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ status ``"ok"`` when the hub returned HTTP 2xx; ``"error"`` otherwise. hub_url URL that was pinged. hostname Host[:port] display form. reachable ``True`` when the hub returned a 2xx response. message Human-readable result: ``"HTTP 200 OK"`` on success, or an error reason (timeout string, DNS error, HTTP status code, etc.). """ status: str hub_url: str hostname: str reachable: bool message: str class _RepoEntry(TypedDict, total=False): """A single repository entry in MuseHub list and search API responses. All fields are optional (``total=False``) because different endpoints return different subsets. Callers must guard with ``.get()`` or check key presence before accessing. Fields ------ owner Handle of the repository owner (e.g. ``"gabriel"``). slug URL-safe repository slug (e.g. ``"muse"``). repoId ID of the repository (camelCase — matches the raw API response). """ owner: str slug: str repoId: str class _ProposalEntry(TypedDict, total=False): """A single proposal entry in MuseHub proposal list API responses. All fields are optional (``total=False``) because list responses may omit fields that require an extra database lookup. The ``read`` endpoint populates the full set; the ``list`` endpoint populates a subset. Fields ------ proposalId ID of the proposal (camelCase — matches raw API response). title Human-readable proposal title. state Lifecycle state: ``"open"`` | ``"merged"`` | ``"closed"``. fromBranch Source branch name (camelCase — matches raw API response). toBranch Target branch name (camelCase — matches raw API response). author Handle of the proposal author. createdAt ISO-8601 UTC creation timestamp (camelCase — matches raw API). """ proposalId: str title: str state: str fromBranch: str toBranch: str author: str createdAt: str class _RepoJson(EnvelopeJson): """JSON output for ``muse hub repo create --json``. Inherits the 6 standard envelope fields from :class:`~muse.core.envelope.EnvelopeJson`. Fields ------ repo_id ID of the newly created repository. name Repository name as provided by the caller. owner Handle of the owner (authenticated user or ``--owner`` override). slug URL-safe slug derived from the name. visibility ``"public"`` | ``"private"``. description Free-text description (may be empty string). clone_url Full push/pull URL for ``muse remote add`` / ``muse clone``. tags List of topic tags applied at creation time. created_at ISO-8601 UTC timestamp of creation (e.g. ``"2025-01-15T12:00:00Z"``). """ repo_id: str name: str owner: str slug: str visibility: str description: str clone_url: str tags: list[str] created_at: str class _HubApiResponse(TypedDict, total=False): """Generic MuseHub API response envelope used internally by ``_hub_api``. All fields are optional (``total=False``) because different endpoints populate different subsets. This TypedDict is an internal contract — it is never emitted directly to callers. Callers extract the specific fields they need and construct domain-typed output dicts. Fields ------ repo_id ID of a repo, present on create/read repo endpoints. repos List of :class:`_RepoEntry` dicts from list/search endpoints. proposals List of :class:`_ProposalEntry` dicts from proposal endpoints. """ repo_id: str repos: list[_RepoEntry] proposals: list[_ProposalEntry] class _LabelEntry(TypedDict, total=False): """A single label row returned by the MuseHub labels API. All fields are optional (``total=False``) because create, read, and list endpoints return different subsets. The ``list`` endpoint populates all fields; the ``create`` response may omit ``repo_id``. Fields ------ label_id ID of the label. repo_id ID of the repository this label belongs to. name Display name of the label (e.g. ``"bug"``, ``"enhancement"``). color Hex color code without ``#`` prefix (e.g. ``"d73a4a"``). description Free-text description shown on the label chip. """ label_id: str repo_id: str name: str color: str description: str # ── Security — redirect refusal for ping ────────────────────────────────────── class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): """Refuse all HTTP redirects for the ping health check. No credentials travel on the ping request, but silently following redirects (potentially cross-scheme or cross-host) is misleading about what was actually reached and normalises insecure redirect behaviour. """ 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}): hub redirected to {newurl!r}. Update the hub URL.", headers, fp, ) _PING_OPENER = urllib.request.build_opener(_NoRedirectHandler()) # ── URL / hostname helpers ──────────────────────────────────────────────────── def _normalise_url(url: str) -> str: """Normalise *url* to an https:// URL (or http:// for loopback addresses). Adds ``https://`` when no scheme is present. Raises ``ValueError`` when an explicit ``http://`` scheme is given for a non-loopback host — sending sending credentials over cleartext HTTP to a remote server is never acceptable. Exception: ``http://localhost``, ``http://127.0.0.1``, and ``http://[::1]`` (with or without a port) are allowed because loopback traffic never leaves the machine and cannot be intercepted in transit. Args: url: Raw user-supplied URL. Returns: Normalised URL without a trailing slash. Raises: ValueError: If the URL uses ``http://`` for a non-loopback host, or uses a disallowed scheme (``file://``, ``ftp://``, etc.). """ stripped = url.strip().rstrip("/") # Check for an explicit URL scheme BEFORE adding https://, catching # file:///etc/passwd, ftp://, data:text/plain, javascript:, etc. # # We can't unconditionally rely on urlparse's scheme detection because # bare host:port inputs like "musehub.ai:8443" are misidentified: # urlparse("musehub.ai:8443").scheme == "musehub.ai". # # Heuristic: if the URL contains "://" OR if the part after the first ":" # is NOT a pure port number (digits), treat the prefix as a scheme. # Examples: # "musehub.ai:8443" → after ":" is "8443" (digits) → host:port, skip # "localhost:8080" → after ":" is "8080" (digits) → host:port, skip # "data:text/plain,x" → after ":" is "text/plain,x" → scheme, check # "javascript:alert()" → after ":" is "alert()" → scheme, check # "https://musehub.ai" → "://" present → scheme, check colon_idx = stripped.find(":") if colon_idx > 0: after_colon = stripped[colon_idx + 1:].lstrip("/") is_port = after_colon.isdigit() # bare port: "8443", "10003", etc. if not is_port: pre_scheme = urllib.parse.urlparse(stripped).scheme.lower() if pre_scheme and pre_scheme not in _ALLOWED_API_SCHEMES: raise ValueError( f"URL scheme '{pre_scheme}' is not allowed. " "Use https:// (or http:// for localhost)." ) if not stripped.startswith(("http://", "https://")): stripped = f"https://{stripped}" if stripped.startswith("http://"): # Use urlparse to correctly extract the hostname — handles IPv6 # bracket notation (e.g. http://[::1]:8080) where manual split(":")[0] # would yield "[" instead of "::1". _loopback = {"localhost", "127.0.0.1", "::1"} parsed_host = urllib.parse.urlparse(stripped).hostname or "" if parsed_host not in _loopback: host = stripped[len("http://"):] raise ValueError( f"Insecure URL rejected: {stripped!r}\n" f"MuseHub requires HTTPS for non-loopback hosts. Did you mean: https://{host}" ) return stripped def _hub_hostname(url: str) -> str: """Extract the display hostname (host:port) from a hub URL.""" stripped = url.strip().rstrip("/") if "://" in stripped: stripped = stripped.split("://", 1)[1] return stripped.split("/")[0] def _ping_hub(url: str) -> tuple[bool, str]: """Attempt an HTTP GET to ``/health``. Returns a ``(reachable, message)`` tuple. Never raises — all errors are captured and surfaced as human-readable strings. Security note ------------- Only ``http`` and ``https`` schemes are attempted. Any other scheme (``file://``, ``ftp://``, etc.) returns ``(False, "scheme not allowed")`` without opening a socket — prevents callers from accidentally probing the local filesystem via a ``--hub`` override. Exception coverage ------------------ - ``urllib.error.HTTPError`` — non-2xx from a reachable server - ``urllib.error.URLError`` — DNS failure, connection refused - ``TimeoutError`` — connect/read timeout - ``OSError`` — network-layer errors (SSL, connection reset, etc.) - ``http.client.HTTPException`` — malformed HTTP response (``BadStatusLine``, ``InvalidURL``) from a misbehaving server; NOT a subclass of OSError so must be caught separately """ # Scheme guard — never probe non-HTTP(S) targets. scheme = urllib.parse.urlparse(url).scheme.lower() if scheme not in _ALLOWED_API_SCHEMES: return False, f"URL scheme '{scheme}' is not allowed (use http or https)" # For localhost HTTPS, build a one-shot opener that trusts the mkcert CA. # _PING_OPENER uses the default CA bundle (no mkcert), so it fails against # the local dev server's self-signed cert. ssl_ctx = _hub_api_ssl_context(url) if ssl_ctx is not None: opener = urllib.request.build_opener( _NoRedirectHandler(), urllib.request.HTTPSHandler(context=ssl_ctx), ) else: opener = _PING_OPENER health_url = f"{url.rstrip('/')}/health" try: req = urllib.request.Request(health_url, method="GET") with opener.open(req, timeout=_CONNECT_TIMEOUT) as resp: status = resp.status if 200 <= status < 300: return True, f"HTTP {status} OK" return False, f"HTTP {status}" except urllib.error.HTTPError as exc: return False, f"HTTP {exc.code} {exc.reason}" except urllib.error.URLError as exc: return False, str(exc.reason) except TimeoutError: return False, "timed out" except OSError as exc: return False, str(exc) except http.client.HTTPException as exc: # Malformed HTTP response (BadStatusLine, InvalidURL, etc.). # Not an OSError subclass — must be caught explicitly. return False, f"malformed response: {type(exc).__name__}" # ── MuseHub HTTP helper ─────────────────────────────────────────────────────── def _hub_api_ssl_context(url: str) -> "ssl.SSLContext | None": """Return an SSL context appropriate for *url*. - ``https://localhost`` / ``https://127.0.0.1``: loads the local-dev self-signed CA cert bundled with musehub so that muse CLI tools work against the local hub without installing the cert system-wide. - All other HTTPS URLs: return ``None``, letting ``urllib`` use the OS CA store (standard verification). - HTTP URLs: return ``None`` (no TLS at all). The CA cert path mirrors ``muse.core.transport._httpx_verify`` so both the push (httpx) and hub-API (urllib) paths trust the same certificate. Path layout (6 levels up from this file reaches the ecosystem root): ``hub/_core.py → hub/ → commands/ → cli/ → muse/ → muse/ → ecosystem/`` """ import ssl from muse.core.transport import _mkcert_ca if url.startswith("https://localhost") or url.startswith("https://127.0.0.1"): ca = _mkcert_ca() if ca is not None: ctx = ssl.create_default_context(cafile=str(ca)) return ctx return None # Read-only HTTP methods that are permitted without an Authorization header so # that unauthenticated clients can access public resources (e.g. reading issues # on a public repo). Mutating methods (POST/PUT/DELETE/PATCH) always require # a signing identity. _READ_ONLY_METHODS: frozenset[str] = frozenset({"GET", "HEAD", "OPTIONS"}) def _hub_api( hub_url: str, identity: IdentityEntry, method: str, path: str, body: Mapping[str, str | bool | list[str] | None] | None = None, timeout: float = 10.0, ) -> _HubApiResponse: """Make a JSON request to the MuseHub API, signing it when credentials exist. For read-only methods (GET/HEAD/OPTIONS) on public resources this function proceeds without an ``Authorization`` header when no signing identity is configured — allowing unauthenticated access to public repos (e.g. reading issues, listing releases). For mutating methods (POST/PUT/DELETE/PATCH) a signing identity is always required and the function exits non-zero when none is found. Security: - Validates the hub URL scheme (``http``/``https`` only) before opening any socket — prevents SSRF if ``config.toml`` is tampered with. - Caps the response body at ``_MAX_API_RESPONSE_BYTES`` to prevent OOM from a hostile or misbehaving hub. - Sanitizes error detail from HTTP responses before printing. - For ``https://localhost`` uses the local-dev CA cert (not CERT_NONE) so self-signed certs are verified against a known CA rather than bypassed. Args: hub_url: Repository-level hub URL (e.g. ``https://localhost:1337/gabriel/muse``). identity: Identity entry from ``~/.muse/identity.toml``. method: HTTP method (``GET``, ``POST``, ``DELETE``, …). path: API path appended to the server root (e.g. ``/gabriel/muse/issues/5``). body: Optional JSON body for POST/PUT requests. timeout: Connect+read timeout in seconds. Returns: Parsed JSON response body as a dict. Raises: SystemExit: On invalid URL scheme, missing token for mutating requests, network errors, or non-2xx responses. """ parsed = urllib.parse.urlparse(hub_url) scheme = parsed.scheme.lower() if scheme not in _ALLOWED_API_SCHEMES: print( f"❌ Hub URL scheme '{sanitize_display(scheme)}' is not allowed. " "Use http or https.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) server_root = f"{parsed.scheme}://{parsed.netloc}" url = f"{server_root}{path}" from muse.cli.config import get_signing_identity signing = get_signing_identity(remote_url=hub_url) normalized_method = method.upper() if not signing and normalized_method not in _READ_ONLY_METHODS: print( "❌ No signing identity for this hub. Run:\n" " muse auth keygen --hub \n" " muse auth register --hub --handle ", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) data: bytes | None = None body_bytes: bytes | None = None if body is not None: body_bytes = json.dumps(body).encode() data = body_bytes headers: Metadata = {"Accept": "application/json"} if signing: from muse.core.msign import build_msign_header headers["Authorization"] = build_msign_header(signing, normalized_method, url, body_bytes) if body_bytes is not None: headers["Content-Type"] = "application/json" ssl_ctx = _hub_api_ssl_context(url) req = urllib.request.Request(url, data=data, headers=headers, method=normalized_method) try: with urllib.request.urlopen(req, timeout=timeout, context=ssl_ctx) as resp: # noqa: S310 raw = resp.read(_MAX_API_RESPONSE_BYTES + 1) if len(raw) > _MAX_API_RESPONSE_BYTES: print( f"❌ Response from {sanitize_display(url)} exceeds size limit.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) text = raw.decode("utf-8", errors="replace") parsed = json.loads(text) if text.strip() else {} return parsed if isinstance(parsed, dict) else {} except urllib.error.HTTPError as exc: raw_body = exc.read().decode(errors="replace") try: detail = json.loads(raw_body).get("detail", raw_body) except (json.JSONDecodeError, AttributeError): detail = raw_body[:200] print( f"❌ MuseHub API error {exc.code}: {sanitize_display(str(detail))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc except urllib.error.URLError as exc: print( f"❌ Cannot reach MuseHub: {sanitize_display(str(exc.reason))}", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) from exc # ── Repo / identity resolution helpers ─────────────────────────────────────── def _enrich_hub_url_from_remote(hub_base_url: str) -> str: """Append owner/slug to a bare hub URL by cross-referencing configured remotes. When the hub URL stored by ``muse hub connect`` is a bare base URL (e.g. ``https://localhost:1337``) with no owner/slug path component, this function searches the configured remotes for one whose netloc matches the hub netloc and extracts the owner/slug from its path. Returns the enriched URL on a match; returns *hub_base_url* unchanged when no matching remote is found or when the caller is not inside a repo. """ base_parsed = urllib.parse.urlparse(hub_base_url) if len(base_parsed.path.strip("/").split("/")) >= 2 and base_parsed.path.strip("/"): # Already has owner/slug — nothing to do. return hub_base_url root = find_repo_root() if root is None: return hub_base_url for remote in list_remotes(root): r = urllib.parse.urlparse(remote["url"]) if r.netloc == base_parsed.netloc: parts = r.path.strip("/").split("/") if len(parts) >= 2 and parts[0]: return f"{base_parsed.scheme}://{base_parsed.netloc}/{parts[0]}/{parts[1]}" return hub_base_url def _resolve_repo_id(hub_url: str, identity: IdentityEntry) -> str: """Look up the MuseHub repo ID from the hub URL's owner/slug path. The hub URL is ``http://host/owner/slug``. This calls ``GET /{owner}/{slug}/refs`` to obtain the ID, falling back to the search endpoint if the refs response does not include ``repo_id``. When the hub URL is a bare base URL (no owner/slug), the function first attempts to enrich it from the configured remotes via :func:`_enrich_hub_url_from_remote`. Raises: SystemExit: If the URL has no owner/slug or the repo ID cannot be resolved. """ hub_url = _enrich_hub_url_from_remote(hub_url) parsed = urllib.parse.urlparse(hub_url) parts = parsed.path.strip("/").split("/") if len(parts) < 2: print( f"❌ Hub URL '{sanitize_display(hub_url)}' does not contain " "owner/slug — cannot resolve repo ID.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) owner, slug = parts[0], parts[1] data = _hub_api(hub_url, identity, "GET", f"/{owner}/{slug}/refs") repo_id: str = str(data.get("repo_id", "")) if not repo_id: # Fallback: search endpoint search = _hub_api( hub_url, identity, "GET", f"/api/search?q={urllib.parse.quote(slug)}&limit=5", ) repos_val = search.get("repos", []) repos_list: list[_RepoEntry] = ( [r for r in repos_val if isinstance(r, dict)] if isinstance(repos_val, list) else [] ) for repo in repos_list: if repo.get("owner") == owner and repo.get("slug") == slug: repo_id = str(repo.get("repoId", "")) break if not repo_id: print( f"❌ Could not resolve repo ID for {sanitize_display(owner)}/{sanitize_display(slug)}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) return repo_id def _resolve_repo_id_from_owner_repo( hub_url: str, identity: IdentityEntry, owner_repo: str, ) -> str: """Resolve a repo ID from an ``owner/repo`` string. Calls ``GET /{owner}/{slug}/refs`` to obtain the ID. Args: hub_url: Base hub URL (e.g. ``https://localhost:1337``). identity: Caller identity for auth. owner_repo: String in ``owner/slug`` format (e.g. ``gabriel/muse``). Raises: SystemExit: If owner_repo is malformed or the repo cannot be resolved. """ parts = owner_repo.strip("/").split("/") if len(parts) != 2: print(f"❌ --repo must be in owner/repo format, got {owner_repo!r}.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) owner, slug = parts[0], parts[1] # Build a repo-scoped URL to piggyback on _resolve_repo_id base = hub_url.split("/api")[0].split("/#")[0].rstrip("/") repo_hub_url = f"{base}/{owner}/{slug}" return _resolve_repo_id(repo_hub_url, identity) def _resolve_body(args: argparse.Namespace) -> str: """Return the effective body string from ``--body`` or ``--body-file``. Resolution order: 1. ``--body-file PATH`` — read the entire file at PATH as UTF-8. Pass ``-`` to read from stdin instead. 2. ``--body TEXT`` — use the literal string. 3. (neither) — return ``""`` (empty string). If both are provided ``--body-file`` wins and ``--body`` is ignored. This helper exists because large bodies (markdown, code blocks, ASCII art) cannot survive shell quoting when passed inline via ``--body``. Writing the body to a temp file and using ``--body-file`` is the ergonomic path for agents and humans alike. Args: args: Parsed :class:`argparse.Namespace`. Must have ``body`` and ``body_file`` attributes (both optional/defaulting to ``None``). Returns: Body string (may be empty). Raises: SystemExit: If ``--body-file PATH`` cannot be read (file not found, permission denied, etc.). """ import os body_file: str | None = getattr(args, "body_file", None) body: str = getattr(args, "body", "") or "" if not body_file: return body if body_file == "-": return sys.stdin.read() try: with open(body_file, encoding="utf-8") as fh: return fh.read() except OSError as exc: print(f"❌ Cannot read --body-file {body_file!r}: {exc}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) from exc def _validate_assignee(handle: str, *, allow_empty: bool = False) -> None: """Validate a MuseHub handle before sending it to the API. Called on every user-supplied handle — at create time (``--assignee``), update time (``--assign``), and the dedicated ``issue assign`` command. Threat model ------------ Handles are echoed back to the terminal and embedded in API requests. Malformed values could cause: * **Terminal injection** — ANSI escape sequences hidden inside handle strings can rewrite terminal output after the fact. * **Null-byte truncation** — ``\\x00`` in a string truncates C-string paths, and some JSON parsers mis-handle it. * **Newline injection** — ``\\n`` in a value echoed to stdout/stderr can make error messages look like success messages. * **Control-character abuse** — backspace, carriage-return, etc. can visually overwrite earlier output. All of the above are rejected before any network I/O or terminal output. Args: handle: The handle string to validate. Trailing/leading whitespace has already been stripped by argparse. allow_empty: When ``True`` (used by ``issue assign``), an empty string is accepted and means "unassign". When ``False`` (used by ``issue create``), an empty string is an error. Raises: SystemExit(ExitCode.USER_ERROR): On any validation failure. """ if not handle: if allow_empty: return print("❌ --assignee must not be empty.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # Reject null bytes and every ASCII control character (0x00–0x1F, 0x7F). # These can cause terminal injection, string truncation, or newline spoofing. if any(ord(c) < 0x20 or ord(c) == 0x7F for c in handle): print( "❌ --assignee contains a control character (null byte, newline, " "etc.) which is not allowed in handles.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Reject non-ASCII entirely — Unicode confusables and RTL override # characters can impersonate other handles visually. if not handle.isascii(): print( f"❌ --assignee {sanitize_display(handle)!r} contains non-ASCII " "characters; handles must be plain ASCII.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if len(handle) > _MAX_HANDLE_LEN: print( f"❌ --assignee is too long ({len(handle)} chars); " f"maximum is {_MAX_HANDLE_LEN}.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if not _HANDLE_RE.match(handle): print( f"❌ --assignee {sanitize_display(handle)!r} is not a valid handle. " "Handles must start with a letter or digit and contain only " "letters, digits, hyphens (-), and underscores (_).", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) def _get_hub_and_identity( remote: str | None = None, hub_url_override: str | None = None, ) -> tuple[str, IdentityEntry]: """Load hub URL and identity for the current repo, or exit with a helpful error. Falls back to a named remote's URL when no hub is configured — useful for local development where the remote IS the hub. When *hub_url_override* is provided it takes precedence over the hub URL stored in ``.muse/config.toml``. This is the value supplied via the ``--hub`` CLI flag and lets callers (e.g. containerised agents) target a hub at a different address than the one recorded in the repo config without having to modify ``config.toml``. Raises: SystemExit: If not inside a repo, no hub/remote is configured, or not authenticated. """ root = find_repo_root() if root is None: print("❌ Not inside a Muse repository.", file=sys.stderr) raise SystemExit(ExitCode.REPO_NOT_FOUND) hub_url = hub_url_override or get_hub_url(root) if hub_url is None: # No hub configured — try the specified remote, then 'local', then any remote. candidate_name = remote or "local" candidate_url = get_remote(candidate_name, root) if candidate_url is None and remote is None: remotes = list_remotes(root) candidate_url = remotes[0]["url"] if remotes else None if candidate_url: candidate_name = remotes[0]["name"] if candidate_url: hub_url = candidate_url logger.debug( "No hub configured — using remote '%s' as hub URL: %s", candidate_name, hub_url, ) else: print("❌ No hub connected and no remotes configured.", file=sys.stderr) print( " Run `muse hub connect ` or `muse remote add local `.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) identity = load_identity(hub_url) if identity is None: print( f"❌ Not authenticated for {sanitize_display(hub_url)}.", file=sys.stderr, ) print(f" Run: muse auth register --hub {hub_url}", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) return hub_url, identity def _get_hub_and_optional_identity( remote: str | None = None, hub_url_override: str | None = None, ) -> tuple[str, IdentityEntry]: """Like ``_get_hub_and_identity`` but returns an empty identity on missing auth. Use this for read-only commands (``hub issue read``, ``hub issue list``, …) that should work against public repos even when no signing identity is configured. The empty identity dict is passed to ``_hub_api``, which will omit the Authorization header for GET requests — allowing unauthenticated access to public resources. When an identity IS configured it is returned normally, so authenticated callers get signed requests and can access private repos. Mutating operations must use ``_get_hub_and_identity`` (strict) to fail fast with a helpful error before making any network request. Returns: ``(hub_url, identity)`` where *identity* may be ``{"type": "human"}`` when no signing identity is configured. """ import io as _io from muse.core.repo import find_repo_root from muse.cli.config import get_hub_url, get_remote, list_remotes # Resolve the hub URL first (independent of identity). root = find_repo_root() hub_url: str | None = hub_url_override or (get_hub_url(root) if root else None) if hub_url is None and root: candidate_name = remote or "local" candidate_url = get_remote(candidate_name, root) if candidate_url is None and remote is None: remotes = list_remotes(root) candidate_url = remotes[0]["url"] if remotes else None hub_url = candidate_url if not hub_url: print("❌ No hub connected and no remotes configured.", file=sys.stderr) print( " Run `muse hub connect ` or `muse remote add local `.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) # Try loading the identity; silently fall back to empty dict on failure so # callers can proceed with unauthenticated GET requests on public resources. identity = load_identity(hub_url) if identity is None: empty_identity: IdentityEntry = {"type": "human"} return hub_url, empty_identity return hub_url, identity def _resolve_proposal_id( hub_url: str, identity: IdentityEntry, repo_id: str, proposal_id_or_prefix: str, ) -> str: """Resolve a proposal ID prefix to a full ID. If *proposal_id_or_prefix* looks like a full ID (≥32 chars with hyphens), it is returned as-is. Otherwise the most recent ``_PROPOSAL_PREFIX_RESOLVE_LIMIT`` proposals are fetched and the first match whose ``proposalId`` starts with the prefix is returned. All user-supplied and API-sourced strings are sanitized before printing. Note ---- Prefix resolution is capped at :data:`_PROPOSAL_PREFIX_RESOLVE_LIMIT` proposals. On repositories with many more open proposals, very old proposals may not be findable by prefix — pass the full ID in that case. Raises: SystemExit: When no match is found or multiple proposals match the prefix. """ # Recognise a full content-addressed ID (sha256:<64-hex>, blake3:<64-hex>) # or a legacy hyphenated UUID — return as-is without fetching the list. try: split_id(proposal_id_or_prefix) return proposal_id_or_prefix except ValueError: pass if "-" in proposal_id_or_prefix and len(proposal_id_or_prefix) >= 32: return proposal_id_or_prefix # legacy UUID passthrough data = _hub_api( hub_url, identity, "GET", f"/api/repos/{repo_id}/proposals?limit={_PROPOSAL_PREFIX_RESOLVE_LIMIT}", ) proposals_val = data.get("proposals", []) proposals: list[_ProposalEntry] = ( [r for r in proposals_val if isinstance(r, dict)] if isinstance(proposals_val, list) else [] ) matches = [p for p in proposals if str(p.get("proposalId", "")).startswith(proposal_id_or_prefix)] if not matches: print( f"❌ No proposal found with ID prefix '{sanitize_display(proposal_id_or_prefix)}'.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if len(matches) > 1: print( f"❌ Ambiguous prefix '{sanitize_display(proposal_id_or_prefix)}' " f"— matches {len(matches)} proposals:", file=sys.stderr, ) for m in matches: print( f" {sanitize_display(str(m.get('proposalId', '')))} " f"{sanitize_display(str(m.get('title', '')))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) return str(matches[0].get("proposalId", proposal_id_or_prefix)) def _format_proposal(entry: _ProposalEntry, *, verbose: bool = False) -> str: """Format a single proposal for human-readable display. All fields from the API response are sanitized via :func:`sanitize_display` before inclusion in the returned string — including ``author`` and ``createdAt`` in verbose mode. Args: entry: Raw proposal dict from the MuseHub API. verbose: When ``True``, append author name and creation date (YYYY-MM-DD). """ proposal_id = sanitize_display(str(entry.get("proposalId", ""))) title = sanitize_display(str(entry.get("title", "(no title)"))) state = str(entry.get("state", "?")) from_b = sanitize_display(str(entry.get("fromBranch", "?"))) to_b = sanitize_display(str(entry.get("toBranch", "?"))) state_icon = {"open": "🟢", "merged": "🟣", "closed": "⛔"}.get(state, "❓") line = f" {state_icon} {proposal_id} {from_b} → {to_b} {title}" if verbose: author = sanitize_display(str(entry.get("author", ""))) created = sanitize_display(str(entry.get("createdAt", ""))[:10]) line += f"\n by {author or '?'} {created}" return line def _resolve_hub_override(args: argparse.Namespace) -> str | None: """Return the effective hub URL override, resolving --repo as an alias for --hub. ``--repo owner/repo`` is the idiomatic git/gh CLI style. When provided, the hub base URL is read from the repo config and the owner/repo path is appended, producing a full ``--hub``-compatible URL. """ if args.hub: return args.hub repo_arg: str | None = getattr(args, "repo", None) if not repo_arg: return None root = find_repo_root() if root is not None: configured = get_hub_url(root) if configured: # Strip any trailing owner/repo path — keep only scheme://host[:port]. parsed = urllib.parse.urlparse(configured) base = f"{parsed.scheme}://{parsed.netloc}" return f"{base}/{repo_arg.strip('/')}" # Fallback: no config — caller will get a helpful error from _get_hub_and_identity. return None __all__ = ['logger', '_CONNECT_TIMEOUT', '_MAX_API_RESPONSE_BYTES', '_ALLOWED_API_SCHEMES', '_PROPOSAL_PREFIX_RESOLVE_LIMIT', '_MAX_PROPOSAL_BODY_LINES', '_MAX_PROPOSAL_TITLE_LEN', '_MAX_ISSUE_TITLE_LEN', '_MAX_REPO_NAME_LEN', '_MAX_REPO_DESC_LEN', '_MAX_ISSUE_LABEL_LEN', '_MAX_ISSUE_COMMENT_LEN', '_MAX_LABEL_NAME_LEN', '_MAX_LABEL_DESC_LEN', '_MAX_HANDLE_LEN', '_HANDLE_RE', '_ConnectJson', '_StatusJson', '_DisconnectJson', '_PingJson', '_RepoEntry', '_ProposalEntry', '_RepoJson', '_HubApiResponse', '_LabelEntry', '_NoRedirectHandler', '_PING_OPENER', '_normalise_url', '_hub_hostname', '_ping_hub', '_hub_api', '_enrich_hub_url_from_remote', '_resolve_repo_id', '_resolve_repo_id_from_owner_repo', '_resolve_body', '_validate_assignee', '_get_hub_and_identity', '_get_hub_and_optional_identity', '_resolve_proposal_id', '_format_proposal', '_resolve_hub_override', 'argparse', 'http', 'json', 'logging', 're', 'sys', 'textwrap', 'urllib', 'Mapping', 'IO', 'TypedDict', 'clear_hub_url', 'get_hub_url', 'get_remote', 'list_remotes', 'set_hub_url', 'ExitCode', 'IdentityEntry', 'load_identity', 'find_repo_root', 'read_current_branch', 'Metadata', 'short_id', 'clamp_int', 'sanitize_display', 'EnvelopeJson', 'make_envelope', 'start_timer', '_HubPayload', '_ProposalPayload']