external_content_gate.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/external_content_gate.py
 
"""
External-content confirmation gate (M8.0 WU-8; §7.3, F67 / F68).
 
Purpose
-------
`fetch_url` and `web_search` pull EXTERNAL, untrusted content into the deciding
context — the residual T1 risk is a generic injection in fetched content driving
an in-process ACE and outbound exfil. The T1 "trusted inputs" premise is enforced
rather than assumed: such a fetch is **confirmed before it happens** with three
choices — **allow once · trust this source for the session · deny.** A session-
trusted source skips later prompts; a headless run with no confirmer fails closed
(the react-loop relay denies it). The gate is **on by default in every profile**;
it is disabled only out-of-band (``OC_CHAT_EXTERNAL_CONTENT_GATE``), and the
disablement is logged as a risk.
 
``recall`` is intentionally NOT gated in M8.0: no stored memory carries a
non-``trusted`` provenance yet, so there is nothing to gate; recall-gating rides
on the M8.3 memory-provenance derivation.
 
Responsibilities
----------------
- :func:`gate_enabled` / :func:`normalize_source` — enablement + the canonical
  host key for a URL/host source.
- :func:`external_fetch_allowed` — the gate decision (off / trusted / one-shot).
- :func:`trust_source_for_session` / :func:`reset_session_trust` /
  :func:`allow_external_source_once` — the session trust set + the one-shot grant.
- :func:`make_external_gate_block` / :func:`deny_external_fetch` — the
  ``requires_confirmation`` signal a gated tool returns, and its headless denial.
- :func:`blast_radius_banner` — the F68 ``permissive`` account-compromise banner.
 
Contracts
---------
- The decision state lives here (a session-scoped trusted-source set + a one-shot
  ``ContextVar`` grant the model cannot reach); the tools (``web.py``) ask
  :func:`external_fetch_allowed` and, when blocked, return
  :func:`make_external_gate_block`; the react-loop relays the three-choice prompt
  and, on approval, re-runs the tool under :func:`allow_external_source_once` or
  after :func:`trust_source_for_session`.
- A one-shot grant is reset on context exit and is never recorded as session trust.
- :func:`blast_radius_banner` (AC-V09-6) returns the banner ONLY when the active
  profile is ``permissive`` AND the gate is active; otherwise ``None``.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.risk_profile (lazy; banner profile resolution)
- oc_chat.tools._local_fetch (lazy; local host set that makes a local source's trust key carry its port)
 
Diagnostics
-----------
Domain: OC_CHAT.CORE
Levels:
    L1 — gate disablement (a risk signal)
    L2 — trust grants / lifecycle events
    L3 — per-decision detail
    L4 — deep tracing
"""
from __future__ import annotations
 
import contextlib
import contextvars
import ipaddress
import os
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from urllib.parse import urlsplit
 
from oc_diagnostics import _dbg
 
#: Out-of-band disable knob (process env only; stripped from a workspace ``.env``
#: by :mod:`oc_chat._env_guard`). Any of these values turns the gate OFF.
GATE_ENV_VAR = "OC_CHAT_EXTERNAL_CONTENT_GATE"
_DISABLE_VALUES = frozenset({"0", "off", "false", "no", "disabled"})
 
#: Sources the operator has trusted for THIS session (process-lifetime; reset on a
#: new/cleared session). Host-granularity, exact match — no implicit subdomain share.
_SESSION_TRUSTED_SOURCES: set[str] = set()
 
#: One-shot "allow once" grant the react-loop sets around a single re-dispatch.
#: A ``ContextVar`` (not a parameter/env) so the model cannot forge it.
_allow_once_source: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "oc_chat_allow_external_source_once", default=None
)
 
_disable_logged = False
 
