diff --git a/tartarus/jail.py b/tartarus/jail.py index 4f5f47a..d03dc5f 100644 --- a/tartarus/jail.py +++ b/tartarus/jail.py @@ -21,6 +21,10 @@ import uuid from collections.abc import Callable from dataclasses import dataclass, field from queue import Empty, Queue +from typing import Literal + +from pydantic import ConfigDict, ValidationError, field_validator +from pydantic.dataclasses import dataclass as strict_dataclass from tartarus.manifest import Grant from tartarus.network_proxy import FilteringProxy @@ -28,12 +32,14 @@ from tartarus.network_proxy import FilteringProxy DEFAULT_JAIL_TIMEOUT_SECONDS = 30 TIMEOUT_EXIT_CODE = 124 # matches coreutils `timeout` +_STRICT = ConfigDict(frozen=True, extra="forbid", strict=True) + class JailError(Exception): """Raised when a jail cannot be built or run as requested.""" -@dataclass(frozen=True) +@strict_dataclass(config=_STRICT) class ExecResult: code: int stdout: str @@ -58,7 +64,7 @@ class BackgroundHandle: proxy: FilteringProxy | None = None -@dataclass(frozen=True) +@strict_dataclass(config=_STRICT) class JailSpec: work_tree: str shell_path: str @@ -71,9 +77,22 @@ class JailSpec: # declared closure (PLAN.md ยง13). bind_paths: list[str] = field(default_factory=list) allowed_hosts: list[str] = field(default_factory=list) - network: str = "none" # "none" | "proxy" + network: Literal["none", "proxy"] = "none" unrestricted: bool = False + @field_validator("writable") + @classmethod + def _validate_writable(cls, paths: list[str]) -> list[str]: + # Belt-and-suspenders with Grant._validate_writable: JailBuilder.build + # always feeds an already-validated Grant, but a JailSpec built directly + # must still refuse anything that escapes the work tree. + for path in paths: + if path.startswith("/"): + raise ValueError(f"writable path '{path}' must be relative") + if ".." in path.split("/"): + raise ValueError(f"writable path '{path}' escapes the work tree") + return paths + class JailBuilder: def __init__( @@ -94,17 +113,20 @@ class JailBuilder: self._shell_closure = list(shell_closure or []) def build(self, grant: Grant) -> JailSpec: - return JailSpec( - work_tree=self._work_tree, - shell_path=self._shell_path, - base_env=self._base_env, - writable=_validated_writable_paths(grant.writable), - extra_path=list(grant.package_bins), - bind_paths=_dedup(self._shell_closure + list(grant.closure_paths)), - allowed_hosts=list(grant.allowed_hosts), - network="proxy" if grant.allowed_hosts else "none", - unrestricted=grant.unrestricted, - ) + try: + return JailSpec( + work_tree=self._work_tree, + shell_path=self._shell_path, + base_env=self._base_env, + writable=list(grant.writable), + extra_path=list(grant.package_bins), + bind_paths=_dedup(self._shell_closure + list(grant.closure_paths)), + allowed_hosts=list(grant.allowed_hosts), + network="proxy" if grant.allowed_hosts else "none", + unrestricted=grant.unrestricted, + ) + except ValidationError as error: + raise JailError(str(error)) from error def exec( self, @@ -457,15 +479,6 @@ def _dedup(paths: list[str]) -> list[str]: return unique -def _validated_writable_paths(paths: list[str]) -> list[str]: - for path in paths: - if path.startswith("/"): - raise JailError(f"writable path '{path}' must be relative") - if ".." in path.split("/"): - raise JailError(f"writable path '{path}' escapes the work tree") - return list(paths) - - def _host_writable_path(work_tree: str, relative_path: str) -> str: host_path = os.path.abspath(os.path.join(work_tree, relative_path)) if host_path != work_tree and not host_path.startswith(work_tree + os.sep): diff --git a/tartarus/policy.py b/tartarus/policy.py index 0699083..5afdf9f 100644 --- a/tartarus/policy.py +++ b/tartarus/policy.py @@ -13,19 +13,24 @@ defaulting to No. In headless mode there is no human, so every ask-* policy deni import sys from collections.abc import Callable -from dataclasses import dataclass +from typing import Literal + +from pydantic import ConfigDict +from pydantic.dataclasses import dataclass from tartarus.manifest import Capability, Grant # Decides one prompt: (capability, arguments, interpolated command) -> approved? PromptFn = Callable[[Capability, dict, str], bool] +_STRICT = ConfigDict(frozen=True, extra="forbid", strict=True) + -@dataclass(frozen=True) +@dataclass(config=_STRICT) class Decision: allowed: bool reason: str - approver: str # "auto" | "deny" | "session" | "human" | "headless" + approver: Literal["auto", "deny", "session", "human", "headless", "broker"] class PolicyEngine: