"""Demo-safe boundaries: secret redaction, tool policy, download containment. Three independent jobs, deliberately kept in one small module so a reviewer can see the whole public-exposure surface at once: 1. **Secrets never leave the process** — not in logs, not in the UI, not in a traceback, not in a downloadable file. Three distinct escapes have to be closed for that claim to hold: the answer *stream* (where a secret can straddle two SSE chunks, so :class:`StreamRedactor` is stateful), *tool arguments* (which reach the tool verbatim, hence the rewrite in ``containment.SecretArgumentObserver``), and *file contents* (hence the optional scan in :func:`list_output_files`). 2. **The agent gets a demo-safe toolset** — real research and real file deliverables, no arbitrary shell, no package installs, no subprocess. 3. **Downloads cannot escape the session's ``outputs/``** — not via ``..``, not via an absolute path, not via a symlink. """ from __future__ import annotations import logging import os import re from collections.abc import Iterable, Sequence from dataclasses import dataclass from pathlib import Path from typing import Any #: Shortest secret we will substring-match. Below this, a "secret" would match #: so much ordinary text that redaction would destroy the logs it protects. _MIN_SECRET_LEN = 8 logger = logging.getLogger(__name__) REDACTED = "***REDACTED***" #: Credential shapes worth catching even when we were never told the value — #: e.g. a key the *upstream endpoint* echoes back inside an error body. _SECRET_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"\bsk-[A-Za-z0-9_\-]{16,}"), re.compile(r"\bhf_[A-Za-z0-9]{16,}"), re.compile(r"\b[Bb]earer\s+[A-Za-z0-9._\-]{16,}"), # The key name may be quoted (JSON), so allow a closing quote before the # separator: {"api_key": "…"} must match as readily as api_key=… re.compile( r"(?i)\b(api[_-]?key|authorization|token)\b[\"']?\s*[:=]\s*[\"']?" r"([A-Za-z0-9._\-]{12,})", ), ) #: Tools the demo refuses regardless of what the allowlist asks for. This is #: the fail-closed half of the policy: ``allowed_tools`` can only ever narrow #: the set further, never widen it past this line. HARD_DENIED_TOOLS: frozenset[str] = frozenset({ # Arbitrary command / code execution. "bash", "run_python_code", # Unbounded network writes to the local filesystem. "download_file", # Sub-agent orchestration: denied by default; Agent Team admits only this # internal subset explicitly in ``demo_safe_tool_policy`` below. "create_subagent", "assign_task", "stop_subagent", "collect_reports", "submit_report", # Extra editing surface with no P0 use. ``write_file`` is the one write path # the demo keeps: it writes in-process through ``plugins/tools/_path_auth``, # which is fail-closed and confined to the session's authorised tree. "file_editor_create", "file_editor_str_replace", }) # Internal message-passing tools required by Agent Team. They remain denied in # ReAct mode and are selectively admitted only when the adapter starts the # agent-team pipeline. None executes code or writes outside the runtime's own # in-memory coordination/report channel. AGENT_TEAM_INTERNAL_TOOLS: frozenset[str] = frozenset({ "create_subagent", "assign_task", "stop_subagent", "collect_reports", "submit_report", }) #: Names never offered for download even when they land in ``outputs/``. _BLOCKED_DOWNLOAD_NAMES: frozenset[str] = frozenset({ ".env", ".env.local", "config.yaml", "id_rsa", ".netrc", ".git-credentials", }) #: Substrings that make a filename undownloadable (defence in depth with the #: runtime's own ``plugins/tools/_path_auth._BLOCKED_PATTERNS``). _BLOCKED_DOWNLOAD_SUBSTRINGS: tuple[str, ...] = ( "secret", "credential", "password", "token", ".pem", ".key", ".crt", ) class DownloadDenied(PermissionError): """A requested download is outside the session's ``outputs/`` directory.""" @dataclass(frozen=True) class Redactor: """Replaces known secret values (and credential-shaped text) with a mask.""" literals: tuple[str, ...] = () @classmethod def for_secrets(cls, secrets: Iterable[str]) -> Redactor: literals = tuple(sorted( {s for s in (str(x or "").strip() for x in secrets) if len(s) >= _MIN_SECRET_LEN}, key=len, reverse=True, # longest first, so a prefix never masks a superstring )) return cls(literals=literals) def __call__(self, value: str) -> str: return self.redact(value) def redact(self, value: str) -> str: if not value: return value out = self.redact_literals(value) for pattern in _SECRET_PATTERNS: out = pattern.sub(_mask_match, out) return out def redact_literals(self, value: str) -> str: """Literal-only pass, for rewriting content rather than previewing it. Used where the text is data the agent will act on (tool arguments, file content), so only *configured* secrets are touched. Credential-shaped text that is not one of our secrets is left alone — mangling it would corrupt legitimate deliverables to protect nothing. """ if not value: return value out = value for literal in self.literals: if literal in out: out = out.replace(literal, REDACTED) return out def contains_secret(self, value: str) -> bool: """True when ``value`` carries a configured secret verbatim.""" return bool(value) and any(lit in value for lit in self.literals) @property def longest_literal(self) -> int: return max((len(lit) for lit in self.literals), default=0) #: Minimum tail a :class:`StreamRedactor` holds back. Sized so the regex #: patterns above also get a chance to match across a chunk boundary, not just #: the known literals. _MIN_STREAM_HOLD = 48 class StreamRedactor: """Redacts a token stream, tolerating secrets split across chunks. SSE chunk boundaries are arbitrary. Redacting each delta independently is therefore unsound: a secret split as ``"sk-fake-demo-"`` + ``"key-1234"`` matches neither half, both are emitted verbatim, and whoever concatenates the deltas — the browser, a log, an event consumer — gets the whole thing back. So this keeps a rolling buffer and withholds the last ``hold`` characters until more text arrives or :meth:`flush` is called. The cost is that the visible answer trails the stream by at most ``hold`` characters. """ def __init__(self, redactor: Redactor, *, hold: int | None = None) -> None: self._redactor = redactor # A literal of length L can only be recognised if at least L-1 of its # characters are still buffered when the rest arrives. needed = max(redactor.longest_literal - 1, 0) self._hold = max(hold if hold is not None else needed, _MIN_STREAM_HOLD) self._buffer = "" @property def pending(self) -> int: return len(self._buffer) def feed(self, chunk: str) -> str: """Absorb ``chunk`` and return the text that is now safe to emit.""" if not chunk: return "" redacted = self._redactor.redact(self._buffer + chunk) if len(redacted) <= self._hold: self._buffer = redacted return "" self._buffer = redacted[-self._hold:] return redacted[: -self._hold] def flush(self) -> str: """Emit whatever is still held back. Call this at end of stream.""" remaining = self._redactor.redact(self._buffer) self._buffer = "" return remaining def discard(self) -> None: """Throw away the held-back tail. For an attempt the runtime discarded: that text is not part of the answer, so flushing it would splice a dead draft onto the retry. """ self._buffer = "" def redact_deep(value: Any, redactor: Redactor) -> Any: """Redact configured secrets in every string inside a nested structure. Tool arguments are not always flat — ``create_file`` takes a nested ops program, for instance — so a shallow pass would miss the interesting cases. """ if isinstance(value, str): return redactor.redact_literals(value) if isinstance(value, dict): return {k: redact_deep(v, redactor) for k, v in value.items()} if isinstance(value, (list, tuple)): rebuilt = [redact_deep(v, redactor) for v in value] return type(value)(rebuilt) if isinstance(value, tuple) else rebuilt return value def _mask_match(match: re.Match[str]) -> str: """Keep any label the pattern captured; mask only the credential itself.""" if match.re.groups >= 2: return f"{match.group(1)}={REDACTED}" return REDACTED class RedactingLogFilter(logging.Filter): """Scrub secrets from every log record that passes through a handler. Installed on the *root* logger's handlers, so it also covers third-party libraries (openai/httpx) that may echo a request header on error. """ def __init__(self, redactor: Redactor) -> None: super().__init__() self._redactor = redactor def filter(self, record: logging.LogRecord) -> bool: try: record.msg = self._redactor.redact(str(record.msg)) if record.args: if isinstance(record.args, dict): record.args = { k: self._redactor.redact(v) if isinstance(v, str) else v for k, v in record.args.items() } elif isinstance(record.args, tuple): record.args = tuple( self._redactor.redact(a) if isinstance(a, str) else a for a in record.args ) except Exception: pass return True def install_log_redaction(redactor: Redactor) -> None: """Attach ``redactor`` to root logging (idempotent).""" root = logging.getLogger() if not root.handlers: logging.basicConfig(level=logging.WARNING) for handler in root.handlers: if any(isinstance(f, RedactingLogFilter) for f in handler.filters): continue handler.addFilter(RedactingLogFilter(redactor)) def demo_safe_tool_names( allowed: Sequence[str], *, public_mode: bool = True, ) -> tuple[str, ...]: """Intersect the requested allowlist with what a demo may expose.""" denied = HARD_DENIED_TOOLS if public_mode else frozenset() return tuple( name for name in dict.fromkeys(allowed) if name and name.lower() not in denied ) def demo_safe_tool_policy( allowed: Sequence[str], *, public_mode: bool = True, workflow: str = "react", ) -> Any: """Build the runtime's process-wide tool policy for this demo. Returns a ``ToolPermissionContext`` combining an allowlist (only these tools may bind) with the hard deny list (fail-closed even if a profile, a role definition, or a future allowlist edit tries to re-add one). """ from frontier_agent.core.runtime.resources.tool_permission import ( ToolPermissionContext, ) names = set(demo_safe_tool_names(allowed, public_mode=public_mode)) denied = set(HARD_DENIED_TOOLS) if public_mode else set() # Independent of ``public_mode``: the allowlist is the *only* thing that # lets a tool bind, and ``DEMO_ALLOWED_TOOLS`` never names these internal # orchestration tools. Gating this on public mode left a non-public # operator with a coordinator that had no delegation tools at all — # ``_resolve_profile_tools`` skips denied names with a log line, so Agent # Team degraded silently into a solo agent. if workflow == "agent_team": names.update(AGENT_TEAM_INTERNAL_TOOLS) denied.difference_update(AGENT_TEAM_INTERNAL_TOOLS) return ToolPermissionContext.from_iterables( allow_names=names, deny_names=denied, ) def is_downloadable(path: Path) -> bool: """True when ``path`` is a plain file safe to offer to a browser.""" name = path.name if name.startswith("."): return False if name in _BLOCKED_DOWNLOAD_NAMES: return False lowered = name.lower() if any(token in lowered for token in _BLOCKED_DOWNLOAD_SUBSTRINGS): return False try: return path.is_file() and not path.is_symlink() except OSError: return False #: Largest file whose *contents* are scanned for secrets before being offered #: for download. Beyond this, reading the file to check it would cost more than #: the check is worth; the tool-call boundary is the primary defence. _MAX_SCAN_BYTES = 4 * 1024 * 1024 def list_output_files( outputs_dir: Path, *, limit: int = 200, redactor: Redactor | None = None, ) -> list[Path]: """Every downloadable file under ``outputs_dir``, sorted, symlinks excluded. When ``redactor`` is supplied, files whose *contents* carry a configured secret are withheld as well. This is the second line: secrets are already stripped from tool arguments before a file can be written, but a filename check alone would let any other write path leak a key to a visitor. """ root = Path(outputs_dir) if not root.is_dir(): return [] found: list[Path] = [] try: real_root = root.resolve(strict=True) except OSError: return [] for candidate in sorted(root.rglob("*")): if len(found) >= limit: break if candidate.is_symlink() or not is_downloadable(candidate): continue try: # A file reached through a symlinked *directory* must not escape. candidate.resolve(strict=True).relative_to(real_root) except (OSError, ValueError): continue if redactor is not None and _holds_secret(candidate, redactor): logger.warning( "withholding %s from download: it contains a configured secret", candidate.name, ) continue found.append(candidate) return found def _holds_secret(path: Path, redactor: Redactor) -> bool: """True when ``path`` looks like text and contains a configured secret.""" if not redactor.literals: return False try: if path.stat().st_size > _MAX_SCAN_BYTES: return False body = path.read_text(encoding="utf-8", errors="ignore") except OSError: return True # unreadable: fail closed rather than serve it blind return redactor.contains_secret(body) def resolve_download(outputs_dir: Path, requested: str) -> Path: """Resolve ``requested`` to a real file inside ``outputs_dir``. Raises :class:`DownloadDenied` for anything else — traversal, absolute paths, symlinks out of the tree, non-files, and blocked names. This is the only function the UI may use to turn user input into a served path. """ root = Path(outputs_dir) try: real_root = root.resolve(strict=True) except OSError as exc: raise DownloadDenied("this session has no outputs directory") from exc raw = str(requested or "").strip() if not raw: raise DownloadDenied("no file requested") if raw.startswith(("/", "\\")) or (os.name == "nt" and ":" in raw[:3]): raise DownloadDenied("absolute paths are not downloadable") candidate = (real_root / raw).resolve() try: candidate.relative_to(real_root) except ValueError as exc: raise DownloadDenied( "the requested path is outside this session's outputs directory", ) from exc if candidate.is_symlink() or not candidate.is_file(): raise DownloadDenied("not a downloadable file") if not is_downloadable(candidate): raise DownloadDenied(f"{candidate.name!r} is not offered for download") return candidate __all__ = [ "AGENT_TEAM_INTERNAL_TOOLS", "HARD_DENIED_TOOLS", "REDACTED", "DownloadDenied", "RedactingLogFilter", "Redactor", "StreamRedactor", "demo_safe_tool_names", "demo_safe_tool_policy", "install_log_redaction", "is_downloadable", "list_output_files", "redact_deep", "resolve_download", ]