#: Trust key of the session bound on the most recent turn (F67 session-scope).
#: ``("sid", <canonical uuid4>)`` for a session carrying a validated canonical
#: ``session_id``; ``("legacy", id, name)`` for a legacy double that lacks one
#: (plan 152 D9 — the explicitly-scoped compatibility fallback).
_active_session_key: tuple | None = None
 
 
#: In-process ``session_id -> canonical_record_locator`` binding (plan 152 R11-10).
#: Seeded at EVERY activation (startup, resume, new, load, switch) and updated on
#: save-as/rename. Presenting an id from a DIFFERENT record than the one that owns
#: it in this process means the presenting record is not canonical.
_id_to_locator: dict[str, str] = {}
 
 
def seed_record_binding(session_id: object, locator: object) -> None:
    """Bind *session_id* to the record it was activated from.
 
    Called at every activation. Without seeding, a process that starts on a
    COPIED record B(X), establishes trust, and then live-loads the original
    A(X) would accept A as X without rotating trust — the "a fresh process
    starts empty" argument stops applying the moment trust precedes the first
    live load.
    """
    if isinstance(session_id, str) and session_id and isinstance(locator, str) and locator:
        _id_to_locator[session_id] = locator
 
 
def is_canonical_record(session_id: object, locator: object) -> bool:
    """True if *locator* is the record this process already binds *session_id* to.
 
    An id never seen before is accepted (and should then be seeded) — the rule
    detects a SECOND record presenting an id this process already owns, which is
    the reachable trust-inheritance schedule. A same-record reload is NOT a
    duplicate, because the locator is the session directory rather than the
    per-save stamped meta filename.
 
    Residual, stated honestly: a cross-RUN whole-directory clone with an edited
    name is indistinguishable without a global uniqueness index. That closes
    when the sessions store makes ``session_id`` a primary key.
    """
    if not isinstance(session_id, str) or not session_id:
        return False
    known = _id_to_locator.get(session_id)
    if known is None:
        return True                        # first sighting in this process
    return isinstance(locator, str) and locator == known
 
 
def _reset_record_bindings_for_tests() -> None:
    """Test-only: clear the in-process id->locator bindings."""
    _id_to_locator.clear()
 
 
@dataclass(frozen=True)
class PreparedRecordBinding:
    """A prebuilt ``session_id -> canonical_record_locator`` mapping (plan 152-b D4).
 
    :func:`seed_record_binding` mutates the process-global ``_id_to_locator``
    IMMEDIATELY, which is a live effect a transition's PREPARE phase must not
    perform: on abort it would leave the D9 record-ownership map polluted, and
    the landed :class:`PreparedF67Binding` carries no locator to install it with.
 
    The load/switch seam therefore prepares this value in PREPARE (pure, zero
    global effect) and installs it via :func:`install_prepared_record_binding`
    as ONE no-throw step inside the commit critical section, immediately before
    identity publication — so ``_id_to_locator`` joins the all-or-nothing
    old/new tuple oracle instead of being written ahead of the commit.
    """
 
    session_id: str
    locator: str
 
 
def prepare_record_binding(session_id: object, locator: object) -> "PreparedRecordBinding | None":
    """Build the :class:`PreparedRecordBinding` for a load/switch candidate — PREPARE half.
 
    Mutates nothing. Returns ``None`` when either input is not a usable string,
    mirroring :func:`seed_record_binding`'s own guard, so a ``None`` result is a
    silent no-op at commit exactly as an unusable ``seed_record_binding`` call is.
    """
    if isinstance(session_id, str) and session_id and isinstance(locator, str) and locator:
        return PreparedRecordBinding(session_id=session_id, locator=locator)
    return None
 
 
def install_prepared_record_binding(binding: "PreparedRecordBinding") -> None:
    """Publish a prepared record binding — COMMIT half. CANNOT RAISE.
 
    A single dict assignment on the process-global ``_id_to_locator`` — no
    validation, no allocation, no logging — so it is safe inside the no-fail
    commit block. The validation already happened in :func:`prepare_record_binding`.
    """
    _id_to_locator[binding.session_id] = binding.locator
 
 
