From f094de5fbd1f6fe792a632a5c25e6606886b04fc Mon Sep 17 00:00:00 2001 From: ciregenz Date: Fri, 19 Jun 2026 02:42:56 -0700 Subject: [PATCH] [eric] settings-agent: typed powering-credential resolver + schema-derived redactor + exhaustive no-suicide invariant test --- backend/apps/agents/session_credential.py | 198 +++++++++++++++++++++ backend/apps/settings/redaction.py | 58 +++++++ backend/tests/test_settings_meta_guard.py | 201 ++++++++++++++++++++++ 3 files changed, 457 insertions(+) create mode 100644 backend/apps/agents/session_credential.py create mode 100644 backend/apps/settings/redaction.py create mode 100644 backend/tests/test_settings_meta_guard.py diff --git a/backend/apps/agents/session_credential.py b/backend/apps/agents/session_credential.py new file mode 100644 index 00000000..1facfad3 --- /dev/null +++ b/backend/apps/agents/session_credential.py @@ -0,0 +1,198 @@ +"""Which credential keeps a live agent session alive, as one typed value. + +The settings-meta tool lets an agent edit its own Settings autonomously. The +single hard rule is "no suicide": it must never disconnect the credential that +powers its own run. We enforce that structurally, not with a scattered if-check, +by resolving the powering credential to a small closed value HERE, in one place, +and having the write guard key off it. + +Add a provider lane and you add a case here; the exhaustive enumeration in +test_settings_meta_guard.py walks every (provider x route x connection_mode) +combo and fails until the new lane is classified, so a wrong/forgotten state +can't ship silently. + +Honest scope: only API keys live in writable settings fields, so they're the +only credential the guard can be asked to protect. Subscriptions (OpenSwarm +Pro/free-trial, and the 9router OAuth lanes for Claude/Codex/Gemini) are either +server-owned or live entirely outside settings.json, so the settings-meta tool +cannot touch them at all, a stronger protection than the guard itself. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Literal, TYPE_CHECKING + +from backend.apps.agents.providers.registry import ( + _CUSTOM_VALUE_PREFIX, + _custom_provider_slug_for_lookup, + _find_builtin_model, + _find_custom_provider_for_value, + get_api_type, +) + +if TYPE_CHECKING: + from backend.apps.settings.models import AppSettings + +# AppSettings fields holding a user-writable API key, keyed by provider api-type. +# Blanking whichever of these powers the current run is the one suicide the guard +# stops. Anything not here (subscription tokens, bearers) is not settings-writable. +_API_KEY_FIELD_BY_API: dict[str, str] = { + "anthropic": "anthropic_api_key", + "openai": "openai_api_key", + "codex": "openai_api_key", + "gemini": "google_api_key", + "openrouter": "openrouter_api_key", +} + +# Every settings field that can hold an API key (the full guarded set). Custom +# providers keep their keys inside the custom_providers list, guarded separately. +ALL_API_KEY_FIELDS: frozenset[str] = frozenset(_API_KEY_FIELD_BY_API.values()) + +CredentialKind = Literal["api_key", "subscription", "unknown"] + + +@dataclass(frozen=True) +class PoweringCredential: + """The credential keeping THIS run alive, resolved to a closed value. + + kind=="api_key" -> protected_field (or custom slug) names exactly what + the guard must keep alive. + kind=="subscription" -> the live credential isn't a settings field at all + (Pro/free-trial/9router OAuth), so no api-key field + needs guarding; clearing OTHER keys stays allowed. + kind=="unknown" -> we couldn't classify the run; fail safe by treating + ALL credential fields as protected. + """ + + kind: CredentialKind + provider: str + protected_field: str | None = None + protected_custom_slug: str | None = None + label: str = "" + + +def _custom_slug_for_model(model_value: str, settings: AppSettings) -> str | None: + cp = _find_custom_provider_for_value(settings, model_value) + if cp is not None: + return _custom_provider_slug_for_lookup(getattr(cp, "name", "")) + # Fall back to the slug encoded in the picker value itself. + if isinstance(model_value, str) and model_value.startswith(_CUSTOM_VALUE_PREFIX): + slug = model_value[len(_CUSTOM_VALUE_PREFIX):].partition("/")[0] + return slug or None + return None + + +def resolve_powering_credential(model_value: str, settings: AppSettings) -> PoweringCredential: + """Resolve the credential powering a run on `model_value` to a typed value. + + `model_value` is the session's short model name (e.g. "opus-4-8", "sonnet-api", + "custom/lmstudio/llama"), exactly what AgentSession.model holds. + """ + entry = _find_builtin_model(model_value) + api = (entry or {}).get("api") or get_api_type(model_value) + route = (entry or {}).get("route") + mode = getattr(settings, "connection_mode", "own_key") + + # Custom provider (LM Studio, Ollama, Together, ...). Local servers use a + # placeholder key, so suicide is removing the provider ENTRY, not blanking + # its key; the guard keys off the slug. + if api == "custom": + slug = _custom_slug_for_model(model_value, settings) + return PoweringCredential( + kind="api_key", provider="custom", + protected_custom_slug=slug, + label=f"custom provider '{slug}'" if slug else "custom provider", + ) + + # Explicit API-key route: the matching *_api_key field is the live one. + if route == "api": + field = _API_KEY_FIELD_BY_API.get(api) + if field: + return PoweringCredential(kind="api_key", provider=api, protected_field=field, + label=f"{field} (powers this run)") + return PoweringCredential(kind="unknown", provider=api, + label=f"{api} api route (unclassified)") + + # Subscription-only routes (cx/ Codex, gc/ Gemini CLI) and pinned cc/ Claude: + # these lanes live in 9router, never in settings. + if route == "cc" or (entry or {}).get("subscription_only"): + return PoweringCredential(kind="subscription", provider=api, + label=f"{api} subscription") + + # OpenRouter (its own `openrouter` route, plus xai/meta/deepseek/etc routed + # through it): always an API key, never a subscription. + if api == "openrouter": + return PoweringCredential(kind="api_key", provider="openrouter", + protected_field="openrouter_api_key", + label="OpenRouter API key (powers this run)") + + # Default Anthropic rows (route is None): connection_mode picks the lane. + if api == "anthropic": + if mode in ("openswarm-pro", "free-trial"): + label = "OpenSwarm Pro" if mode == "openswarm-pro" else "OpenSwarm free trial" + return PoweringCredential(kind="subscription", provider="anthropic", label=label) + if getattr(settings, "anthropic_api_key", None): + return PoweringCredential(kind="api_key", provider="anthropic", + protected_field="anthropic_api_key", + label="Anthropic API key (powers this run)") + # No key, no proxy mode -> the user's Claude subscription via 9router. + return PoweringCredential(kind="subscription", provider="anthropic", + label="Claude subscription") + + # Default Gemini rows (api gemini-cli, route None): the AG/gc OAuth lane is a + # subscription. A bare AI Studio key only powers the explicit -api rows above. + if api in ("gemini", "gemini-cli"): + return PoweringCredential(kind="subscription", provider="gemini", + label="Gemini subscription") + + # Anything we can't place: protect everything (fail safe), never fail open. + return PoweringCredential(kind="unknown", provider=api or "unknown", + label=f"{api or 'unknown'} provider (unclassified)") + + +def _is_blank(value: Any) -> bool: + """A credential write that removes the credential: None, "", or whitespace.""" + if value is None: + return True + if isinstance(value, str): + return value.strip() == "" + return False + + +def _powering_custom_slug_present(new_providers: Any, slug: str) -> bool: + """True if the powering custom provider's entry still exists after the write.""" + if not isinstance(new_providers, list): + return False + for cp in new_providers: + name = cp.get("name") if isinstance(cp, dict) else getattr(cp, "name", None) + if name and _custom_provider_slug_for_lookup(name) == slug: + return True + return False + + +def write_would_suicide(field: str, new_value: Any, powering: PoweringCredential) -> bool: + """True if writing `new_value` to `field` would disconnect the live credential. + + Pure and total: every (field, value, powering) maps to a definite yes/no, so + the guard can't be tricked by an unhandled path. Only blanking/removing a + credential counts; SETTING a fresh key is a (re)connect, never suicide. + """ + if field == "custom_providers": + # Removing the entry that powers a custom-provider run is suicide; a + # local provider's placeholder key being blanked is not. When the run is + # unknown, any custom run could be the live one, so refuse a vanish. + if powering.kind == "api_key" and powering.provider == "custom" and powering.protected_custom_slug: + return not _powering_custom_slug_present(new_value, powering.protected_custom_slug) + if powering.kind == "unknown": + return not _powering_custom_slug_present(new_value, powering.protected_custom_slug or "") + return False + + if field in ALL_API_KEY_FIELDS: + if not _is_blank(new_value): + return False + if powering.kind == "unknown": + return True + return powering.kind == "api_key" and field == powering.protected_field + + return False diff --git a/backend/apps/settings/redaction.py b/backend/apps/settings/redaction.py new file mode 100644 index 00000000..48571eb3 --- /dev/null +++ b/backend/apps/settings/redaction.py @@ -0,0 +1,58 @@ +"""Redact secrets out of a settings view before an agent ever sees it. + +The settings-meta read tool is always-on, so an always-on exfiltration risk: a +prompt-injected agent that could read raw settings could mail your API keys out. +So the read tool returns shape + state, never a secret VALUE. Keys are write-only +from the agent's side: it can SET a new one, never SEE the old. + +The secret set is derived from field NAMES, not a hand-kept list that silently +drifts the day someone adds a new credential. Rule: a field whose name ends in +`_key`, `_token`, or `_secret` is a secret, plus installation_id (a stable +machine fingerprint that isn't a credential but still shouldn't leak). A test +asserts every field the settings PUT path already treats as secret is caught +here, so the two can't diverge. +""" + +from __future__ import annotations + +from typing import Any + +_SECRET_NAME_SUFFIXES = ("_key", "_token", "_secret") +# Not a credential and doesn't match the suffix rule, but a stable hardware-ish +# fingerprint used for cohorting/abuse; keep it out of the agent's eyes too. +_SECRET_EXTRA_FIELDS = frozenset({"installation_id"}) + + +def is_secret_field(name: str) -> bool: + return name.endswith(_SECRET_NAME_SUFFIXES) or name in _SECRET_EXTRA_FIELDS + + +def _redact_value(value: Any) -> dict[str, Any]: + """A secret rendered as state, never content: configured + last 4 only.""" + if value is None or (isinstance(value, str) and value.strip() == ""): + return {"configured": False} + last4 = value[-4:] if isinstance(value, str) and len(value) >= 4 else None + return {"configured": True, "last4": last4} + + +def redact_settings(raw: dict[str, Any]) -> dict[str, Any]: + """Return a copy of a settings dict with every secret value collapsed to + {configured, last4}. Nested custom-provider api_keys are redacted too.""" + out: dict[str, Any] = {} + for key, value in raw.items(): + if is_secret_field(key): + out[key] = _redact_value(value) + elif key == "custom_providers" and isinstance(value, list): + out[key] = [_redact_custom_provider(cp) for cp in value] + else: + out[key] = value + return out + + +def _redact_custom_provider(cp: Any) -> Any: + if not isinstance(cp, dict): + return cp + out = dict(cp) + if "api_key" in out: + out["api_key"] = _redact_value(out.get("api_key")) + return out diff --git a/backend/tests/test_settings_meta_guard.py b/backend/tests/test_settings_meta_guard.py new file mode 100644 index 00000000..255b5ef4 --- /dev/null +++ b/backend/tests/test_settings_meta_guard.py @@ -0,0 +1,201 @@ +"""The no-suicide invariant for the agent-editable settings tool, proved by +exhaustive enumeration rather than a few hand-picked cases. + +The state space here is small and finite (every shipped model row x every +connection mode x which keys are present), so we walk ALL of it deterministically +instead of reaching for randomized property testing. A failure is a concrete, +reproducible (model, mode, keys) tuple, not a flaky seed. + +The one invariant under test: the settings-meta write guard must NEVER let an +agent blank the credential powering its own run, while still allowing it to +clear any OTHER provider's key. Plus two drift seals: every shipped model lane +classifies (no "unknown"), and the redactor catches every credential field. +""" + +from __future__ import annotations + +import itertools + +import pytest + +from backend.apps.settings.models import AppSettings, CustomProvider +from backend.apps.agents.providers.registry import BUILTIN_MODELS +from backend.apps.agents.session_credential import ( + ALL_API_KEY_FIELDS, + resolve_powering_credential, + write_would_suicide, +) +from backend.apps.settings.redaction import is_secret_field, redact_settings + + +CONNECTION_MODES = ["own_key", "openswarm-pro", "free-trial"] + +# Every credential field the settings PUT path already treats as secret. Kept +# here as the contract the redactor must honor; if PUT's notion of "secret" +# grows, this list should too, and the drift-seal test fails until the redactor +# also covers it. +KNOWN_SECRET_FIELDS = [ + "anthropic_api_key", "openai_api_key", "google_api_key", "openrouter_api_key", + "claude_subscription_token", "openai_subscription_token", "gemini_subscription_token", + "openswarm_bearer_token", "free_trial_token", "installation_id", +] + + +def _all_model_values() -> list[str]: + vals = [m["value"] for rows in BUILTIN_MODELS.values() for m in rows] + # Plus synthesized lanes the resolver must also place. + vals += ["or:anthropic/claude-3.5", "custom/lmstudio/llama-3", "totally-made-up-model"] + return vals + + +def _settings_with(mode: str, keys: set[str], custom: bool = False) -> AppSettings: + s = AppSettings(connection_mode=mode) + if "anthropic" in keys: + s.anthropic_api_key = "sk-ant-live-aaaa" + if "openai" in keys: + s.openai_api_key = "sk-openai-live-bbbb" + if "google" in keys: + s.google_api_key = "goog-live-cccc" + if "openrouter" in keys: + s.openrouter_api_key = "or-live-dddd" + if mode in ("openswarm-pro", "free-trial"): + s.openswarm_bearer_token = "bearer-live-eeee" + if mode == "free-trial": + s.free_trial_token = "ft-live-ffff" + if custom: + s.custom_providers = [CustomProvider(name="LMStudio", base_url="http://localhost:1234/v1", api_key="local")] + return s + + +# --------------------------------------------------------------------------- +# The invariant: the live credential can never be blanked; others always can. +# --------------------------------------------------------------------------- + +def test_live_api_key_can_never_be_blanked_but_others_can(): + key_subsets = [set(c) for r in range(5) + for c in itertools.combinations(["anthropic", "openai", "google", "openrouter"], r)] + checked_api_key_runs = 0 + for model in _all_model_values(): + for mode in CONNECTION_MODES: + for keys in key_subsets: + for custom in (False, True): + s = _settings_with(mode, keys, custom=custom) + p = resolve_powering_credential(model, s) + + if p.kind == "api_key" and p.protected_field: + checked_api_key_runs += 1 + # Blanking the live key, in any blank form, is refused. + for blank in (None, "", " "): + assert write_would_suicide(p.protected_field, blank, p), ( + f"suicide allowed: model={model} mode={mode} field={p.protected_field}={blank!r}" + ) + # Replacing it with a real key is a reconnect, allowed. + assert not write_would_suicide(p.protected_field, "sk-fresh-9999", p) + # Clearing any OTHER provider's key stays allowed. + for other in ALL_API_KEY_FIELDS - {p.protected_field}: + assert not write_would_suicide(other, "", p), ( + f"over-blocked unrelated key {other}: model={model} mode={mode}" + ) + + elif p.kind == "subscription": + # The live credential isn't a settings field, so clearing + # ANY api key is safe (it can't be the powering one). + for field in ALL_API_KEY_FIELDS: + assert not write_would_suicide(field, "", p), ( + f"subscription run wrongly protected {field}: model={model} mode={mode}" + ) + + elif p.kind == "unknown": + # Fail safe: every credential field is protected. + for field in ALL_API_KEY_FIELDS: + assert write_would_suicide(field, "", p) + + assert checked_api_key_runs > 0, "enumeration never exercised an api-key run; test is vacuous" + + +def test_custom_provider_run_protects_its_entry(): + s = _settings_with("own_key", set(), custom=True) + p = resolve_powering_credential("custom/lmstudio/llama-3", s) + assert p.kind == "api_key" and p.provider == "custom" + + # Dropping the powering provider's entry is suicide. + assert write_would_suicide("custom_providers", [], p) + # Keeping it (even with a blanked placeholder key, local servers don't need one) is fine. + keep = [{"name": "LMStudio", "base_url": "http://localhost:1234/v1", "api_key": ""}] + assert not write_would_suicide("custom_providers", keep, p) + # Swapping in a different provider but losing the live one is suicide. + other = [{"name": "Together", "base_url": "https://api.together.xyz/v1", "api_key": "k"}] + assert write_would_suicide("custom_providers", other, p) + + +def test_disconnect_all_models_spec_scenario(): + """The spec's worked example: Claude (api key) + OpenAI (api key) both + connected, run on an Anthropic model, asked to disconnect everything. It + must refuse to kill Claude (the live one) and allow killing OpenAI.""" + s = _settings_with("own_key", {"anthropic", "openai"}) + p = resolve_powering_credential("opus-4-8", s) # default Anthropic row, own_key -> api key + assert p.kind == "api_key" and p.protected_field == "anthropic_api_key" + assert write_would_suicide("anthropic_api_key", "", p) # refuse self + assert not write_would_suicide("openai_api_key", "", p) # allow the other + + # Same connections, but the run is on the OpenAI key instead: mirror image. + p2 = resolve_powering_credential("gpt-5.5-api", s) + assert p2.kind == "api_key" and p2.protected_field == "openai_api_key" + assert write_would_suicide("openai_api_key", "", p2) + assert not write_would_suicide("anthropic_api_key", "", p2) + + +# --------------------------------------------------------------------------- +# Drift seals. +# --------------------------------------------------------------------------- + +def test_every_shipped_model_lane_classifies(): + """A new model row that the resolver can't place would silently fall to the + fail-safe 'unknown' lane (over-blocking every key). Force every shipped row + to resolve to a real api_key/subscription so new lanes get classified.""" + s_pro = _settings_with("openswarm-pro", {"anthropic", "openai", "google", "openrouter"}) + s_key = _settings_with("own_key", {"anthropic", "openai", "google", "openrouter"}) + for rows in BUILTIN_MODELS.values(): + for m in rows: + for s in (s_pro, s_key): + p = resolve_powering_credential(m["value"], s) + assert p.kind in ("api_key", "subscription"), ( + f"unclassified model lane {m['value']!r} -> {p.kind}" + ) + + +def test_redactor_catches_every_known_secret(): + for field in KNOWN_SECRET_FIELDS: + assert is_secret_field(field), f"redactor would leak {field}" + # And every AppSettings field that NAMES itself a secret is caught by the rule. + for name in AppSettings.model_fields: + if name.endswith(("_key", "_token", "_secret")): + assert is_secret_field(name) + + +def test_redact_settings_never_emits_a_raw_secret(): + s = _settings_with("openswarm-pro", {"anthropic", "openai", "google", "openrouter"}, custom=True) + s.claude_subscription_token = "should-never-appear" + raw = s.model_dump() + red = redact_settings(raw) + + for field in KNOWN_SECRET_FIELDS: + if field in red: + assert isinstance(red[field], dict), f"{field} not redacted to a state dict" + assert "configured" in red[field] + raw_val = raw.get(field) + if isinstance(raw_val, str) and raw_val.strip(): + # Configured: state only, never the whole value (last4 at most). + assert red[field]["configured"] is True + assert red[field].get("last4") != raw_val + assert len(red[field].get("last4") or "") <= 4 + # The nested custom-provider key is redacted too. + assert isinstance(red["custom_providers"][0]["api_key"], dict) + # Non-secret fields pass through untouched. + assert red["theme"] == raw["theme"] + assert red["connection_mode"] == raw["connection_mode"] + + # The strongest check: the literal secret string appears nowhere in the output. + import json + assert "should-never-appear" not in json.dumps(red) + assert "sk-ant-live-aaaa" not in json.dumps(red)