apodex/frontier-agent-demo
14
1"""Local path authorization — the fail-closed gate for host filesystem access."""2 3from __future__ import annotations4 5import logging6import os7import re8from collections.abc import Callable9from pathlib import Path10 11from frontier_agent.core.execution_context import get_current_execution_scope12 13logger = logging.getLogger(__name__)14 15_SERVICE_CHECKOUT_ROOT = Path(__file__).resolve().parents[2]16 17# Allowed directory prefixes (relative paths for project dirs, absolute for output)18_ALLOWED_RELATIVE_PREFIXES = [19 "plugins/skills/",20 "data/",21]22 23#: The operator-curated skills tree. Its symlinks are trusted (see24#: :func:`_candidate_paths`), so it is named once rather than spelled inline.25_SKILLS_DIR = str(_SERVICE_CHECKOUT_ROOT / "plugins" / "skills")26 27_ALLOWED_ABSOLUTE_PREFIXES = [28 "/tmp/agent-outputs/",29 _SKILLS_DIR,30]31 32# Blocked file names (security). Matched on the BASENAME at word level rather33# than as a substring of the whole path: a substring test also refused every34# path under a directory that merely contains one of these words, plus names35# like ``tokenizer_config.json``, ``secretary_notes.md`` and ``deck.keynote``.36_BLOCKED_NAME_WORDS = frozenset({37 "credential", "credentials", "secret", "secrets",38 "password", "passwords", "token", "tokens",39})40_BLOCKED_SUFFIXES = (".key", ".pem", ".cert")41_WORD_SPLIT = re.compile(r"[^a-z0-9]+")42 43 44def _configured_workspace_root() -> Path | None:45 """Return the explicit workspace root, if one was configured for this task."""46 scope = get_current_execution_scope()47 metadata = scope.metadata if scope else {}48 raw_root = (49 str(metadata.get("coding_workspace_root") or metadata.get("workspace_root") or "").strip()50 or os.getenv("CODING_WORKSPACE_ROOT", "").strip()51 )52 if not raw_root:53 return None54 55 workspace_root = Path(raw_root).expanduser().resolve()56 if not workspace_root.is_dir():57 logger.warning("Ignoring invalid CODING_WORKSPACE_ROOT '%s'", raw_root)58 return None59 return workspace_root60 61 62def _is_isolated_workspace_root(workspace_root: Path) -> bool:63 """Host writes are only allowed for workspaces outside the service checkout."""64 service_root = _SERVICE_CHECKOUT_ROOT.resolve()65 try:66 workspace_root.relative_to(service_root)67 return False68 except ValueError:69 pass70 71 try:72 service_root.relative_to(workspace_root)73 return False74 except ValueError:75 return True76 77 78def _candidate_paths(79 file_path: str, workspace_root: Path | None, *, write_access: bool = False,80) -> list[Path]:81 """Resolve a path against the explicit workspace root before falling back locally."""82 raw_path = Path(file_path)83 candidates: list[Path] = []84 if raw_path.is_absolute():85 resolved = raw_path.resolve()86 if resolved == raw_path:87 return [resolved]88 # A symlink may only WIDEN access from inside the operator-curated89 # plugins/skills/ tree, whose links deliberately point at SKILL.md90 # bodies outside the project — and then only for READS, since nothing91 # in that tree is a write target. Everywhere else (and for every write)92 # the resolved target is the only candidate: the task workspace is93 # model-writable, so also accepting the unresolved path there let the94 # model ``ln -s ~/.ssh`` into the workspace and read the target95 # straight back through this gate. That reaches past bubblewrap too —96 # it jails ``bash``, while the file tools do in-process host IO, so97 # this gate is their only boundary.98 #99 # The resolved form comes FIRST so the blocked-name check in100 # :func:`_authorized_local_path` sees the real target: ordered the other101 # way, a curated ``SKILL.md -> .env.prod`` link would be judged by the102 # link's own harmless name.103 if _is_skill_path(raw_path) and not write_access:104 return [resolved, raw_path]105 return [resolved]106 107 if workspace_root is not None:108 candidates.append((workspace_root / raw_path).resolve())109 candidates.append(raw_path.resolve())110 111 deduped: list[Path] = []112 seen: set[str] = set()113 for candidate in candidates:114 key = str(candidate)115 if key in seen:116 continue117 seen.add(key)118 deduped.append(candidate)119 return deduped120 121 122def _is_skill_path(path: Path) -> bool:123 """True for a path inside the operator-curated ``plugins/skills/`` tree."""124 return _path_within(str(path), _resolve_prefix(_SKILLS_DIR))125 126 127def _blocked_name(name: str) -> str:128 """The blocked pattern *name* trips, else ``""``.129 130 Word-level so ``token`` refuses ``api_token.txt`` but not131 ``tokenizer_config.json``, and suffix-level so ``.key`` refuses132 ``server.key`` but not ``deck.keynote``.133 """134 lower = name.lower()135 if lower == ".env" or lower.startswith(".env."):136 return ".env"137 for suffix in _BLOCKED_SUFFIXES:138 if lower.endswith(suffix):139 return suffix140 for word in _WORD_SPLIT.split(lower):141 if word in _BLOCKED_NAME_WORDS:142 return word143 return ""144 145 146def _authorized_local_path(file_path: str, *, write_access: bool = False) -> tuple[Path | None, str]:147 """Resolve a local path and verify it stays inside approved prefixes."""148 normalized = os.path.normpath(file_path)149 150 # Block path traversal151 if ".." in normalized:152 return None, "Path traversal (..) is not allowed"153 154 workspace_root = _configured_workspace_root()155 all_prefixes = _allowed_local_prefixes(156 write_access=write_access,157 workspace_root=workspace_root,158 )159 # The blocked-name test must see the REAL target, so it runs against the160 # fully-resolved name as well as the candidate's own. A curated161 # ``plugins/skills/x/SKILL.md -> .env.prod`` link is authorized through the162 # unresolved candidate (that is the point of the exception), and judging163 # only that candidate would let the link's harmless name stand in for the164 # secret it points at.165 target_name = Path(normalized).resolve().name166 for candidate in _candidate_paths(167 normalized, workspace_root, write_access=write_access,168 ):169 resolved = str(candidate)170 for prefix in all_prefixes:171 prefix_resolved = _resolve_prefix(prefix)172 if _path_within(resolved, prefix_resolved):173 blocked = _blocked_name(candidate.name) or _blocked_name(target_name)174 if blocked:175 return None, (176 f"Access to files matching '{blocked}' is blocked for security"177 )178 return candidate, ""179 180 return None, f"Access restricted. Allowed directories: {', '.join(all_prefixes)}"181 182 183def _is_path_allowed(file_path: str, *, write_access: bool = False) -> tuple[bool, str]:184 """Check if a local path is allowed for the requested access mode."""185 resolved_path, reason = _authorized_local_path(file_path, write_access=write_access)186 return resolved_path is not None, reason187 188 189def _resolve_prefix(prefix: str) -> str:190 """Resolve an allowed prefix to an absolute, symlink-resolved path.191 192 Uses Path.resolve() for all paths so that symlink targets match193 (e.g., on macOS /tmp → /private/tmp).194 """195 return str(Path(prefix).resolve())196 197 198def _path_within(path: str, prefix: str) -> bool:199 """Check whether a resolved path is inside a prefix."""200 norm_path = os.path.normpath(path)201 norm_prefix = os.path.normpath(prefix)202 return norm_path == norm_prefix or norm_path.startswith(norm_prefix + os.sep)203 204 205def _resolve_inputs_dir() -> Path | None:206 """Return the mounted read-only ``/inputs`` dir, but only when it exists.207 208 Task input files are bind-mounted read-only at ``/inputs`` (container/serve209 mode; overridable via ``FRONTIER_AGENT_INPUTS_DIR``). Gating on the dir actually210 existing means non-container runs (local bwrap, tests — no ``/inputs``) get211 no new prefix and are unaffected. Imported lazily to avoid an import cycle212 with ``_sandbox``.213 """214 try:215 from plugins.tools._sandbox import resolve_mount_dirs216 217 inputs_dir = Path(resolve_mount_dirs()[2]).expanduser().resolve()218 except Exception:219 return None220 return inputs_dir if inputs_dir.is_dir() else None221 222 223def task_input_matcher() -> Callable[[str | Path], bool]:224 """Resolve the read-only input root once and return a per-path predicate.225 226 Task inputs may intentionally live below a repository-ignored runtime227 directory (for example ``.apodex/``). Search tools use this signal to avoid228 applying repository ignore rules to the separately-authorized input mount;229 normal path authorization and per-result symlink checks still apply.230 231 The root lookup imports ``_sandbox``, reads the environment and stats the232 mount, so it must not run once per candidate file: a search over a large233 checkout would spend more time re-deriving a constant than reading files.234 Runs with no input mount get a predicate that costs nothing at all.235 """236 inputs_dir = _resolve_inputs_dir()237 if inputs_dir is None:238 return lambda _file_path: False239 root = str(inputs_dir)240 241 def _within(file_path: str | Path) -> bool:242 try:243 candidate = Path(file_path).expanduser().resolve()244 except (OSError, RuntimeError):245 return False246 return _path_within(str(candidate), root)247 248 return _within249 250 251def _resolve_spill_dirs() -> list[Path]:252 """Return the spill directories this conversation may read.253 254 Authorized for READ so ``read_file`` / ``grep_search`` can recover a body255 compaction dropped, and never for write — the same shape as ``/inputs``. The256 canonical ``/spill`` path a model sees is rewritten to this by257 ``resolve_runtime_path`` before it reaches here. Gating on existence means a258 run that never spilled adds no prefix. Imported lazily to avoid an import259 cycle with ``_sandbox``.260 """261 try:262 from plugins.tools._overflow import _created_stores, _current_task_id263 from plugins.tools._overflow import _scope_component as scope_of264 from plugins.tools._sandbox import spill_root265 266 root = spill_root()267 except Exception:268 return []269 if not root.is_dir():270 return []271 272 # Narrower than the root on purpose. The root is shared — a temp directory,273 # or a run directory — so authorizing it would let one conversation read274 # another's spilled tool results, which the old in-workspace layout made275 # impossible. Two things are authorized instead:276 #277 # * this conversation's own scope, which is what its recovery index names;278 # * every store THIS process created, because in-process sub-agents spill279 # under their own scope and a fan-in report can carry one of those paths280 # back to the parent.281 #282 # A different session in a different process matches neither.283 allowed: list[Path] = []284 scope = scope_of(_current_task_id())285 if scope and (root / scope).is_dir():286 allowed.append(root / scope)287 allowed.extend(store for store in _created_stores if store.is_dir())288 return allowed289 290 291def _allowed_local_prefixes(292 *,293 write_access: bool = False,294 workspace_root: Path | None = None,295) -> list[str]:296 """Return the local path prefixes allowed in the current execution context."""297 prefixes = list(_ALLOWED_RELATIVE_PREFIXES) + list(_ALLOWED_ABSOLUTE_PREFIXES)298 resolved_workspace_root = workspace_root or _configured_workspace_root()299 if resolved_workspace_root is not None:300 if write_access and not _is_isolated_workspace_root(resolved_workspace_root):301 logger.warning(302 "Refusing local write access to non-isolated workspace root '%s'",303 resolved_workspace_root,304 )305 else:306 prefixes.append(str(resolved_workspace_root))307 # Uploaded task inputs live under a read-only ``/inputs`` mount. Authorize308 # them for READ so grep_search / glob_search / read_text can list and search309 # them; never for write (the mount is read-only).310 if not write_access:311 inputs_dir = _resolve_inputs_dir()312 if inputs_dir is not None:313 prefixes.append(str(inputs_dir))314 # Recovery reads of spilled tool results. READ ONLY, and deliberately315 # absent from the write branch: that omission is what makes the store316 # read-only to every file tool, replacing a special case each writer had317 # to remember.318 prefixes.extend(str(path) for path in _resolve_spill_dirs())319 return prefixes320