"""Trust-boundary validation primitives for Muse. Every function in this module operates on untrusted input and either returns a safe value or raises ValueError / TypeError with a descriptive message. No other muse module is imported here — this module must stay at the bottom of the dependency graph so it can be safely imported by every layer. """ from __future__ import annotations import math import pathlib import re # --------------------------------------------------------------------------- # Size ceilings # --------------------------------------------------------------------------- MAX_FILE_BYTES: int = 256 * 1024 * 1024 # 256 MB — per-file read cap MAX_RESPONSE_BYTES: int = 64 * 1024 * 1024 # 64 MB — small-endpoint cap (refs, push, etc.) MAX_FETCH_BYTES: int = 2 * 1024 * 1024 * 1024 # 2 GiB — fetch/clone streaming cap MAX_SYSEX_BYTES: int = 65_536 # 64 KiB — MIDI sysex data truncation point MAX_AST_BYTES: int = 2 * 1024 * 1024 # 2 MB — per-file Python AST parse cap # Pack-format limits — applied in apply_pack and write_object to prevent # object-store poisoning and pack-bomb DoS attacks. # # MAX_OBJECT_WRITE_BYTES mirrors MAX_FILE_BYTES so the write and read paths # enforce the same ceiling — an object that can be written can always be read. # # MAX_PACK_OBJECTS caps the number of blobs, snapshots, and commits that a # single pack may inject in one apply_pack call. A trillion-agent world will # push many small packs, not one pack with a billion objects. MAX_OBJECT_WRITE_BYTES: int = MAX_FILE_BYTES # 256 MB — per-object write cap MAX_PACK_OBJECTS: int = 100_000 # maximum blobs per pack invocation # --------------------------------------------------------------------------- # Regex patterns # --------------------------------------------------------------------------- _HEX64_RE = re.compile(r"^[0-9a-f]{64}$") # Branch/ref names: follow Git conventions (git-check-ref-format rules). # Forbidden: # - All C0 control chars 0x00-0x1F (incl. null, tab, CR, LF) and DEL 0x7F # - Space 0x20 (causes shell interpolation issues and log-parsing ambiguity) # - Backslash (path separator on Windows; always banned for portability) # - Git-banned punctuation: ~ ^ : ? * [ (ancestry, refspec, glob syntax) # - Leading or trailing dot (hidden-file confusion, git pack index convention) # - Consecutive dots (..) — path traversal # - Leading or trailing slash — absolute-path ambiguity # - Consecutive slashes (//) — normalised differently on some filesystems # - Single-dot path component (/./): resolves to parent component on disk; # two branch names would silently share the same ref file # - Component ending in .lock: Muse uses .muse-tmp- prefix for temp files, # but .lock files are a widespread VCS convention; allowing them risks # ambiguity in any tooling that scans the ref directory # - @{ sequence: git reflog syntax; causes parser confusion in pipelines # - Bare "@": git ref notation; rejected for the same reason # Allowed: forward slash (enables feature/my-branch style namespacing). # Max 255 chars. _BRANCH_FORBIDDEN_RE = re.compile( r"[\\\x00-\x20\x7f~^:?*\[]" # backslash, all C0 + space, DEL, git punctuation r"|^\." # leading dot r"|\.$" # trailing dot r"|\.{2,}" # consecutive dots (.., ...) r"|//" # consecutive slashes r"|^/" # leading slash r"|/$" # trailing slash r"|/\.(?:/|$)" # single-dot path component (/./ or /.) r"|\.lock(?:/|$)" # .lock component suffix (matches main.lock, feat/x.lock) r"|@\{" # git reflog @{ sequence r"|^@$" # bare @ (git HEAD shorthand) ) # Valid domain plugin name: lowercase letters, digits, hyphens, underscores; # must start with a lowercase letter. _DOMAIN_RE = re.compile(r"^[a-z][a-z0-9_-]{0,62}$") # Control characters to strip from terminal output. # Removes all C0 (0x00-0x1F) except \t (0x09) and \n (0x0A), # plus DEL (0x7F), and C1 (0x80-0x9F). _CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f\x80-\x9f]") # Glob metacharacters that must not appear in prefix arguments. _GLOB_META_RE = re.compile(r"[*?\[\]{}]") # --------------------------------------------------------------------------- # ID validation # --------------------------------------------------------------------------- def validate_object_id(s: str) -> str: """Return *s* unchanged if it is a valid SHA-256 hex string (64 lowercase hex chars). Raises ValueError for anything else, preventing path-traversal attacks built from crafted object IDs. """ if not isinstance(s, str): raise TypeError(f"object_id must be str, got {type(s).__name__}") if not _HEX64_RE.match(s): raise ValueError( f"Invalid object ID {s!r}: expected exactly 64 lowercase hex characters." ) return s def validate_ref_id(s: str) -> str: """Return *s* unchanged if it is a valid commit/snapshot/tag ID. Uses the same 64-char hex rule as object IDs; the two functions exist as separate names so call-sites are self-documenting. """ if not isinstance(s, str): raise TypeError(f"ref_id must be str, got {type(s).__name__}") if not _HEX64_RE.match(s): raise ValueError( f"Invalid ref ID {s!r}: expected exactly 64 lowercase hex characters." ) return s # --------------------------------------------------------------------------- # Branch / repo-id validation # --------------------------------------------------------------------------- def validate_branch_name(name: str) -> str: """Return *name* unchanged if it is a safe branch name. Follows ``git check-ref-format`` conventions so Muse branch names round-trip safely through any Git-backed transport: Allowed ------- - ASCII letters, digits, ``-``, ``_``, ``.`` (within a path component). - Forward slash ``/`` as a namespace separator (``feat/my-branch``). Rejected -------- - All C0 control characters (0x00–0x1F), space (0x20), and DEL (0x7F). Prevents terminal-injection attacks via ``for-each-ref --format text``. - Backslash — path separator on Windows; banned for portability. - Git-banned punctuation: ``~``, ``^``, ``:``, ``?``, ``*``, ``[``. - Leading or trailing dot — hidden-file confusion. - Consecutive dots (``..``) — path-traversal vector. - Leading or trailing slash — absolute-path ambiguity. - Consecutive slashes (``//``) — filesystem normalisation differences. - Single-dot path component (``/./``): ``feat/./sub`` resolves to the same inode as ``feat/sub`` on every POSIX and Windows filesystem; two branch names would silently share one ref file. - Any component ending in ``.lock`` (``main.lock``, ``feat/x.lock``): the VCS convention reserves ``.lock`` for exclusive-lock files; allowing them creates ambiguity for any tooling that scans the ref directory. - The sequence ``@{`` — git reflog notation; causes parser confusion. - The bare string ``@`` — git HEAD shorthand. - Empty string. - Names longer than 255 characters. """ if not isinstance(name, str): raise TypeError(f"branch name must be str, got {type(name).__name__}") if not name: raise ValueError("Branch name must not be empty.") if len(name) > 255: raise ValueError( f"Branch name too long ({len(name)} chars); maximum is 255." ) if _BRANCH_FORBIDDEN_RE.search(name): raise ValueError( f"Branch name {name!r} contains forbidden characters " "(path separators, null bytes, or consecutive dots)." ) return name def validate_repo_id(repo_id: str) -> str: """Return *repo_id* if it contains no path-traversal components. repo_id values are UUIDs in normal operation. We enforce that they contain no path separators or dot-sequences rather than enforcing UUID format strictly, to allow future flexibility. """ if not isinstance(repo_id, str): raise TypeError(f"repo_id must be str, got {type(repo_id).__name__}") if not repo_id: raise ValueError("repo_id must not be empty.") if len(repo_id) > 255: raise ValueError(f"repo_id too long ({len(repo_id)} chars).") if _BRANCH_FORBIDDEN_RE.search(repo_id): raise ValueError( f"repo_id {repo_id!r} contains forbidden characters." ) return repo_id def validate_domain_name(domain: str) -> str: """Return *domain* if it is a valid plugin domain name.""" if not _DOMAIN_RE.match(domain): raise ValueError( f"Domain name {domain!r} is invalid. " "Must start with a lowercase letter and contain only " "lowercase letters, digits, hyphens, or underscores (max 63 chars)." ) return domain # --------------------------------------------------------------------------- # Path containment # --------------------------------------------------------------------------- # Maximum length for a workspace-relative path argument (one per component; # Linux PATH_MAX is 4096 — we use 4096 as the sane ceiling). _MAX_WORKSPACE_PATH_LEN: int = 4096 # Detect path components that escape the root: ".." alone or leading "/". # We split on both POSIX "/" and Windows "\\" for portability. _DOTDOT_RE = re.compile(r"(?:^|[/\\])\.\.(?:[/\\]|$)") def validate_workspace_path(path: str) -> str: """Validate a user-supplied workspace-relative path. Workspace paths are relative labels used for attribute lookup, ignore matching, and similar pattern-only operations. They are *never* used directly to open files — but we still gate them here so that: - Null bytes, CR, and C0 control characters that could corrupt terminal output or downstream shell pipelines are rejected. - Absolute paths (starting with ``/`` or a drive letter) are rejected — they make no sense as workspace-relative labels and almost certainly indicate a confused caller or an injection attempt. - Traversal sequences (``..`` as a path component) are rejected as defense-in-depth, even though no file open occurs on these paths. - Paths longer than 4096 characters are rejected to prevent O(N) fnmatch calls from becoming a denial-of-service vector. Returns *path* unchanged when valid. Raises :exc:`ValueError` otherwise. """ if not isinstance(path, str): raise TypeError(f"workspace path must be str, got {type(path).__name__}") if not path: raise ValueError("Workspace path must not be empty.") if len(path) > _MAX_WORKSPACE_PATH_LEN: raise ValueError( f"Workspace path too long ({len(path)} chars); " f"maximum is {_MAX_WORKSPACE_PATH_LEN}." ) if "\x00" in path: raise ValueError("Workspace path contains a null byte.") # Reject all C0 control characters except tab (path labels should be # printable; CR/LF would corrupt text output; ESC enables ANSI injection). for ch in path: cp = ord(ch) if cp < 0x20 and cp not in (0x09,): # allow tab only raise ValueError( f"Workspace path contains a control character (U+{cp:04X})." ) # Absolute paths are not workspace-relative. if path.startswith("/") or path.startswith("\\") or ( len(path) >= 2 and path[1] == ":" ): raise ValueError( f"Workspace path must be relative, not absolute: {path!r}" ) # Traversal sequences — even though no filesystem open occurs, reject them # as defense-in-depth and to keep output predictable for downstream tools. if _DOTDOT_RE.search(path) or path == ".." or path.startswith("../") or path.startswith("..\\"): raise ValueError( f"Workspace path contains a traversal sequence (..): {path!r}" ) return path def validate_path_prefix(prefix: str) -> str: """Validate a ``--path-prefix`` filter string. Prefix filters are used as ``str.startswith()`` predicates on manifest keys — they are not filesystem paths, but they must not contain: - Null bytes (protocol confusion / injection). - Control characters other than tab. - Glob metacharacters (``*``, ``?``, ``[``, ``]``, ``{``, ``}``) which would be meaningless here and suggest a confused caller. Returns *prefix* unchanged when valid. Raises :exc:`ValueError` otherwise. """ if not isinstance(prefix, str): raise TypeError(f"path prefix must be str, got {type(prefix).__name__}") if "\x00" in prefix: raise ValueError("Path prefix contains a null byte.") for ch in prefix: cp = ord(ch) if cp < 0x20 and cp not in (0x09,): raise ValueError( f"Path prefix contains a control character (U+{cp:04X})." ) if _GLOB_META_RE.search(prefix): raise ValueError( f"Path prefix contains glob metacharacters: {prefix!r}" ) return prefix def assert_not_symlink(path: pathlib.Path, label: str = "") -> None: """Raise ValueError if *path* is a symbolic link. Used as a defence-in-depth guard on critical repository directories (``.muse/``, ``.muse/objects/``, shard directories, and metadata subdirectories). An attacker who replaces one of these with a symlink pointing to an attacker-controlled location could redirect all subsequent writes outside the repository root. Args: path: The filesystem path to check. label: Human-readable name for *path* used in the error message (e.g. ``".muse/"`` or ``"objects store"``). Raises: ValueError: If *path* is a symbolic link. """ if path.is_symlink(): name = label or str(path) raise ValueError( f"Security violation: {name} is a symbolic link. " "Muse refuses to use a symlinked repository directory to prevent " "writes from being redirected outside the repository root." ) def assert_write_inside_repo(repo_root: pathlib.Path, path: pathlib.Path) -> None: """Raise ValueError if *path* resolves outside *repo_root*. Resolves both *repo_root* and *path* (following symlinks) and verifies the resolved *path* is a descendant of the resolved *repo_root*. This catches TOCTOU symlink-swap attacks: even if a symlink is placed inside ``.muse/`` after the startup check, the resolved destination will escape the repo root and be rejected here. Args: repo_root: The repository root (parent of ``.muse/``). path: The destination path to validate. Raises: ValueError: If *path* resolves to a location outside *repo_root*. """ real_root = repo_root.resolve() real_path = path.resolve() if not real_path.is_relative_to(real_root): raise ValueError( f"Security violation: write destination {path!r} resolves to " f"{real_path} which is outside the repository root {real_root}. " "Possible symlink attack." ) def contain_path(base: pathlib.Path, rel: str) -> pathlib.Path: """Join *base* / *rel*, resolve, and assert the result stays inside *base*. This is the central defence against zip-slip and path-traversal attacks in manifest keys, rel_path arguments, and any other user-controlled path component that is joined onto a trusted base directory. Raises ValueError if the resolved path escapes *base*. """ if not isinstance(rel, str): raise TypeError(f"rel must be str, got {type(rel).__name__}") if not rel: raise ValueError("Relative path component must not be empty.") # Absolute paths on POSIX cause pathlib to discard the base entirely. joined = base / rel resolved = joined.resolve() base_resolved = base.resolve() if not resolved.is_relative_to(base_resolved): raise ValueError( f"Path traversal detected: {rel!r} escapes the base directory " f"{base_resolved}." ) return resolved # --------------------------------------------------------------------------- # Glob safety # --------------------------------------------------------------------------- def sanitize_glob_prefix(prefix: str) -> str: """Return *prefix* with glob metacharacters removed. Used in _find_commit_by_prefix to prevent glob injection turning a targeted lookup into an arbitrary filesystem scan. """ return _GLOB_META_RE.sub("", prefix) # --------------------------------------------------------------------------- # Display sanitization # --------------------------------------------------------------------------- def sanitize_display(s: str) -> str: """Strip terminal control characters from *s* before echoing to the user. Preserves newline (\\n) and tab (\\t) as these are legitimate in multi-line commit messages. Removes all other C0/C1 control characters including ESC (0x1B), BEL (0x07), and CSI (0x9B) — the entry points for ANSI/OSC terminal escape injection. Storage is never mutated; sanitization happens only at display time. """ return _CONTROL_CHARS_RE.sub("", s) # Strips ALL C0 (0x00-0x1F), DEL (0x7F), and C1 (0x80-0x9F) from single-line # identity strings. Unlike _CONTROL_CHARS_RE this also strips \t and \n — # provenance fields are single-line; embedded newlines are never legitimate. _PROVENANCE_STRIP_RE = re.compile(r"[\x00-\x1f\x7f\x80-\x9f]") def sanitize_provenance(s: str) -> str: """Sanitize a provenance string (agent_id, model_id, toolchain_id, prompt_hash). These are single-line identity fields stored in commit records. Unlike :func:`sanitize_display`, this function also strips ``\\t`` and ``\\n`` because provenance fields are never multi-line. Strips all C0 control characters (0x00–0x1F), DEL (0x7F), and C1 control characters (0x80–0x9F). This prevents: - Terminal injection when provenance is rendered in ``muse log`` or ``muse show`` output in future display paths. - Log-line splitting if the agent_id is written to a log file. - JSON structure attacks if the value is interpolated unsafely. - Unicode right-to-left override (U+202E) and similar confusable chars that could cause visual spoofing in terminals and web UIs. The function does **not** truncate — callers are responsible for applying the 256-character limit using ``[:_MAX_PROV]``. Args: s: Raw provenance string from environment variable or CLI flag. Returns: A copy of *s* with all C0/DEL/C1 control characters removed. """ return _PROVENANCE_STRIP_RE.sub("", s) _TOKEN_CTRL_RE = re.compile(r"[\x00-\x1f\x7f\x80-\x9f]") # Maximum sensible token length — covers opaque tokens and API keys, # while rejecting accidental env var pollution. _MAX_TOKEN_LEN: int = 8192 def sanitize_token(raw: str) -> str | None: """Validate and sanitise a token string read from an environment variable or file. Tokens must be single-line printable ASCII/UTF-8 strings. Any value that contains C0/C1 control characters (including CR and LF) would corrupt HTTP headers at the transport layer. Python's ``http.client`` already blocks CRLF injection at the wire, but defense-in-depth means we reject the value before it ever reaches the HTTP stack so we can emit a clear diagnostic rather than a confusing ``ValueError`` from deep inside ``urllib``. Args: raw: The raw token string (from env var, config file, or prompt). Returns: The token stripped of leading/trailing whitespace, or ``None`` if the value is empty, too long, or contains control characters. """ token = raw.strip() if not token: return None if len(token) > _MAX_TOKEN_LEN: return None if _TOKEN_CTRL_RE.search(token): return None return token # --------------------------------------------------------------------------- # Numeric guards # --------------------------------------------------------------------------- def clamp_int(value: int, lo: int, hi: int, name: str = "value") -> int: """Return *value* clamped to [lo, hi], raising ValueError if out of range.""" if not lo <= value <= hi: raise ValueError( f"{name} must be between {lo} and {hi}, got {value}." ) return value def clamp_natural(value: int, max_val: int, name: str = "value") -> int: """Return *value* clamped to [0, max_val], raising ValueError if out of range. Convenience wrapper around clamp_int for non-negative counts and limits. """ return clamp_int(value, 0, max_val, name) def finite_float(value: float, fallback: float, name: str = "value") -> float: """Return *value* if finite, else *fallback* (and log nothing here — caller logs).""" if not math.isfinite(value): return fallback return value def validate_output_path(path_str: str, root: pathlib.Path) -> pathlib.Path: """Resolve an --output path argument; must stay inside *root*. Prevents a malicious agent from directing output to an arbitrary absolute path such as /etc/cron.d/evil. Raises ValueError if *path_str* resolves to a location outside *root*. """ return contain_path(root, path_str)