def is_canonical_session_id(raw: object) -> bool:
    """True iff *raw* is EXACTLY the canonical lowercase RFC form of a uuid4.
 
    The pinned predicate (plan 152 D9 / R09-05), with NO pre-parse
    normalisation::
 
        parsed = uuid.UUID(raw); accept = (parsed.version == 4 and raw == str(parsed))
 
    Because ``str(parsed)`` is the canonical lowercase 8-4-4-4-12 form, this
    accepts only that exact spelling and REJECTS uppercase, outer whitespace,
    braces, ``urn:uuid:``, compact hex, and every non-v4 version. A ``.lower()``
    or ``.strip()`` anywhere in here would silently accept forms the acceptance
    criteria require to be rejected.
 
    A trust key derived from an unvalidated string is not a complete gate, so
    this is the single definition both the gate and the persistence boundary
    (:func:`oc_chat.core.session.apply_session_metadata_fields`) share.
    """
    if not isinstance(raw, str):
        return False
    try:
        parsed = uuid.UUID(raw)
    except Exception:
        return False                       # malformed / non-parseable -> not canonical
    return parsed.version == 4 and raw == str(parsed)
 
 
def session_trust_key(session: object) -> tuple:
    """Derive the F67 trust key for *session* (plan 152 D9).
 
    Keyed on the validated canonical ``session_id`` so that identity — not
    object address and not display name — decides trust scope:
 
    - a RENAME preserves the id, so trust is RETAINED (same identity);
    - a load/switch to a DIFFERENT id rotates the key even when the object
      address and name are unchanged (the pre-fix leak);
    - a session lacking a canonical id falls back to the legacy ``(id, name)``
      shape. That fallback exists for test doubles only; production sessions
      always carry a canonical id because the persistence boundary regenerates
      one on a malformed or missing value.
    """
    sid = getattr(session, "session_id", None)
    if is_canonical_session_id(sid):
        return ("sid", sid)
    return ("legacy", id(session), getattr(session, "name", None))
 
 
@dataclass(frozen=True)
class PreparedF67Binding:
    """A fully-built F67 trust rotation, ready for a no-throw install (R10-01).
 
    Everything fallible or allocating — reading the candidate's id, validating
    it, deriving the key, deciding whether trust rotates — happens when this
    value is constructed during a transition's PREPARE phase, so
    :func:`install_prepared_session_binding` can be a bare pair of in-memory
    mutations inside the no-fail commit block.
    """
 
    #: The key to publish (already validated).
    key: tuple
    #: Whether installing this binding clears the session trust set.
    clears_trust: bool
 
 
def prepare_session_binding(session: object) -> PreparedF67Binding:
    """Build the :class:`PreparedF67Binding` for *session* — PREPARE half.
 
    Does the validation, the key derivation and the rotate-or-retain decision
    up front. Safe to call outside a transition; it mutates nothing.
    """
    key = session_trust_key(session)
    return PreparedF67Binding(key=key, clears_trust=key != _active_session_key)
 
 
def install_prepared_session_binding(binding: PreparedF67Binding) -> None:
    """Publish a prepared trust rotation — COMMIT half. CANNOT RAISE.
 
    Deliberately contains ONLY in-memory mutations: no validation, no logging,
    no formatting, no allocation. The previous shape called
    :func:`bind_active_session`, which branches and calls ``_dbg`` — and ``_dbg``
    emits a ``warnings.warn`` before diagnostics are initialised, which RAISES
    under warnings-as-errors. That made trust rotation fallible and bundled it
    with identity publication; this split is what lets the commit block install
    trust BEFORE identity with no window where the new identity carries old
    trust (plan 152 R10-01).
 
    The lifecycle diagnostic is deferred to
    :func:`log_session_binding_lifecycle`, called post-transaction, best-effort.
    """
    global _active_session_key
    _active_session_key = binding.key
    if binding.clears_trust:
        _SESSION_TRUSTED_SOURCES.clear()
 
 
def log_session_binding_lifecycle(binding: PreparedF67Binding) -> None:
    """Emit the F67 rotation diagnostic AFTER the transaction (best-effort).
 
    Separated from :func:`install_prepared_session_binding` so that nothing
    inside the transition's critical section can raise. Callers run this in the
    post-transaction best-effort step and contain its failure.
    """
    func = "log_session_binding_lifecycle"
    if binding.clears_trust:
        _dbg("OC_CHAT.CORE", func, "active session changed -> external trust reset", 2)
 
 
def gate_enabled() -> bool:
    """True unless the gate is disabled out-of-band (logged once as a risk)."""
    func = "gate_enabled"
    global _disable_logged
    raw = os.environ.get(GATE_ENV_VAR, "").strip().lower()
    if raw in _DISABLE_VALUES:
        if not _disable_logged:
            _dbg("OC_CHAT.CORE", func,
                 f"external-content gate DISABLED via {GATE_ENV_VAR}={raw!r} "
                 f"(F67 off — fetched content is no longer confirm-gated)", 1, override=True)
            _disable_logged = True
        return False
    return True
 
 
def normalize_source(target: str) -> str:
    """Canonical trust key for *target* (a URL or a bare host).
 
    Uses ``urlsplit().hostname`` (already lowercased, trailing-dot-stripped,
    punycode-decoded, IPv6 unbracketed; userinfo excluded) and compresses an IP
    literal via :mod:`ipaddress`. Exact-host granularity: ``trust example.com``
    grants neither ``sub.example.com`` nor a parent domain.
 
    **Plan 232 P7 — a LOCAL source's key carries its port** (``localhost:8000``,
    ``[::1]:8000``). On the public internet a host IS the trust boundary and the
    port is noise, which is why every non-local key stays host-only and unchanged.
    On loopback the opposite holds: one port is a dev server the human chose to
    expose, while the others are whatever else that developer happens to be
    running — a database admin console, another project's API, an SSH forward. So
    "trust this source for the session" must mean one dev server, not the whole
    loopback surface. The local host set is imported from the local-fetch module
    rather than restated here, so the set that ADMITS a local fetch and the set
    that KEYS its trust can never drift apart. A local URL with no explicit port
    keeps the bare-host key: it is never admitted by the fetch rule anyway, and a
    bare key can only ever fail to match a port-qualified grant, never widen one.
    """
    s = (target or "").strip()
    parts = urlsplit(s if "://" in s else "//" + s, scheme="https")
    host = (parts.hostname or "").strip().lower().rstrip(".")
    if not host:
        return s.lower()  # malformed — egress will reject it anyway
    try:
        host = ipaddress.ip_address(host).compressed
    except ValueError:
        pass  # not an IP literal — keep the DNS name
    # Lazy import: `_local_fetch` pulls in nothing from oc_chat at module level,
    # so this cannot cycle back through core.
    from oc_chat.tools._local_fetch import LOCAL_HOSTS
    if host in LOCAL_HOSTS:
        try:
            port = parts.port
        except ValueError:
            port = None  # malformed port — fall back to the bare host key
        if port is not None:
            return f"[{host}]:{port}" if ":" in host else f"{host}:{port}"
    return host
 
 
def is_source_trusted(source: str) -> bool:
    """True if *source* was trusted for the session (or one-shot-granted now)."""
    return source in _SESSION_TRUSTED_SOURCES or _allow_once_source.get() == source
 
 
def external_fetch_allowed(source: str) -> bool:
    """The gate decision: allowed if the gate is off, or *source* is trusted/granted."""
    if not gate_enabled():
        return True
    return is_source_trusted(source)
 
 
def trust_source_for_session(source: str) -> None:
    """Trust *source* for the rest of this session (skips later prompts)."""
    func = "trust_source_for_session"
    _SESSION_TRUSTED_SOURCES.add(source)
    _dbg("OC_CHAT.CORE", func, f"session-trusted external source: {source!r}", 2)
 
 
def reset_session_trust() -> None:
    """Clear all session-trusted sources (call at session start / new / clear)."""
    _SESSION_TRUSTED_SOURCES.clear()
 
 
