secrets-engine/src/secrets_engine/exec_owner.py

"""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)