[eric] settings-agent: shared secret-scanner (backend/common) + fail-safe value-shape redaction + second-wall credential restore + unified settings write lock

backend/apps/settings/redaction.py

@@ -17,6 +17,8 @@ from __future__ import annotations
 
 from typing import Any
 
+from backend.common.secret_scan import looks_secret
+
 _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.
@@ -27,6 +29,14 @@ def is_secret_field(name: str) -> bool:
     return name.endswith(_SECRET_NAME_SUFFIXES) or name in _SECRET_EXTRA_FIELDS
 
 
+def _value_is_secret_shaped(value: Any) -> bool:
+    """Fail-safe behind the name rule: a field the name rule misses (a future
+    secret with an off-convention name) is still caught if its VALUE looks like
+    a credential (sk-..., ghp_..., Bearer ...). So a leak needs BOTH a bad name
+    AND a non-credential-shaped value, not just one."""
+    return isinstance(value, str) and looks_secret(value)
+
+
 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() == ""):
@@ -40,7 +50,7 @@ def redact_settings(raw: dict[str, Any]) -> dict[str, Any]:
     {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):
+        if is_secret_field(key) or _value_is_secret_shaped(value):
             out[key] = _redact_value(value)
         elif key == "custom_providers" and isinstance(value, list):
             out[key] = [_redact_custom_provider(cp) for cp in value]

backend/apps/settings/settings.py

@@ -146,24 +146,46 @@ SERVER_OWNED_FIELDS = (
 )
 
 
+# One serialization point for EVERY settings write (renderer PUT + agent tool),
+# so a renderer save and an autonomous agent edit can't interleave and clobber
+# each other mid read-modify-write. Callers hold it across read->build->save;
+# apply_settings_update itself does NOT acquire it (would deadlock the agent path
+# that reads under the same lock), so every caller must wrap apply in it.
+settings_write_lock = asyncio.Lock()
+
+
 @settings.router.put("")
 async def update_settings(body: AppSettings):
-    saved = await apply_settings_update(body)
+    async with settings_write_lock:
+        saved = await apply_settings_update(body)
     return {"ok": True, "settings": saved.model_dump()}
 
 
-async def apply_settings_update(body: AppSettings) -> AppSettings:
+async def apply_settings_update(body: AppSettings, protect_fields: set[str] | None = None) -> AppSettings:
     """Persist a full settings object with all the safety side effects: restore
     server-owned fields, hand the wheel back from the free trial when a real
     model is connected, reconcile 9router provider connections, and sync
     analytics/identity. The PUT route and the agent settings tool both call this
-    so the write semantics can't drift between them. Returns the saved body."""
+    so the write semantics can't drift between them. Returns the saved body.
+
+    Caller must hold settings_write_lock. `protect_fields` names credential fields
+    that must never be blanked by this write (the agent tool passes the field
+    powering the live run): a SECOND, independent wall behind the endpoint's
+    suicide-guard, so a guard bug still can't disconnect a run."""
     from backend.apps.service.client import sync as _sync
 
     old = load_settings()
     for k in SERVER_OWNED_FIELDS:
         setattr(body, k, getattr(old, k, None))
 
+    # Second wall: if a write tries to clear a credential that's currently set and
+    # flagged as powering this run, restore it (like server-owned fields). The
+    # endpoint guard already strips these; this is the backstop that can't be
+    # bypassed by a logic slip upstream.
+    for f in (protect_fields or ()):
+        if getattr(old, f, None) and not getattr(body, f, None):
+            setattr(body, f, getattr(old, f, None))
+
     # If the user connects their own model while the free trial is armed, hand
     # the wheel back to their provider. Without this, connection_mode (server-
     # owned, so the loop above just restored it to "free-trial") would keep them

backend/apps/skill_registry/skill_registry.py

@@ -404,7 +404,7 @@ async def resolve_community_skill(source: str, skill_id: str) -> dict:
     # Reuse the .swarm importer's content scan: flag files holding secret-shaped
     # literals (the author's leaked key, or a sketchy skill) so the user sees it
     # before installing from an unvetted repo.
-    from backend.apps.swarm.redact import find_secrets_in_files
+    from backend.common.secret_scan import find_secrets_in_files
     secret_findings = find_secrets_in_files({rel: data.encode("utf-8", "ignore") for rel, data in files.items()})
     return {
         "name": meta.get("name") or skill_id,

backend/apps/swarm/redact.py

@@ -23,15 +23,13 @@ _DENY_EXACT = {
     "credentials", "sdk_session_id",
 }
 
-REDACTED = "[redacted]"
+# The secret-shape scanner moved to backend.common so skills + settings reuse it
-
+# without reaching into swarm; re-exported here so ziputil/closure keep their API.
-# Literal-secret shapes someone might paste into a file or skill body.
+from backend.common.secret_scan import (  # noqa: E402
-_CONTENT_PATTERNS = (
+    REDACTED,
-    re.compile(r"sk-ant-[A-Za-z0-9_\-]{16,}"),
+    find_secrets_in_files,
-    re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
+    looks_secret as _looks_secret,
-    re.compile(r"AIza[A-Za-z0-9_\-]{20,}"),          # Google API key shape
+    redact_secret_shapes as scrub_text,
-    re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"),       # GitHub tokens
-    re.compile(r"Bearer\s+[A-Za-z0-9._\-]{16,}"),
 )
 
 
@@ -42,12 +40,6 @@ def is_denied_key(key: str) -> bool:
     return any(sub in k for sub in _DENY_SUBSTRINGS)
 
 
-def scrub_text(text: str) -> str:
-    for pat in _CONTENT_PATTERNS:
-        text = pat.sub(REDACTED, text)
-    return text
-
-
 def scrub_payload(value: Any) -> Any:
     """Recursively drop denied keys and redact secret-shaped strings in a
     JSON-able structure. Returns a new structure; never mutates the input."""
@@ -81,19 +73,5 @@ def find_denied_keys(value: Any, _path: str = "") -> list[str]:
     return found
 
 
-def _looks_secret(text: str) -> bool:
+# _looks_secret + find_secrets_in_files now come from backend.common.secret_scan
-    return any(pat.search(text) for pat in _CONTENT_PATTERNS)
+# (imported at the top); kept re-exported so ziputil's audit import is unchanged.
-
-
-def find_secrets_in_files(files: dict[str, bytes]) -> list[str]:
-    """Paths of any file whose text body holds a secret-shaped literal. Payloads
-    get scrubbed key-and-content, but raw workspace files (an app's source) were
-    only key-scanned, so a key hardcoded in a .js would slip. Binary files are
-    skipped (a null byte means it isn't text someone pasted a token into)."""
-    hits: list[str] = []
-    for path, data in files.items():
-        if b"\x00" in data[:4096]:
-            continue
-        if _looks_secret(data.decode("utf-8", errors="ignore")):
-            hits.append(path)
-    return hits

backend/common/secret_scan.py

@@ -0,0 +1,46 @@
+"""Shared secret-shape scanner: spot credential-shaped literals in text/files.
+
+Lives in backend.common so the .swarm importer, the skills registry, and the
+settings redactor all pull it DOWN from one place instead of one feature app
+reaching sideways into another. It catches a secret by its SHAPE (sk-ant-...,
+ghp_..., AIza...), which is the fail-safe behind name-based redaction: a key
+that's misnamed (so a name rule misses it) still gets caught by its shape."""
+
+from __future__ import annotations
+
+import re
+
+REDACTED = "[redacted]"
+
+# Literal-secret shapes someone might paste into a file, skill body, or setting.
+SECRET_SHAPE_PATTERNS = (
+    re.compile(r"sk-ant-[A-Za-z0-9_\-]{16,}"),
+    re.compile(r"sk-[A-Za-z0-9_\-]{16,}"),
+    re.compile(r"AIza[A-Za-z0-9_\-]{20,}"),          # Google API key shape
+    re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"),       # GitHub tokens
+    re.compile(r"Bearer\s+[A-Za-z0-9._\-]{16,}"),
+)
+
+
+def looks_secret(text: str) -> bool:
+    """True if `text` contains a credential-shaped literal."""
+    return any(p.search(text) for p in SECRET_SHAPE_PATTERNS)
+
+
+def redact_secret_shapes(text: str) -> str:
+    """Replace every secret-shaped literal in `text` with the redacted marker."""
+    for p in SECRET_SHAPE_PATTERNS:
+        text = p.sub(REDACTED, text)
+    return text
+
+
+def find_secrets_in_files(files: dict[str, bytes]) -> list[str]:
+    """Paths of any file whose text body holds a secret-shaped literal. Binary
+    files (a null byte in the first 4KB) are skipped, they aren't pasted text."""
+    hits: list[str] = []
+    for path, data in files.items():
+        if b"\x00" in data[:4096]:
+            continue
+        if looks_secret(data.decode("utf-8", errors="ignore")):
+            hits.append(path)
+    return hits

backend/main.py

@@ -742,11 +742,6 @@ async def mcp_meta(action: str, request: Request):
     return JSONResponse({"error": f"unknown action: {action}"}, status_code=400)
 
 
-# Serializes agent-side SettingsWrite read-modify-writes so concurrent autonomous
-# agents can't clobber each other's edits (see the lock's use below).
-_settings_meta_write_lock = asyncio.Lock()
-
-
 @app.post("/api/settings-meta/{action}")
 async def settings_meta(action: str, request: Request):
     """Back the openswarm-settings-meta stdio MCP server (agent-editable Settings).
@@ -766,9 +761,9 @@ async def settings_meta(action: str, request: Request):
     from backend.apps.settings.store import load_settings
     from backend.apps.settings.models import AppSettings
     from backend.apps.settings.redaction import redact_settings
-    from backend.apps.settings.settings import SERVER_OWNED_FIELDS, apply_settings_update
+    from backend.apps.settings.settings import SERVER_OWNED_FIELDS, apply_settings_update, settings_write_lock
     from backend.apps.agents.session_credential import (
-        PoweringCredential, resolve_powering_credential, write_would_suicide,
+        ALL_API_KEY_FIELDS, PoweringCredential, resolve_powering_credential, write_would_suicide,
     )
     from backend.apps.agents.agent_manager import agent_manager
     from pydantic import ValidationError
@@ -791,7 +786,7 @@ async def settings_meta(action: str, request: Request):
         # other's fields while BOTH got an "applied" result). The lock makes agent
         # writes serial so the last load always sees the prior write. (Agent vs the
         # renderer's own PUT stays the pre-existing full-object-replace race.)
-        async with _settings_meta_write_lock:
+        async with settings_write_lock:
             settings = load_settings()
             session = agent_manager.sessions.get(parent_session_id) if parent_session_id else None
             if session is not None:
@@ -800,6 +795,15 @@ async def settings_meta(action: str, request: Request):
                 # No live session to anchor the guard: fail safe, protect every credential.
                 powering = PoweringCredential(kind="unknown", provider="unknown", label="this run")
 
+            # The credential field(s) the second-wall restore in apply_settings_update
+            # must never let a write blank (independent of the per-field guard below).
+            if powering.kind == "unknown":
+                protect_fields = set(ALL_API_KEY_FIELDS)
+            elif powering.kind == "api_key" and powering.protected_field:
+                protect_fields = {powering.protected_field}
+            else:
+                protect_fields = set()
+
             staged: dict = {}
             for field, value in changes.items():
                 if field not in valid_fields:
@@ -828,7 +832,7 @@ async def settings_meta(action: str, request: Request):
                         new_body = AppSettings(**merged)
                 if staged and new_body is not None:
                     try:
-                        await apply_settings_update(new_body)
+                        await apply_settings_update(new_body, protect_fields=protect_fields)
                         for f in staged:
                             outcomes[f] = {"status": "applied"}
                     except Exception as e:

backend/tests/test_settings_meta_endpoint.py

@@ -14,6 +14,36 @@ from fastapi.testclient import TestClient
 from backend.main import app
 
 
+@pytest.mark.asyncio
+async def test_second_wall_restores_protected_credential_even_if_body_blanks_it():
+    """Defense in depth: even if a write reaches apply_settings_update with the
+    live credential blanked (a guard slip upstream), the second-wall restore puts
+    it back. Proves the api-key guard isn't a single point of failure."""
+    from backend.apps.settings.settings import (
+        apply_settings_update, settings_write_lock, load_settings, _save_settings,
+    )
+    original = load_settings().model_copy(deep=True)
+    try:
+        s = load_settings()
+        s.anthropic_api_key = "sk-live-KEEP-ME"
+        _save_settings(s)
+        # A body that (as if a guard bug let it through) clears the live key.
+        body = load_settings()
+        body.anthropic_api_key = ""
+        async with settings_write_lock:
+            saved = await apply_settings_update(body, protect_fields={"anthropic_api_key"})
+        assert saved.anthropic_api_key == "sk-live-KEEP-ME", "second wall failed to restore"
+        assert load_settings().anthropic_api_key == "sk-live-KEEP-ME"
+        # And a NON-protected blank still goes through (only the protected one is restored).
+        body2 = load_settings()
+        body2.openai_api_key = ""
+        async with settings_write_lock:
+            await apply_settings_update(body2, protect_fields={"anthropic_api_key"})
+        assert not load_settings().openai_api_key
+    finally:
+        _save_settings(original)
+
+
 @pytest.fixture
 def client():
     import backend.auth as auth_mod

backend/tests/test_settings_meta_guard.py

@@ -173,6 +173,18 @@ def test_redactor_catches_every_known_secret():
             assert is_secret_field(name)
 
 
+def test_redaction_fail_safe_catches_misnamed_secret_by_value():
+    # The name rule (_key/_token/_secret) would MISS a field named off-convention.
+    # The value-shape backstop must still redact it, so a leak needs BOTH a bad
+    # name AND a non-credential-shaped value, not just one.
+    import json
+    raw = {"theme": "dark", "weird_field": "sk-ant-api03-AAAABBBBCCCCDDDDEEEEFFFF"}
+    red = redact_settings(raw)
+    assert red["theme"] == "dark"
+    assert isinstance(red["weird_field"], dict) and red["weird_field"]["configured"] is True
+    assert "sk-ant-api03" not in json.dumps(red)
+
+
 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"