provider_secrets.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/provider_secrets.py
 
"""
The provider-secrets sidecar — captured API keys, never a tracked byte.
 
Purpose
-------
Plan 216 P1. The onboarding wizard captures a provider API key with
masked input and puts it in exactly one place: a gitignored JSON sidecar
mapping env-var NAME -> value, beside the registry's
``.option_c/agent_secrets.json``. At startup the sidecar is loaded into
``os.environ`` for variables that are not already set, so a lane the
operator added through the wizard behaves exactly like one whose key was
exported by hand.
 
The value never leaves this module by any channel other than
``os.environ``: it is never printed, never logged, never passed to
:func:`_dbg` at any level, and never interpolated into an exception
message. Everything this module traces is a variable NAME.
 
Responsibilities
----------------
- :func:`is_gitignored` / :func:`assert_gitignored` — the W1 canary:
  ask git, at runtime, whether a path is actually covered.
- :func:`capture_key` — masked capture that returns ``None`` on cancel.
- :func:`load_secrets` / :func:`write_secret` — strict read, atomic write.
- :func:`apply_to_environ` — the startup hook; a real environment
  variable always wins.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (sidecar written, variables applied)
    L3 — semantic details (variable NAMES, counts)
    L4 — deep tracing
 
Contracts
---------
- **No key value is ever an argument to** :func:`_dbg`, ``logging``,
  ``print``, or an exception constructor. Messages carry the variable
  NAME and the file PATH only. This is asserted by W2, which captures
  all three channels at once.
- The gitignore check FAILS CLOSED (SECURITY_ERROR, S6 L1). A git error,
  an absent git binary, or a path outside any repository all count as
  "not covered", because none of them can promise the file will stay out
  of a commit. Refusing to write is recoverable; a committed key is not.
- ``check-ignore`` treats a TRACKED file as not-ignored, which is the
  behaviour this relies on: a path git already tracks is refused even if
  a stale rule appears to name it.
- :func:`load_secrets` is STRICT (a corrupt sidecar raises, naming the
  path only) while :func:`apply_to_environ` is FORGIVING (it degrades to
  "no variables applied"). A corrupt file must not stop OC Chat from
  starting, but it must not be silently indistinguishable from an empty
  one either.
- Only ``UPPER_SNAKE`` names reach the file — the same mechanical check
  the lane catalogue uses to keep "names, never values" verifiable.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- getpass (masked key capture; stdlib, never echoes the typed value)
- subprocess (the git check-ignore canary)
- json (sidecar serialisation)
- pathlib (path handling)
"""
from __future__ import annotations
 
import getpass
import json
import os
import re
import secrets as _secrets
import stat
import subprocess
from pathlib import Path
from typing import Callable, Mapping, MutableMapping
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
 
#: Where captured keys live. Override for tests / unusual layouts with
#: ``OC_CHAT_PROVIDER_SECRETS``. Relative by design: it resolves against the
#: launch directory exactly as ``.option_c/agent_secrets.json`` does.
DEFAULT_SECRETS_PATH: Path = Path(
    os.environ.get("OC_CHAT_PROVIDER_SECRETS", ".option_c/provider_secrets.json")
)
 
#: Owner read/write only, where the platform honours POSIX modes.
SECRET_FILE_MODE = 0o600
 
#: Env-var NAMES are UPPER_SNAKE — the same pattern the lane catalogue pins.
_ENV_VAR_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
 
#: Short bound on the git probe so a hung / networked checkout cannot wedge
#: the wizard. A timeout is an error, and an error fails closed.
_GIT_TIMEOUT_S = 5.0
 
 
class SecretsPathError(ValueError):
    """A sidecar target is not covered by gitignore — the write is refused.
 
    SECURITY_ERROR (S6 L1): writing a credential where git can see it is
    not a degraded mode to carry on in. The message names the PATH and
    never the value.
    """
 
 
class SecretsFileError(ValueError):
    """The sidecar exists but could not be parsed.
 
    VALIDATION_ERROR (S6 L3). The message names the PATH only — never the
    file's content, which is by definition credential material.
    """
 
 
def _existing_ancestor(path: Path) -> Path:
    """The nearest existing directory at or above *path*.
 
    ``git check-ignore`` must run somewhere inside a work tree, and the
    sidecar usually does not exist yet (that is the whole point of asking
    before writing). Walking up to something real makes the probe work for
    a path whose parent directory has not been created.
    """
    candidate = path if path.is_dir() else path.parent
    while not candidate.is_dir() and candidate != candidate.parent:
        candidate = candidate.parent
    return candidate
 
 
def is_gitignored(path: str | Path) -> bool:
    """Return whether git would ignore *path* — asked, not assumed (W1).
 
    Fails closed: any error, timeout, missing binary, or path outside a
    work tree returns ``False``.
    """
    func = "is_gitignored"
    target = Path(path).expanduser()
    try:
        target = target.resolve()
    except OSError:
        target = target.absolute()
    try:
        proc = subprocess.run(
            ["git", "check-ignore", "--quiet", "--", str(target)],
            cwd=str(_existing_ancestor(target)),
            capture_output=True,
            text=True,
            timeout=_GIT_TIMEOUT_S,
            check=False,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        # SECURITY_ERROR — cannot vouch for the path, so it is not covered.
        _dbg("OC_CHAT.MAIN", func,
             # OCT-LINT: disable=no_hardcoded_secrets reason="Diagnostic prose recording that the gitignore canary failed closed; names a path and an exception type, carries no credential (the key never reaches this function)"
             f"SECURITY_ERROR: check-ignore unavailable for {target}: "
             f"{type(exc).__name__}", 1, override=True)
        return False
    # 0 = ignored, 1 = not ignored, anything else (128 = not a repo) = error.
    covered = proc.returncode == 0
    _dbg("OC_CHAT.MAIN", func, f"{target} covered={covered} rc={proc.returncode}", 3)
    return covered
 
 
def assert_gitignored(path: str | Path) -> Path:
    """Return *path* when git ignores it, else raise :exc:`SecretsPathError`.
 
    The canary at the heart of P1. The message is actionable — it names
    the path and says what must be true — and carries no key material.
    """
    target = Path(path)
    if not is_gitignored(target):
        raise SecretsPathError(
            # OCT-LINT: disable=no_hardcoded_secrets reason="W1 refusal remediation prose telling the operator how to fix an uncovered path; high-entropy English, not a credential"
            f"refusing to write provider secrets to {target}: the path is not "
            f"covered by gitignore (checked with 'git check-ignore'). Add a "
            f"gitignore rule for it, or point the sidecar at a covered path "
            f"via OC_CHAT_PROVIDER_SECRETS."
        )
    return target
 
 
def capture_key(
    env_var: str,
    *,
    reader: Callable[[str], str] | None = None,
) -> str | None:
    """Prompt for a key with masked input; return ``None`` when cancelled.
 
    A bare Enter, whitespace, Ctrl+C or EOF all mean "cancel" — the
    wizard's cancel-safety contract (P5) starts here. *reader* is injected
    by tests; the default is :func:`getpass.getpass`, never :func:`input`,
    so the value is not echoed to the terminal.
    """
    func = "capture_key"
    prompt = f"  API key for {env_var} (input hidden, Enter to cancel): "
    read = reader if reader is not None else getpass.getpass
    try:
        raw = read(prompt)
    except (EOFError, KeyboardInterrupt):
        _dbg("OC_CHAT.MAIN", func, f"capture cancelled at prompt for {env_var}", 2)
        return None
    if raw is None or not raw.strip():
        _dbg("OC_CHAT.MAIN", func, f"empty capture for {env_var} treated as cancel", 2)
        return None
    # Length only. The value itself never reaches a diagnostic channel.
    _dbg("OC_CHAT.MAIN", func, f"captured a key for {env_var} ({len(raw.strip())} chars)", 2)
    return raw.strip()
 
 
def load_secrets(path: str | Path = DEFAULT_SECRETS_PATH) -> dict[str, str]:
    """Read the sidecar as a ``{NAME: value}`` map; ``{}`` when absent.
 
    Raises :exc:`SecretsFileError` on a corrupt or wrongly-shaped file,
    naming the path only.
    """
    source = Path(path)
    if not source.is_file():
        return {}
    try:
        data = json.loads(source.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        # Deliberately NOT chaining the parse error's text: a JSON decoder is
        # entitled to quote the offending document, and this document is a
        # credential store.
        raise SecretsFileError(
            f"could not parse the provider-secrets sidecar at {source} "
            f"({type(exc).__name__})"
        ) from None
    if not isinstance(data, Mapping):
        raise SecretsFileError(
            f"provider-secrets sidecar at {source} must contain a JSON object"
        )
    return {
        str(name): str(value)
        for name, value in data.items()
        if isinstance(name, str) and isinstance(value, str)
    }
 
 
def _write_atomically(payload: Mapping[str, str], target: Path) -> None:
    """Temp-then-rename write with a restrictive mode applied before publish."""
    target.parent.mkdir(parents=True, exist_ok=True)
    text = json.dumps(dict(sorted(payload.items())), indent=2, sort_keys=True) + "\n"
    tmp = target.with_name(f".tmp-{target.name}-{_secrets.token_hex(8)}")
    try:
        with tmp.open("w", encoding="utf-8", newline="\n") as handle:
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        _chmod_secret(tmp)
        os.replace(tmp, target)
    finally:
        # A failed publish must not leave a world-readable temp file holding
        # the key behind in the directory.
        if tmp.exists():
            try:
                tmp.unlink()
            except OSError:
                pass
    _chmod_secret(target)
 
 
def _chmod_secret(path: Path) -> None:
    """Best-effort owner-only mode; a no-op where the platform ignores it."""
    try:
        path.chmod(SECRET_FILE_MODE)
    except OSError:
        pass
 
 
def write_secret(
    env_var: str,
    value: str | None,
    *,
    path: str | Path = DEFAULT_SECRETS_PATH,
) -> Path:
    """Store *value* under *env_var* in the sidecar at *path*.
 
    Refuses any target git does not ignore (W1) BEFORE touching the disk,
    and refuses a name that is not an UPPER_SNAKE env-var name. Existing
    entries are preserved; the named one is replaced.
    """
    func = "write_secret"
    if not isinstance(env_var, str) or not _ENV_VAR_RE.match(env_var):
        raise ValueError(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Validator remediation message naming a field, matching the waived catalogue.py:497 text; high-entropy English, not a credential"
            f"provider-secrets key must be an UPPER_SNAKE env-var NAME, "
            f"found {env_var!r}"
        )
    if not value or not str(value).strip():
        raise ValueError(f"refusing to store an empty value for {env_var}")
 
    target = assert_gitignored(path)
    payload = load_secrets(target)
    payload[env_var] = str(value)
    _write_atomically(payload, target)
    _dbg("OC_CHAT.MAIN", func,
         f"stored {env_var} in the sidecar at {target} "
         f"({len(payload)} variable(s) held)", 2)
    return target
 
 
def apply_to_environ(
    *,
    path: str | Path = DEFAULT_SECRETS_PATH,
    environ: MutableMapping[str, str] | None = None,
) -> list[str]:
    """Load the sidecar into the environment; return the NAMES applied.
 
    A variable already carrying a non-blank value is left ALONE — a real
    environment variable always wins over the sidecar (P1). Never raises:
    a corrupt or unreadable sidecar degrades to "nothing applied" so a bad
    file cannot stop OC Chat from starting.
    """
    func = "apply_to_environ"
    env = os.environ if environ is None else environ
    try:
        stored = load_secrets(path)
    except SecretsFileError as exc:
        # EXTERNAL_ERROR (S6 L2) — the file is the operator's, not ours.
        _dbg("OC_CHAT.MAIN", func, f"EXTERNAL_ERROR: {exc}", 2, override=True)
        return []
 
    applied: list[str] = []
    for name, value in stored.items():
        if (env.get(name) or "").strip():
            continue  # a real environment variable always wins
        env[name] = value
        applied.append(name)
    if applied:
        _dbg("OC_CHAT.MAIN", func,
             f"applied {len(applied)} sidecar variable(s): {', '.join(applied)}", 2)
    return applied
 
 
def has_key(env_var: str, environ: Mapping[str, str] | None = None) -> bool:
    """Whether *env_var* currently carries a non-blank value (presence only)."""
    env = os.environ if environ is None else environ
    return bool((env.get(env_var) or "").strip())
 
 
__all__ = [
    "DEFAULT_SECRETS_PATH",
    "SECRET_FILE_MODE",
    "SecretsFileError",
    "SecretsPathError",
    "apply_to_environ",
    "assert_gitignored",
    "capture_key",
    "has_key",
    "is_gitignored",
    "load_secrets",
    "write_secret",
]