mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-28 20:44:50 +02:00
[eric] settings-agent: typed powering-credential resolver + schema-derived redactor + exhaustive no-suicide invariant test
This commit is contained in:
ciregenz
@@ -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
|
||||
@@ -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
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user