Bind credential exec to exact owner inputs and approval digest

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

src/secrets_engine/exec_owner.py

@@ -0,0 +1,124 @@
+"""Catalog-bound child delivery. Configuration constrains a grant; it grants none.
+
+The engine and configured owner UID remain trusted. These checks do not fence a
+malicious process with that same UID or replace runtime/profile verification.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import stat
+from typing import Any
+
+from secrets_engine.errors import CatalogError, DeliveryError
+
+
+def _text(value: object) -> bool:
+    return isinstance(value, str) and bool(value.strip()) and "\0" not in value
+
+
+def _absolute(value: object) -> bool:
+    return _text(value) and Path(value).is_absolute() and str(Path(value)) == value and ".." not in Path(value).parts
+
+
+def validate_exec_owner(value: object) -> dict[str, Any]:
+    if not isinstance(value, dict) or not _text(value.get("owner")):
+        raise CatalogError("delivery_config.exec_owner requires a named owner")
+    if value.get("status") == "pending":
+        if set(value) != {"status", "owner", "reason"} or not _text(value.get("reason")):
+            raise CatalogError("pending exec_owner requires only status, owner and reason")
+        return value
+    required = {"status", "owner", "command", "cwd", "environment", "files"}
+    if value.get("status") != "configured" or set(value) != required:
+        raise CatalogError("configured exec_owner requires exact command/cwd/environment/files")
+    command = value["command"]
+    if not isinstance(command, list) or not command or not all(_text(x) for x in command) or not _absolute(command[0]):
+        raise CatalogError("exec_owner command must be an exact argv with an absolute executable")
+    if not _absolute(value["cwd"]):
+        raise CatalogError("exec_owner cwd must be an absolute canonical directory")
+    env = value["environment"]
+    if not isinstance(env, dict) or not all(
+        isinstance(k, str) and re.fullmatch(r"[A-Z_][A-Z0-9_]*", k)
+        and isinstance(v, str) and "\0" not in v for k, v in env.items()
+    ):
+        raise CatalogError("exec_owner environment must be an explicit string mapping")
+    # The fixed environment is non-secret source configuration, not a way to
+    # carry the invoking engine's credentials or loader injection into the child.
+    if any(k.startswith(("LD_", "DYLD_", "PYTHON", "BAO_", "VAULT_", "SECRETS_ENGINE_", "ANTHROPIC_", "CLAUDE_")) for k in env):
+        raise CatalogError("exec_owner environment contains a forbidden credential/loader variable")
+    files = value["files"]
+    if not isinstance(files, dict) or command[0] not in files:
+        raise CatalogError("exec_owner files must pin the executable")
+    for path, spec in files.items():
+        if not _absolute(path) or not isinstance(spec, dict) or set(spec) != {"sha256", "private"}:
+            raise CatalogError("exec_owner files require canonical paths and sha256/private")
+        if not isinstance(spec["sha256"], str) or not re.fullmatch(r"[0-9a-f]{64}", spec["sha256"]) or type(spec["private"]) is not bool:
+            raise CatalogError("exec_owner file pin must have an exact SHA-256 and boolean private")
+    for arg in command[1:]:
+        if arg.startswith("/") and arg not in files:
+            raise CatalogError("exec_owner absolute file arguments must have file pins")
+    return value
+
+
+def owner_binding(entry) -> dict[str, Any] | None:
+    config = entry.delivery_config
+    if "exec_owner" not in config:
+        return None
+    return validate_exec_owner(config["exec_owner"])
+
+
+def owner_digest(entry) -> str | None:
+    binding = owner_binding(entry)
+    if binding is None:
+        return None
+    return hashlib.sha256(json.dumps(binding, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
+
+
+def _check_path(path: Path, *, directory: bool = False, private: bool = False) -> None:
+    """No symlink or untrusted writable ancestor; root-owned sticky /tmp is safe."""
+    for candidate in [*reversed(path.parents), path]:
+        info = candidate.lstat()
+        leaf = candidate == path
+        is_dir = not leaf or directory
+        if not (stat.S_ISDIR(info.st_mode) if is_dir else stat.S_ISREG(info.st_mode)):
+            raise DeliveryError("exec owner path is not a regular file/directory")
+        if info.st_uid not in {0, os.getuid()}:
+            raise DeliveryError("exec owner path has an untrusted owner")
+        sticky_root = not leaf and info.st_uid == 0 and bool(info.st_mode & stat.S_ISVTX)
+        if info.st_mode & 0o022 and not sticky_root:
+            raise DeliveryError("exec owner path is writable by another identity")
+        if leaf and private and info.st_mode & 0o077:
+            raise DeliveryError("exec owner private path is accessible by another identity")
+
+
+def validate_delivery_target(entry, field: str, command: list[str], mode: str) -> str | None:
+    binding = owner_binding(entry)
+    if binding is None:
+        return None
+    if binding["status"] != "configured":
+        raise DeliveryError("exec owner binding is pending; no delivery is admitted")
+    if mode not in {"auto", "exec-env"} or "exec-env" not in entry.delivery_modes:
+        raise DeliveryError("exec owner requires exec-env delivery")
+    if field not in entry.fields or field.upper() in binding["environment"]:
+        raise DeliveryError("exec owner field is undeclared or conflicts with fixed environment")
+    if command != binding["command"]:
+        raise DeliveryError("command differs from the catalog-bound exec owner")
+    try:
+        _check_path(Path(binding["cwd"]), directory=True, private=True)
+        for name, pin in binding["files"].items():
+            path = Path(name)
+            _check_path(path, private=pin["private"])
+            fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+            with os.fdopen(fd, "rb") as source:
+                if not stat.S_ISREG(os.fstat(source.fileno()).st_mode):
+                    raise DeliveryError("exec owner file changed type")
+                if hashlib.file_digest(source, "sha256").hexdigest() != pin["sha256"]:
+                    raise DeliveryError("exec owner file digest differs from catalog")
+        if not os.access(command[0], os.X_OK):
+            raise DeliveryError("exec owner executable is not executable")
+    except OSError as exc:
+        raise DeliveryError("exec owner path is unavailable or unsafe") from exc
+    return owner_digest(entry)