def bind_active_session(session: object) -> None:
    """Reset the trust set when the active session changes (F67 session-scope).
 
    Called once per turn by the react-loop. Keyed on the VALIDATED CANONICAL
    ``session_id`` (plan 152 D9) via :func:`session_trust_key`, so a new /
    loaded / switched / freshly-created session starts with an empty trust set —
    a source trusted in one session never carries into another, no matter which
    transition path created it. Keying on identity rather than ``(object id,
    name)`` closes two defects of the old key: a same-object/same-name load
    carrying a DIFFERENT persisted id used to keep its trust set (a leak), and a
    RENAME used to reset trust although identity was preserved.
 
    (An in-place ``/session clear`` keeps the same session;
    :meth:`ChatSession.clear` resets trust for that case.)
 
    Replacement transitions no longer depend on this path: they install the
    rotation explicitly at commit via :func:`install_prepared_session_binding`,
    before publishing the new identity. This remains the per-turn safety net.
    """
    func = "bind_active_session"
    global _active_session_key
    key = session_trust_key(session)
    if key != _active_session_key:
        if _SESSION_TRUSTED_SOURCES:
            _dbg("OC_CHAT.CORE", func, "active session changed -> external trust reset", 2)
        _SESSION_TRUSTED_SOURCES.clear()
        _active_session_key = key
 
 
@contextlib.contextmanager
def allow_external_source_once(source: str) -> Iterator[None]:
    """Authorize *source* for exactly one re-dispatch (allow-once).
 
    For use ONLY by the react-loop relay after an interactive 'a'. Not model-
    reachable (a ``ContextVar``, not a tool argument); reset on exit so the grant
    never outlives the single re-run it brackets and is not session-trust.
    """
    token = _allow_once_source.set(source)
    try:
        yield
    finally:
        _allow_once_source.reset(token)
 
 
def make_external_gate_block(tool: str, source: str, detail: str) -> dict:
    """The ``requires_confirmation`` signal a gated tool returns (relayed by the
    react-loop; failed closed by the headless sub-loop)."""
    return {
        "requires_confirmation": True,
        "gate": "external_content",
        "tool": tool,
        "source": source,
        "detail": detail,
        "message": (
            f"{tool} would pull EXTERNAL, untrusted content from '{source}'"
            + (f" ({detail})" if detail else "")
            + ". External content can carry injected instructions. A human must "
            "approve it: allow once, trust this source for the session, or deny."
        ),
    }
 
 
def deny_external_fetch(result: dict, *, note: str = "") -> dict:
    """Convert a gate block into a terminal ``error`` for contexts with no human
    (headless / sub-loop / declined). Any other result passes through unchanged."""
    if isinstance(result, dict) and result.get("requires_confirmation"):
        msg = result.get("message") or "External-content fetch requires human approval."
        return {"error": (msg + " " + note).strip()}
    return result
 
 
def blast_radius_banner() -> str | None:
    """The F68 ``permissive`` account-compromise banner, or ``None`` (AC-V09-6).
 
    Present ONLY when the active profile is ``permissive`` AND the F67 gate is
    active (a disabled gate removes the mitigation the banner is conditioned on).
    """
    if not gate_enabled():
        return None
    from oc_chat.core.risk_profile import resolve_risk_profile
    if resolve_risk_profile() != "permissive":
        return None
    return (
        "permissive profile: the agent runs in-process with your account's "
        "authority. A weaponized in-process code-execution (e.g. via injected "
        "content) can reach your full developer-account authority and exfiltrate "
        "secrets — including through the provider-egress transcript channel. The "
        "external-content confirmation gate (F67) is active; deny fetches you did "
        "not intend. Use the 'guarded' profile for less-trusted projects."
    )
 
 
__all__ = [
    "GATE_ENV_VAR",
    "gate_enabled",
    "normalize_source",
    "is_source_trusted",
    "external_fetch_allowed",
    "trust_source_for_session",
    "reset_session_trust",
    "bind_active_session",
    "allow_external_source_once",
    "make_external_gate_block",
    "deny_external_fetch",
    "blast_radius_banner",
]