feat(vault): bridge-level secret interception + vault CLI skill

CLAUDE.md

@@ -223,6 +223,28 @@ When the user says "tutorial", "walk me through", or "show me what you can do" (
 
 Keep each step conversational and brief — this is spoken, not read. Focus on what to say/try, skip setup details unless asked.
 
+## Vault — secure secret storage
+
+Secrets passed via Slack/Discord (`vault set KEY VALUE`) are intercepted by the bridge and stored in macOS Keychain. They never touch a file on disk.
+
+**When writing any integration that needs an API key, token, or password — always use vault:**
+
+```python
+from vault_intercept import get_vault_key, list_vault_keys
+
+keys = list_vault_keys()  # returns list of stored key names
+api_key = get_vault_key("OPENAI_API_KEY")  # raises KeyError if not found
+```
+
+**CLI (for subprocesses):**
+```bash
+python3 skills/vault/vault.py list                           # list stored key names
+python3 skills/vault/vault.py get KEY                        # print value
+python3 skills/vault/vault.py env KEY1 KEY2 -- python3 x.py  # inject as env vars
+```
+
+If an integration needs a key that isn't in the vault yet, ask the user to send `vault set KEY value` via Slack or Discord — the bridge intercepts it securely before it touches disk.
+
 ## Built-in tools
 
 For the per-tool bash recipes (Calendar, Screen capture, Notes, Email, Contacts, iMessage, WhatsApp, X, Reminders, macOS GUI control, Browser automation, File search, Meeting join, Phone calls, App launcher, Context drop + shortcuts), see [`docs/built-in-tools.md`](docs/built-in-tools.md). Moved out of CLAUDE.md to keep the per-session context budget tight — reach for it on demand rather than carrying it on every turn.

skills/vault/vault.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""vault — manage secrets stored in macOS Keychain via Sutando's vault.
+
+Subcommands:
+  list                    Show all stored key names (no values)
+  get KEY                 Print the value for KEY
+  env KEY [KEY...] -- CMD Run CMD with vault keys injected as environment variables
+
+Examples:
+  vault.py list
+  vault.py get OPENAI_API_KEY
+  vault.py env OPENAI_API_KEY STRIPE_KEY -- python3 my_script.py
+"""
+
+import os
+import subprocess
+import sys
+
+# Allow running from any directory by adding src/ to path
+_SRC = os.path.join(os.path.dirname(__file__), "..", "..", "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from vault_intercept import get_vault_key, list_vault_keys
+
+
+def cmd_list() -> None:
+    keys = list_vault_keys()
+    if not keys:
+        print("(no keys stored)")
+        return
+    for k in keys:
+        print(k)
+
+
+def cmd_get(key: str) -> None:
+    try:
+        print(get_vault_key(key))
+    except KeyError as e:
+        print(str(e), file=sys.stderr)
+        sys.exit(1)
+
+
+def cmd_env(keys: list[str], cmd: list[str]) -> None:
+    if not cmd:
+        print("vault env: missing command after --", file=sys.stderr)
+        sys.exit(1)
+    env = os.environ.copy()
+    for k in keys:
+        try:
+            env[k] = get_vault_key(k)
+        except KeyError:
+            print(f"vault env: key '{k}' not found — aborting", file=sys.stderr)
+            sys.exit(1)
+    result = subprocess.run(cmd, env=env)
+    sys.exit(result.returncode)
+
+
+def main() -> None:
+    args = sys.argv[1:]
+    if not args:
+        print(__doc__)
+        sys.exit(0)
+
+    sub = args[0]
+
+    if sub == "list":
+        cmd_list()
+
+    elif sub == "get":
+        if len(args) < 2:
+            print("vault get: missing KEY", file=sys.stderr)
+            sys.exit(1)
+        cmd_get(args[1])
+
+    elif sub == "env":
+        rest = args[1:]
+        try:
+            sep = rest.index("--")
+        except ValueError:
+            print("vault env: missing -- separator before command", file=sys.stderr)
+            sys.exit(1)
+        cmd_env(rest[:sep], rest[sep + 1:])
+
+    else:
+        print(f"vault: unknown subcommand '{sub}'", file=sys.stderr)
+        print("Usage: vault.py list | get KEY | env KEY... -- CMD", file=sys.stderr)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

src/discord-bridge.py

@@ -56,6 +56,7 @@
 from workspace_default import resolve_workspace  # noqa: E402
 from util_paths import shared_personal_path  # noqa: E402
 from task_priority import default_priority_for_source  # noqa: E402
+from vault_intercept import intercept_vault_commands, redact_vault_commands  # noqa: E402
 REPO = resolve_workspace()
 
 # Lazy import for cloud telemetry — guarded so a missing cloud_metrics.py
@@ -2482,6 +2483,21 @@ async def _handle_discord_message(message, force=False):
     task_id = f"task-{ts}"
     task_file = TASKS_DIR / f"{task_id}.txt"
 
+    # Intercept vault commands before any disk write.
+    # Owner-tier only: secrets go to Keychain, task file gets [STORED-IN-KEYCHAIN].
+    # Non-owner: vault patterns are redacted to prevent Keychain pollution by
+    # untrusted senders — the actual secret never reaches the task file either way.
+    if text:
+        if access_tier == "owner":
+            vault_result = intercept_vault_commands(text)
+            text = vault_result.text
+            if vault_result.stored:
+                print(f"  [vault] stored keys: {vault_result.stored}", flush=True)
+            if vault_result.failed:
+                print(f"  [vault] store failed (still redacted): {vault_result.failed}", flush=True)
+        else:
+            text = redact_vault_commands(text)
+
     # Inject tier-specific in-band instructions so the core agent cannot
     # accidentally process a non-owner task with full capabilities.
     # See CLAUDE.md "Discord access control" section for the policy.

src/slack-bridge.py

@@ -47,6 +47,7 @@
 from task_priority import default_priority_for_source  # noqa: E402
 from result_markers import parse_markers  # noqa: E402
 from workspace_default import resolve_workspace  # noqa: E402
+from vault_intercept import intercept_vault_commands, redact_vault_commands  # noqa: E402
 
 try:
     from slack_bolt import App
@@ -358,6 +359,21 @@ def _write_task(event: dict, prefix: str, text: str, username: str | None) -> st
         # than treating as owner.
         access_tier = "other"
 
+    # Intercept vault commands before any disk write — must happen AFTER
+    # access_tier is resolved so untrusted senders cannot write to Keychain.
+    # Owner-tier: secrets go to Keychain, task file gets [STORED-IN-KEYCHAIN].
+    # Non-owner: patterns redacted, Keychain untouched.
+    if text:
+        if access_tier == "owner":
+            vault_result = intercept_vault_commands(text)
+            text = vault_result.text
+            if vault_result.stored:
+                print(f"  [vault] stored keys: {vault_result.stored}", flush=True)
+            if vault_result.failed:
+                print(f"  [vault] store failed (still redacted): {vault_result.failed}", flush=True)
+        else:
+            text = redact_vault_commands(text)
+
     # Prepend an in-band system instruction for non-owner tiers so the
     # core agent cannot accidentally process a downgraded task with full
     # capabilities. Mirrors the Discord bridge's tier-specific instruction

src/telegram-bridge.py

@@ -29,6 +29,7 @@ def _push_vision_image(path: str, source: str = "telegram") -> bool:  # type: ig
 from result_markers import parse_markers  # noqa: E402
 
 from workspace_default import resolve_workspace  # noqa: E402
+from vault_intercept import intercept_vault_commands  # noqa: E402
 REPO = resolve_workspace()
 TASKS_DIR = REPO / "tasks"
 RESULTS_DIR = REPO / "results"
@@ -581,6 +582,17 @@ def main():
                 task_id = f"task-{ts}"
                 task_file = TASKS_DIR / f"{task_id}.txt"
                 priority = default_priority_for_source("telegram", "owner")
+
+                # Intercept vault commands before disk write — Telegram treats
+                # all senders as owner-tier (allowlist-gated bot token).
+                if text:
+                    vault_result = intercept_vault_commands(text)
+                    text = vault_result.text
+                    if vault_result.stored:
+                        print(f"  [vault] stored keys: {vault_result.stored}", flush=True)
+                    if vault_result.failed:
+                        print(f"  [vault] store failed (still redacted): {vault_result.failed}", flush=True)
+
                 task_file.write_text(
                     f"id: {task_id}\n"
                     f"timestamp: {time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())}\n"

src/vault_intercept.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+"""Bridge-level vault secret interception.
+
+Detects `vault set KEY VALUE` patterns in incoming messages BEFORE they
+are written to task files on disk. Secrets go straight to macOS Keychain
+(encrypted at rest) and the task file receives `[STORED-IN-KEYCHAIN]`
+as a placeholder.
+
+Secret lifecycle:
+  Slack/Discord API → bridge (in-memory, SSL in transit)
+  → Keychain (encrypted) — disk never sees plaintext.
+
+Usage (in any bridge's message handler):
+
+    from vault_intercept import intercept_vault_commands
+
+    result = intercept_vault_commands(raw_message)
+    # result.text  — sanitized, safe to write to disk (plaintext always gone)
+    # result.stored — keys successfully stored to Keychain
+    # result.failed — keys that failed to store (still redacted from text)
+
+Supported syntax (all case-insensitive):
+    vault set KEY value
+    vault set KEY "value with spaces"
+    vault set KEY 'value with spaces'
+    vault set KEY `value`         (backtick-quoted — backticks stripped)
+    vault set KEY `value with spaces`  (backtick-quoted with spaces)
+
+Multiple commands in one message are all intercepted in a single pass.
+
+`redact_vault_commands(text)` is the non-storing variant: it scrubs vault-set
+patterns from text without touching the Keychain.  Use it for non-owner-tier
+messages where we want to prevent accidental secret exposure in task files but
+must not write to the owner's Keychain.
+"""
+
+import json
+import os
+import re
+import subprocess
+from datetime import datetime, timezone
+from typing import NamedTuple
+
+_ACCOUNT = "sutando"
+_MANIFEST_PATH = os.path.expanduser("~/.sutando-vault/keys.json")
+
+# Matches: vault set KEY <value>  where value is:
+#   - double-quoted string   "foo bar"
+#   - single-quoted string   'foo bar'
+#   - backtick-quoted string `foo bar`  (Discord markdown; backticks stripped)
+#   - bare token (no spaces) foobar
+_VAULT_SET_RE = re.compile(
+    r'vault\s+set\s+(\S+)\s+(?:"([^"]*)"|\'([^\']*)\'|`([^`]*)`|(\S+))',
+    re.IGNORECASE,
+)
+
+
+class InterceptResult(NamedTuple):
+    text: str          # sanitized message text, safe to write to disk
+    stored: list[str]  # keys successfully stored to Keychain
+    failed: list[str]  # keys that could NOT be stored (secret still redacted)
+
+
+def _store_in_keychain(key: str, value: str) -> None:
+    # Note: value is passed as an argv element — briefly visible in `ps` to
+    # the same user. Acceptable on a single-user Mac; not a multi-user safe API.
+    result = subprocess.run(
+        [
+            "security", "add-generic-password",
+            "-a", _ACCOUNT,
+            "-s", key,
+            "-w", value,
+            "-U",   # update if already exists
+        ],
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"vault: failed to store '{key}': "
+            f"{result.stderr.decode(errors='replace').strip()}"
+        )
+    _register_key(key)
+
+
+def _register_key(key: str) -> None:
+    os.makedirs(os.path.dirname(_MANIFEST_PATH), exist_ok=True)
+    try:
+        with open(_MANIFEST_PATH) as f:
+            manifest = json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        manifest = {}
+    manifest[key] = {"stored_at": datetime.now(timezone.utc).isoformat()}
+    # Atomic write — concurrent bridge processes won't corrupt keys.json.
+    tmp = _MANIFEST_PATH + ".tmp"
+    with open(tmp, "w") as f:
+        json.dump(manifest, f, indent=2)
+    os.replace(tmp, _MANIFEST_PATH)
+
+
+def list_vault_keys() -> list[str]:
+    """Return all key names stored in the vault manifest (no values)."""
+    try:
+        with open(_MANIFEST_PATH) as f:
+            return sorted(json.load(f).keys())
+    except (FileNotFoundError, json.JSONDecodeError):
+        return []
+
+
+def get_vault_key(key: str) -> str:
+    """Retrieve a secret value from Keychain. Raises KeyError if not found."""
+    result = subprocess.run(
+        ["security", "find-generic-password", "-a", _ACCOUNT, "-s", key, "-w"],
+        capture_output=True,
+    )
+    if result.returncode != 0:
+        raise KeyError(f"vault: key '{key}' not found in Keychain")
+    return result.stdout.decode().strip()
+
+
+def intercept_vault_commands(text: str) -> InterceptResult:
+    """Detect vault-set commands in `text`, store secrets, return sanitized text.
+
+    Fail-closed: the plaintext secret is ALWAYS redacted from the returned text,
+    even when the Keychain write fails. Failed keys are reported in result.failed
+    so the bridge can notify the user without leaking the secret.
+    """
+    if not text:
+        return InterceptResult(text=text, stored=[], failed=[])
+
+    stored: list[str] = []
+    failed: list[str] = []
+
+    def _replacer(m: re.Match) -> str:
+        key = m.group(1)
+        # Groups 2/3/4/5: double-quoted / single-quoted / backtick / bare token.
+        value = next(
+            (g for g in (m.group(2), m.group(3), m.group(4), m.group(5)) if g is not None),
+            "",
+        )
+        if not value:
+            # Reject empty value — ambiguous and almost certainly a mistake.
+            failed.append(key)
+            return f"vault set {key} [VAULT-EMPTY-VALUE]"
+        try:
+            _store_in_keychain(key, value)
+            stored.append(key)
+            return f"vault set {key} [STORED-IN-KEYCHAIN]"
+        except RuntimeError:
+            # Store failed — redact anyway so plaintext never reaches disk.
+            failed.append(key)
+            return f"vault set {key} [VAULT-STORE-FAILED]"
+
+    sanitized = _VAULT_SET_RE.sub(_replacer, text)
+    return InterceptResult(text=sanitized, stored=stored, failed=failed)
+
+
+def redact_vault_commands(text: str) -> str:
+    """Scrub vault-set patterns from text WITHOUT touching the Keychain.
+
+    Use for non-owner-tier messages: prevents secrets from landing in task files
+    while ensuring the Keychain is never written by an untrusted sender.
+    """
+    if not text:
+        return text
+    return _VAULT_SET_RE.sub(
+        lambda m: f"vault set {m.group(1)} [vault: non-owner tier — ignored]",
+        text,
+    )

tests/slack-bridge-access.test.py

@@ -81,7 +81,7 @@ def main() -> int:
     # (the `user_id not in allowed` gate); the budget only needs to be
     # large enough to terminate at the next `\ndef ` boundary.
     write_match = re.search(
-        r"def _write_task\([^)]*\)[^:]*:\s*\n([\s\S]{0,6000}?)(?=\n\ndef |\Z)",
+        r"def _write_task\([^)]*\)[^:]*:\s*\n([\s\S]{0,8000}?)(?=\n\ndef |\Z)",
         src,
     )
     if not write_match: