"""Muse CLI configuration helpers. Reads and writes ``.muse/config.toml`` — the per-repository configuration file. Credentials (signing identities) are **not** stored here; they live in ``~/.muse/identity.toml`` managed by :mod:`muse.core.identity`. Config schema ------------- :: [user] name = "Alice" # display name (human or agent handle) email = "a@example.com" type = "human" # "human" | "agent" [hub] url = "https://musehub.ai" # MuseHub fabric endpoint for this repo [remotes.origin] url = "https://hub.muse.io/repos/my-repo" branch = "main" [domain] # Domain-specific key/value pairs; read by the active domain plugin. # ticks_per_beat = "480" Settable via ``muse config set`` --------------------------------- - ``user.name``, ``user.email``, ``user.type`` - ``hub.url`` (alias: ``muse hub connect ``) - ``domain.*`` Not settable via ``muse config set`` -------------------------------------- - ``remotes.*`` — use ``muse remote add/remove`` - credentials — use ``muse auth register`` Token resolution ---------------- :func:`get_signing_identity` reads the hub URL from this file, then resolves the signing identity from ``~/.muse/identity.toml`` via :func:`muse.core.identity.resolve_token`. The token is **never** logged. """ from __future__ import annotations import logging import pathlib import shutil import subprocess import tomllib from typing import TypedDict from muse.core.store import write_text_atomic logger = logging.getLogger(__name__) type RemotesMap = dict[str, RemoteEntry] # remote_name → remote entry type DomainConfig = dict[str, str] # domain key → value type ConfigSection = dict[str, str] # generic flattened section dict type ConfigTree = dict[str, dict[str, str]] # section → key → value (for JSON output) type DefaultsMap = dict[str, int] # config key → default int value _CONFIG_FILENAME = "config.toml" _MUSE_DIR = ".muse" # --------------------------------------------------------------------------- # Named configuration types # --------------------------------------------------------------------------- class UserConfig(TypedDict, total=False): """``[user]`` section in ``.muse/config.toml``.""" name: str email: str type: str # "human" | "agent" class HubConfig(TypedDict, total=False): """``[hub]`` section in ``.muse/config.toml``.""" url: str class RemoteEntry(TypedDict, total=False): """``[remotes.]`` section in ``.muse/config.toml``.""" url: str branch: str pack_origin: str class LimitsConfig(TypedDict, total=False): """``[limits]`` section in ``.muse/config.toml``. All values are optional — defaults are used when absent. Keys map to the ``[limits]`` TOML table:: [limits] max_walk_commits = 10000 # cap for walk_commits_between / muse log max_ancestors = 50000 # cap for find_merge_base BFS max_graph_commits = 50000 # cap for _collect_all_commits (--graph --all) shard_prefix_length = 2 # object store shard depth: 2 (256 shards) # or 4 (65536 shards) for very large repos """ max_walk_commits: int max_ancestors: int max_graph_commits: int shard_prefix_length: int class MuseConfig(TypedDict, total=False): """Structured view of the entire ``.muse/config.toml`` file.""" user: UserConfig hub: HubConfig remotes: RemotesMap domain: DomainConfig limits: LimitsConfig class RemoteConfig(TypedDict, total=False): """Public-facing remote descriptor returned by :func:`list_remotes`. ``pack_origin`` is optional. When set, pack uploads are sent to ``{pack_origin}/{owner}/{slug}/push/object-pack`` instead of the default ``{url}/push/object-pack``. Use this to route small-object packs through a Cloudflare Worker (or equivalent edge cache) without changing the primary remote URL. """ name: str # always present url: str # always present pack_origin: str # optional — overrides the origin for pack uploads # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _config_path(repo_root: pathlib.Path | None) -> pathlib.Path: """Return the path to .muse/config.toml for the given (or cwd) root.""" root = (repo_root or pathlib.Path.cwd()).resolve() return root / _MUSE_DIR / _CONFIG_FILENAME def _load_config(config_path: pathlib.Path) -> MuseConfig: """Load and parse config.toml; return an empty MuseConfig if absent.""" if not config_path.is_file(): return {} try: with config_path.open("rb") as fh: raw = tomllib.load(fh) except Exception as exc: # noqa: BLE001 logger.warning("⚠️ Failed to parse %s: %s", config_path, exc) return {} config: MuseConfig = {} user_raw = raw.get("user") if isinstance(user_raw, dict): user: UserConfig = {} name_val = user_raw.get("name") if isinstance(name_val, str): user["name"] = name_val email_val = user_raw.get("email") if isinstance(email_val, str): user["email"] = email_val type_val = user_raw.get("type") if isinstance(type_val, str): user["type"] = type_val config["user"] = user hub_raw = raw.get("hub") if isinstance(hub_raw, dict): hub: HubConfig = {} url_val = hub_raw.get("url") if isinstance(url_val, str): hub["url"] = url_val config["hub"] = hub remotes_raw = raw.get("remotes") if isinstance(remotes_raw, dict): remotes: RemotesMap = {} for name, remote_raw in remotes_raw.items(): if isinstance(remote_raw, dict): entry: RemoteEntry = {} rurl = remote_raw.get("url") if isinstance(rurl, str): entry["url"] = rurl branch_val = remote_raw.get("branch") if isinstance(branch_val, str): entry["branch"] = branch_val pack_origin_val = remote_raw.get("pack_origin") if isinstance(pack_origin_val, str): entry["pack_origin"] = pack_origin_val remotes[name] = entry config["remotes"] = remotes domain_raw = raw.get("domain") if isinstance(domain_raw, dict): domain: DomainConfig = {} for key, val in domain_raw.items(): if isinstance(val, str): domain[key] = val config["domain"] = domain limits_raw = raw.get("limits") if isinstance(limits_raw, dict): limits: LimitsConfig = {} mwc = limits_raw.get("max_walk_commits") if isinstance(mwc, int) and mwc > 0: limits["max_walk_commits"] = mwc ma = limits_raw.get("max_ancestors") if isinstance(ma, int) and ma > 0: limits["max_ancestors"] = ma mgc = limits_raw.get("max_graph_commits") if isinstance(mgc, int) and mgc > 0: limits["max_graph_commits"] = mgc spl = limits_raw.get("shard_prefix_length") if isinstance(spl, int) and spl in (2, 4): limits["shard_prefix_length"] = spl config["limits"] = limits return config def _escape(value: str) -> str: """Escape a TOML basic string value (backslash and double-quote only). TOML basic strings allow control characters escaped as ``\\n``, ``\\t``, etc., but we store only printable content — control characters in values are also stripped here so the resulting TOML file remains parseable. """ return ( value.replace("\\", "\\\\") .replace('"', '\\"') .replace("\n", "\\n") .replace("\r", "\\r") .replace("\0", "") ) # Characters that are structurally significant in unquoted TOML keys and # table headers. Any of these in a key name would allow injection of # arbitrary TOML sections or key-value pairs. _TOML_KEY_UNSAFE: frozenset[str] = frozenset('\n\r\0][="') def _validate_toml_key(key: str, context: str = "key") -> None: """Raise ``ValueError`` if *key* contains TOML-structurally unsafe characters. Prevents injection attacks where a crafted key like ``x]\\n[evil`` would break the TOML section structure and allow writing arbitrary sections. Args: key: Key string to validate. context: Human-readable label used in the error message (e.g. ``"domain key"``). Raises: ValueError: If any character in *key* is in ``_TOML_KEY_UNSAFE``. """ bad = _TOML_KEY_UNSAFE & set(key) if bad: chars = ", ".join(sorted(repr(c) for c in bad)) raise ValueError( f"Config {context} {key!r} contains characters not allowed in TOML keys: {chars}" ) def _dump_toml(config: MuseConfig) -> str: """Serialise a MuseConfig to TOML text. Section order: ``[user]``, ``[hub]``, ``[remotes.*]``, ``[domain]``, ``[limits]``. All key names are validated against ``_TOML_KEY_UNSAFE`` before being written, preventing TOML injection via crafted domain keys or remote names. """ lines: list[str] = [] user = config.get("user") if user: lines.append("[user]") name = user.get("name", "") if name: lines.append(f'name = "{_escape(name)}"') email = user.get("email", "") if email: lines.append(f'email = "{_escape(email)}"') utype = user.get("type", "") if utype: lines.append(f'type = "{_escape(utype)}"') lines.append("") hub = config.get("hub") if hub: lines.append("[hub]") url = hub.get("url", "") if url: lines.append(f'url = "{_escape(url)}"') lines.append("") remotes = config.get("remotes") or {} for remote_name in sorted(remotes): # Remote names come from _load_config which parses TOML, so they are # safe at read time. Validate defensively before writing. _validate_toml_key(remote_name, "remote name") entry = remotes[remote_name] lines.append(f"[remotes.{remote_name}]") rurl = entry.get("url", "") if rurl: lines.append(f'url = "{_escape(rurl)}"') branch = entry.get("branch", "") if branch: lines.append(f'branch = "{_escape(branch)}"') lines.append("") domain = config.get("domain") or {} if domain: lines.append("[domain]") for key, val in sorted(domain.items()): _validate_toml_key(key, "domain key") lines.append(f'{key} = "{_escape(val)}"') lines.append("") limits = config.get("limits") or {} if limits: lines.append("[limits]") mwc = limits.get("max_walk_commits") if mwc is not None: lines.append(f"max_walk_commits = {mwc}") ma = limits.get("max_ancestors") if ma is not None: lines.append(f"max_ancestors = {ma}") mgc = limits.get("max_graph_commits") if mgc is not None: lines.append(f"max_graph_commits = {mgc}") spl = limits.get("shard_prefix_length") if spl is not None: lines.append(f"shard_prefix_length = {spl}") lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Auth token resolution (via identity store) # --------------------------------------------------------------------------- def get_signing_identity( repo_root: pathlib.Path | None = None, remote_url: str | None = None, agent_id: str | None = None, ) -> "object | None": """Return a :class:`~muse.core.transport.SigningIdentity` for a hub, or ``None``. Resolution order: 1. ``MUSE_AGENT_KEY`` environment variable — PEM private key bytes. When set the key is used directly, bypassing the identity store entirely. The handle is taken from ``MUSE_AGENT_HANDLE`` (defaults to *agent_id* if that is also set, otherwise ``"agent"``). This is the injection mechanism for agent subprocesses spawned by agentception. 2. Agent-specific entry in ``~/.muse/identity.toml`` keyed by ``"hostname#agent_id"`` — when *agent_id* is provided. 3. Human entry in ``~/.muse/identity.toml`` keyed by bare hostname. 4. Hub URL from ``[hub] url`` in ``.muse/config.toml`` (fallback lookup URL when *remote_url* is not supplied). The private key is **never** logged. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. remote_url: URL of the specific remote being contacted. agent_id: Agent handle, e.g. ``"agentception-abc123"``. Used to try an agent-specific key before falling back to the human key. Returns: :class:`~muse.core.transport.SigningIdentity` or ``None``. """ import os as _os from muse.core.identity import resolve_signing_identity # avoid circular import from muse.core.transport import SigningIdentity # 1. MUSE_AGENT_KEY env var — injected by agentception at spawn time. agent_pem = _os.environ.get("MUSE_AGENT_KEY", "").strip() if agent_pem: from muse.core.keypair import load_private_key_from_pem private_key = load_private_key_from_pem(agent_pem.encode()) if private_key is not None: handle = ( _os.environ.get("MUSE_AGENT_HANDLE", "").strip() or agent_id or "agent" ) logger.debug("✅ Signing identity from MUSE_AGENT_KEY (handle=%s)", handle) return SigningIdentity(handle=handle, private_key=private_key) logger.warning("⚠️ MUSE_AGENT_KEY is set but could not be decoded — ignoring") # 2 & 3. Identity store lookup (agent key → human key fallback). lookup_url: str | None = remote_url or get_hub_url(repo_root) if lookup_url is None: logger.debug("⚠️ No hub configured — skipping signing identity lookup") return None result = resolve_signing_identity(lookup_url, agent_id=agent_id) if result is None: logger.debug( "⚠️ No signing identity for hub %s — run `muse auth keygen && muse auth register`", lookup_url, ) return None handle, private_key = result logger.debug("✅ Signing identity resolved for hub %s (handle=%s)", lookup_url, handle) return SigningIdentity(handle=handle, private_key=private_key) # --------------------------------------------------------------------------- # Hub helpers # --------------------------------------------------------------------------- def get_hub_url(repo_root: pathlib.Path | None = None) -> str | None: """Return the hub URL from ``[hub] url``, or ``None`` if not configured. Resolution order: 1. ``/.muse/config.toml`` — repo-local config (highest priority). 2. ``~/.muse/config.toml`` — global user config (fallback). This fallback ensures ``muse auth whoami`` and other hub-aware commands work without ``--hub`` even when invoked outside a repository, as long as the user has set a default hub in their global config. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: URL string, or ``None``. """ config = _load_config(_config_path(repo_root)) hub = config.get("hub") if hub is not None: url = hub.get("url", "") if url.strip(): return url.strip() # Fall back to ~/.muse/config.toml so hub-aware commands (e.g. `muse auth # whoami`) work without --hub when invoked outside a repository. global_config = _load_config(_GLOBAL_CONFIG_FILE) global_hub = global_config.get("hub") if global_hub is not None: url = global_hub.get("url", "") if url.strip(): return url.strip() return None def set_hub_url(url: str, repo_root: pathlib.Path | None = None) -> None: """Write ``[hub] url`` to ``.muse/config.toml``. Preserves all other sections. Creates the config file if absent. Rejects ``http://`` URLs — Muse never contacts a hub over cleartext HTTP. Args: url: Hub URL (must be ``https://``). repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: ValueError: If *url* does not use the ``https://`` scheme. """ _is_loopback = url.startswith("http://localhost") or url.startswith("http://127.0.0.1") or url.startswith("http://[::1]") if not url.startswith("https://") and not _is_loopback: raise ValueError( f"Hub URL must use HTTPS. Got: {url!r}\n" "Muse never connects to a hub over cleartext HTTP.\n" "(Exception: http://localhost and http://127.0.0.1 are allowed for local development.)" ) cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) config["hub"] = HubConfig(url=url) write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Hub URL set to %s", url) def clear_hub_url(repo_root: pathlib.Path | None = None) -> None: """Remove the ``[hub]`` section from ``.muse/config.toml``. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. """ cp = _config_path(repo_root) config = _load_config(cp) if "hub" in config: del config["hub"] write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Hub disconnected") # --------------------------------------------------------------------------- # User config helpers # --------------------------------------------------------------------------- def set_user_field(key: str, value: str, repo_root: pathlib.Path | None = None) -> None: """Set a single ``[user]`` field by name. Allowed keys: ``name``, ``email``, ``type``. Args: key: Field name within ``[user]``. value: New value. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: ValueError: If *key* is not a recognised user config field. """ if key not in {"name", "email", "type"}: raise ValueError(f"Unknown [user] config key: {key!r}. Valid keys: name, email, type") cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) user: UserConfig = config.get("user") or {} if key == "name": user["name"] = value elif key == "email": user["email"] = value elif key == "type": user["type"] = value config["user"] = user write_text_atomic(cp, _dump_toml(config)) logger.info("✅ user.%s = %r", key, value) # --------------------------------------------------------------------------- # Generic dotted-key helpers # --------------------------------------------------------------------------- _BlockedNS = dict[str, str] _BLOCKED_NAMESPACES: _BlockedNS = { "auth": "Use `muse auth keygen` and `muse auth register` to manage credentials.", "remotes": "Use `muse remote add/remove/rename` to manage remotes.", } _SETTABLE_NAMESPACES = {"user", "hub", "domain", "limits"} # Default cap values — used when the [limits] section is absent or the key # is not set. These are the same values that were previously hardcoded inside # the individual functions. _DEFAULT_MAX_WALK_COMMITS: int = 10_000 _DEFAULT_MAX_ANCESTORS: int = 50_000 _DEFAULT_MAX_GRAPH_COMMITS: int = 50_000 _DEFAULT_SHARD_PREFIX_LENGTH: int = 2 def get_limit(key: str, repo_root: pathlib.Path | None = None) -> int: """Return a ``[limits]`` integer cap from config, or its default. Args: key: Limit key — one of ``max_walk_commits``, ``max_ancestors``, ``max_graph_commits``. repo_root: Repository root; ``None`` falls back to ``Path.cwd()``. Returns: Configured integer value, or the built-in default if not set. """ defaults: DefaultsMap = { "max_walk_commits": _DEFAULT_MAX_WALK_COMMITS, "max_ancestors": _DEFAULT_MAX_ANCESTORS, "max_graph_commits": _DEFAULT_MAX_GRAPH_COMMITS, "shard_prefix_length": _DEFAULT_SHARD_PREFIX_LENGTH, } default = defaults.get(key, 10_000) config = _load_config(_config_path(repo_root)) limits = config.get("limits") or {} # Explicit key dispatch keeps mypy happy on TypedDict literal-required keys. if key == "max_walk_commits": val: int | None = limits.get("max_walk_commits") elif key == "max_ancestors": val = limits.get("max_ancestors") elif key == "max_graph_commits": val = limits.get("max_graph_commits") elif key == "shard_prefix_length": val = limits.get("shard_prefix_length") else: val = None if isinstance(val, int) and val > 0: return val return default def get_config_value(key: str, repo_root: pathlib.Path | None = None) -> str | None: """Get a config value by dotted key (e.g. ``user.name``, ``hub.url``). Returns ``None`` when the key is not set or the namespace is unknown. Args: key: Dotted key in ``.`` form. repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: String value, or ``None``. """ parts = key.split(".", 1) if len(parts) != 2: return None namespace, subkey = parts config = _load_config(_config_path(repo_root)) if namespace == "user": user = config.get("user") or {} if subkey == "name": return user.get("name") if subkey == "email": return user.get("email") if subkey == "type": return user.get("type") return None if namespace == "hub": hub = config.get("hub") or {} if subkey == "url": return hub.get("url") return None if namespace == "domain": domain = config.get("domain") or {} return domain.get(subkey) if namespace == "limits": limits = config.get("limits") or {} if subkey == "max_walk_commits": v = limits.get("max_walk_commits") return str(v) if isinstance(v, int) else None if subkey == "max_ancestors": v = limits.get("max_ancestors") return str(v) if isinstance(v, int) else None if subkey == "max_graph_commits": v = limits.get("max_graph_commits") return str(v) if isinstance(v, int) else None if subkey == "shard_prefix_length": v = limits.get("shard_prefix_length") return str(v) if isinstance(v, int) else None return None return None def set_config_value(key: str, value: str, repo_root: pathlib.Path | None = None) -> None: """Set a config value by dotted key (e.g. ``user.name``, ``domain.ticks_per_beat``). Args: key: Dotted key in ``.`` form. value: New string value. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: ValueError: If the namespace is blocked, unknown, or the subkey is invalid. """ parts = key.split(".", 1) if len(parts) != 2: raise ValueError(f"Key must be in 'namespace.subkey' form, got: {key!r}") namespace, subkey = parts if namespace in _BLOCKED_NAMESPACES: raise ValueError(_BLOCKED_NAMESPACES[namespace]) if namespace not in _SETTABLE_NAMESPACES: raise ValueError( f"Unknown config namespace {namespace!r}. " f"Settable namespaces: {', '.join(sorted(_SETTABLE_NAMESPACES))}" ) cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) if namespace == "user": set_user_field(subkey, value, repo_root) return if namespace == "hub": if subkey != "url": raise ValueError(f"Unknown [hub] config key: {subkey!r}. Valid keys: url") # Route through set_hub_url — it enforces the HTTPS requirement. set_hub_url(value, repo_root) return if namespace == "limits": _LIMITS_KEYS = frozenset({ "max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length", }) if subkey not in _LIMITS_KEYS: raise ValueError( f"Unknown [limits] config key: {subkey!r}. " f"Valid keys: {', '.join(sorted(_LIMITS_KEYS))}" ) try: int_value = int(value) except ValueError as exc: raise ValueError( f"[limits] {subkey} must be an integer, got: {value!r}" ) from exc if int_value <= 0: raise ValueError(f"[limits] {subkey} must be a positive integer, got: {int_value}") if subkey == "shard_prefix_length" and int_value not in (2, 4): raise ValueError("shard_prefix_length must be 2 or 4") limits_section: LimitsConfig = config.get("limits") or {} if subkey == "max_walk_commits": limits_section["max_walk_commits"] = int_value elif subkey == "max_ancestors": limits_section["max_ancestors"] = int_value elif subkey == "max_graph_commits": limits_section["max_graph_commits"] = int_value elif subkey == "shard_prefix_length": limits_section["shard_prefix_length"] = int_value config["limits"] = limits_section write_text_atomic(cp, _dump_toml(config)) logger.info("✅ limits.%s = %d", subkey, int_value) return # namespace == "domain" _validate_toml_key(subkey, "domain key") domain: DomainConfig = config.get("domain") or {} domain[subkey] = value config["domain"] = domain write_text_atomic(cp, _dump_toml(config)) logger.info("✅ domain.%s = %r", subkey, value) def config_as_dict(repo_root: pathlib.Path | None = None) -> ConfigTree: """Return the full config as a plain ``dict[str, dict[str, str]]`` for JSON output. Credentials are never included — the hub section only contains the URL. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: Nested dict suitable for ``json.dumps``. """ config = _load_config(_config_path(repo_root)) result: ConfigTree = {} user = config.get("user") if user: user_dict: ConfigSection = {} uname = user.get("name") if uname: user_dict["name"] = uname uemail = user.get("email") if uemail: user_dict["email"] = uemail utype = user.get("type") if utype: user_dict["type"] = utype if user_dict: result["user"] = user_dict hub = config.get("hub") if hub: hub_url = hub.get("url", "") if hub_url: result["hub"] = {"url": hub_url} remotes = config.get("remotes") or {} if remotes: remotes_dict: ConfigSection = {} for rname, entry in sorted(remotes.items()): url = entry.get("url", "") if url: remotes_dict[rname] = url if remotes_dict: result["remotes"] = remotes_dict domain = config.get("domain") or {} if domain: result["domain"] = dict(sorted(domain.items())) limits = config.get("limits") or {} if limits: limits_dict: ConfigSection = {} for lk in ("max_walk_commits", "max_ancestors", "max_graph_commits", "shard_prefix_length"): lv = limits.get(lk) if lv is not None: limits_dict[lk] = str(lv) if limits_dict: result["limits"] = limits_dict return result def config_path_for_editor(repo_root: pathlib.Path | None = None) -> pathlib.Path: """Return the config path for the ``config edit`` command.""" return _config_path(repo_root) # --------------------------------------------------------------------------- # Remote helpers # --------------------------------------------------------------------------- def get_remote(name: str, repo_root: pathlib.Path | None = None) -> str | None: """Return the URL for remote *name*, or ``None`` when not configured. Args: name: Remote name (e.g. ``"origin"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: URL string, or ``None``. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return None entry = remotes.get(name) if entry is None: return None url = entry.get("url", "") return url.strip() if url.strip() else None def get_remote_pack_origin( name: str, repo_root: pathlib.Path | None = None, ) -> str | None: """Return the ``pack_origin`` override for remote *name*, or ``None``. When set, pack uploads (``POST /push/object-pack``) are directed to this origin instead of the primary remote URL. The path component (owner/slug) is preserved from the primary URL so the same MWP path structure applies. Args: name: Remote name (e.g. ``"local"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: ``pack_origin`` string, or ``None`` when not configured. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return None entry = remotes.get(name) if entry is None: return None origin = entry.get("pack_origin", "") return origin.strip() if origin.strip() else None def set_remote( name: str, url: str, repo_root: pathlib.Path | None = None, ) -> None: """Write ``[remotes.] url`` to ``.muse/config.toml``. Preserves all other sections. Creates the file if absent. Args: name: Remote name (e.g. ``"origin"``). url: Remote URL. repo_root: Repository root. Defaults to ``Path.cwd()``. """ cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) existing_remotes = config.get("remotes") remotes: RemotesMap = {} if existing_remotes: remotes.update(existing_remotes) existing_entry = remotes.get(name) entry: RemoteEntry = {} if existing_entry is not None: if "url" in existing_entry: entry["url"] = existing_entry["url"] if "branch" in existing_entry: entry["branch"] = existing_entry["branch"] entry["url"] = url remotes[name] = entry config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Remote %r set to %s", name, url) def remove_remote( name: str, repo_root: pathlib.Path | None = None, ) -> None: """Remove a named remote and its tracking refs. Args: name: Remote name to remove. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: KeyError: If *name* is not a configured remote. """ cp = _config_path(repo_root) config = _load_config(cp) remotes = config.get("remotes") if remotes is None or name not in remotes: raise KeyError(name) del remotes[name] config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Remote %r removed from config", name) root = (repo_root or pathlib.Path.cwd()).resolve() refs_dir = root / _MUSE_DIR / "remotes" / name if refs_dir.is_symlink(): # Refuse to rmtree a symlink — following a symlink placed by an # attacker could delete files outside the repository tree. logger.warning("⚠️ Skipping rmtree: remotes dir %s is a symlink", refs_dir) elif refs_dir.is_dir(): shutil.rmtree(refs_dir) logger.debug("✅ Removed tracking refs dir %s", refs_dir) def rename_remote( old_name: str, new_name: str, repo_root: pathlib.Path | None = None, ) -> None: """Rename a remote and move its tracking refs. Args: old_name: Current remote name. new_name: Desired new remote name. repo_root: Repository root. Defaults to ``Path.cwd()``. Raises: KeyError: If *old_name* is not a configured remote. ValueError: If *new_name* is already configured. """ cp = _config_path(repo_root) config = _load_config(cp) remotes = config.get("remotes") if remotes is None or old_name not in remotes: raise KeyError(old_name) if new_name in remotes: raise ValueError(new_name) remotes[new_name] = remotes.pop(old_name) config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Remote %r renamed to %r", old_name, new_name) root = (repo_root or pathlib.Path.cwd()).resolve() old_refs_dir = root / _MUSE_DIR / "remotes" / old_name new_refs_dir = root / _MUSE_DIR / "remotes" / new_name if old_refs_dir.is_dir(): old_refs_dir.rename(new_refs_dir) logger.debug("✅ Moved tracking refs dir %s → %s", old_refs_dir, new_refs_dir) def list_remotes(repo_root: pathlib.Path | None = None) -> list[RemoteConfig]: """Return all configured remotes sorted alphabetically by name. Args: repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: List of ``{"name": str, "url": str}`` dicts. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return [] result: list[RemoteConfig] = [] for remote_name in sorted(remotes): entry = remotes[remote_name] url = entry.get("url", "") if not url.strip(): continue rc = RemoteConfig(name=remote_name, url=url.strip()) pack_origin = entry.get("pack_origin", "") if pack_origin.strip(): rc["pack_origin"] = pack_origin.strip() result.append(rc) return result # --------------------------------------------------------------------------- # Remote tracking-head helpers # --------------------------------------------------------------------------- def _remote_head_path( remote_name: str, branch: str, repo_root: pathlib.Path | None = None, ) -> pathlib.Path: """Return the path to the remote tracking pointer file.""" root = (repo_root or pathlib.Path.cwd()).resolve() return root / _MUSE_DIR / "remotes" / remote_name / branch def get_remote_head( remote_name: str, branch: str, repo_root: pathlib.Path | None = None, ) -> str | None: """Return the last-known remote commit ID for *remote_name*/*branch*. Returns ``None`` when the tracking pointer does not exist. Args: remote_name: Remote name (e.g. ``"origin"``). branch: Branch name (e.g. ``"main"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: Commit ID string, or ``None``. """ pointer = _remote_head_path(remote_name, branch, repo_root) if not pointer.is_file(): return None raw = pointer.read_text(encoding="utf-8").strip() return raw if raw else None def set_remote_head( remote_name: str, branch: str, commit_id: str, repo_root: pathlib.Path | None = None, ) -> None: """Write the remote tracking pointer for *remote_name*/*branch*. Args: remote_name: Remote name (e.g. ``"origin"``). branch: Branch name. commit_id: Commit ID to record as the known remote HEAD. repo_root: Repository root. Defaults to ``Path.cwd()``. """ pointer = _remote_head_path(remote_name, branch, repo_root) write_text_atomic(pointer, commit_id) logger.debug("✅ Remote head %s/%s → %s", remote_name, branch, commit_id[:8]) def delete_remote_head( remote_name: str, branch: str, repo_root: pathlib.Path | None = None, ) -> bool: """Remove the local remote-tracking pointer for *remote_name*/*branch*. Used after ``muse push --delete`` deletes the branch on the server, or when pruning stale tracking refs with ``muse branch -dr``. Args: remote_name: Remote name (e.g. ``"origin"``). branch: Branch name (e.g. ``"feat/my-thing"``). repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: ``True`` if the pointer file existed and was removed, ``False`` if it was already absent (idempotent). """ pointer = _remote_head_path(remote_name, branch, repo_root) if not pointer.is_file(): return False pointer.unlink() # Remove now-empty parent directories (mirrors _cleanup_empty_dirs in branch.py). remotes_dir = pointer.parent while remotes_dir.name != remote_name: try: remotes_dir.rmdir() except OSError: break remotes_dir = remotes_dir.parent logger.debug("🗑 Remote tracking ref %s/%s removed", remote_name, branch) return True # --------------------------------------------------------------------------- # Upstream tracking helpers # --------------------------------------------------------------------------- def set_upstream( branch: str, remote_name: str, repo_root: pathlib.Path | None = None, ) -> None: """Record *remote_name* as the upstream remote for *branch*. Args: branch: Local (and remote) branch name. remote_name: Remote name. repo_root: Repository root. Defaults to ``Path.cwd()``. """ cp = _config_path(repo_root) cp.parent.mkdir(parents=True, exist_ok=True) config = _load_config(cp) existing_remotes = config.get("remotes") remotes: RemotesMap = {} if existing_remotes: remotes.update(existing_remotes) existing_entry = remotes.get(remote_name) entry: RemoteEntry = {} if existing_entry is not None: if "url" in existing_entry: entry["url"] = existing_entry["url"] if "branch" in existing_entry: entry["branch"] = existing_entry["branch"] entry["branch"] = branch remotes[remote_name] = entry config["remotes"] = remotes write_text_atomic(cp, _dump_toml(config)) logger.info("✅ Upstream for branch %r set to %s/%r", branch, remote_name, branch) def get_upstream( branch: str, repo_root: pathlib.Path | None = None, ) -> str | None: """Return the configured upstream remote name for *branch*, or ``None``. Args: branch: Local branch name. repo_root: Repository root. Defaults to ``Path.cwd()``. Returns: Remote name string, or ``None``. """ config = _load_config(_config_path(repo_root)) remotes = config.get("remotes") if remotes is None: return None for rname, entry in remotes.items(): tracked = entry.get("branch", "") if tracked.strip() == branch: return rname return None # --------------------------------------------------------------------------- # Global user config — ~/.muse/config.toml (safe_dirs) # --------------------------------------------------------------------------- _GLOBAL_MUSE_DIR = pathlib.Path.home() / ".muse" _GLOBAL_CONFIG_FILE = _GLOBAL_MUSE_DIR / "config.toml" def _load_global_config() -> dict[str, list[str]]: """Load ``~/.muse/config.toml`` and return the ``[security]`` section. Returns a dict with key ``safe_dirs`` mapping to a list of path strings. Returns ``{"safe_dirs": []}`` when the file is absent or unparseable. """ if not _GLOBAL_CONFIG_FILE.is_file(): return {"safe_dirs": []} try: with _GLOBAL_CONFIG_FILE.open("rb") as fh: raw: dict[str, object] = tomllib.load(fh) except Exception as exc: # noqa: BLE001 logger.warning("⚠️ Failed to parse %s: %s", _GLOBAL_CONFIG_FILE, exc) return {"safe_dirs": []} security_raw = raw.get("security") if not isinstance(security_raw, dict): return {"safe_dirs": []} dirs_raw = security_raw.get("safe_dirs") if not isinstance(dirs_raw, list): return {"safe_dirs": []} safe: list[str] = [d for d in dirs_raw if isinstance(d, str) and d.strip()] return {"safe_dirs": safe} def _save_global_config(safe_dirs: list[str]) -> None: """Write ``[security] safe_dirs`` to ``~/.muse/config.toml``. Preserves any other sections that may exist in the file. """ import os _GLOBAL_MUSE_DIR.mkdir(parents=True, exist_ok=True) # Read existing raw content to preserve other sections. existing_lines: list[str] = [] if _GLOBAL_CONFIG_FILE.is_file(): try: existing_lines = _GLOBAL_CONFIG_FILE.read_text("utf-8").splitlines() except Exception: # noqa: BLE001 existing_lines = [] # Strip any existing [security] section from the file. filtered: list[str] = [] in_security = False for line in existing_lines: stripped = line.strip() if stripped == "[security]": in_security = True continue if in_security and stripped.startswith("["): in_security = False if not in_security: filtered.append(line) # Remove trailing blank lines before appending the new section. while filtered and not filtered[-1].strip(): filtered.pop() # Append the new [security] section. filtered.append("") filtered.append("[security]") if safe_dirs: items = ", ".join(f'"{_escape(d)}"' for d in safe_dirs) filtered.append(f"safe_dirs = [{items}]") else: filtered.append("safe_dirs = []") filtered.append("") content = "\n".join(filtered) tmp = _GLOBAL_CONFIG_FILE.with_suffix(".toml.tmp") tmp.write_text(content, encoding="utf-8") os.replace(tmp, _GLOBAL_CONFIG_FILE) def get_global_safe_dirs() -> list[str]: """Return the ``safe_dirs`` list from ``~/.muse/config.toml``. Returns an empty list when not configured. """ return _load_global_config().get("safe_dirs", []) def add_global_safe_dir(path: str) -> None: """Add *path* to the ``safe_dirs`` list in ``~/.muse/config.toml``. Normalises the path (``os.path.abspath``) before storing. Idempotent — adding the same path twice has no effect. Args: path: Absolute or relative path to trust. """ import os abs_path = os.path.abspath(path) current = get_global_safe_dirs() if abs_path not in current: current.append(abs_path) _save_global_config(current) logger.info("✅ Trusted path added: %s", abs_path) def remove_global_safe_dir(path: str) -> None: """Remove *path* from the ``safe_dirs`` list in ``~/.muse/config.toml``. Normalises the path before matching. No-op when the path is not present. Args: path: Absolute or relative path to remove from the trust list. """ import os abs_path = os.path.abspath(path) current = get_global_safe_dirs() updated = [d for d in current if d != abs_path] _save_global_config(updated) logger.info("✅ Trusted path removed: %s", abs_path)