[Phase 1] Leak Detection: Scan outbound messages for accidental secret exposure [ON HOLD - needs avault integration] #145

Open
opened 2026-02-27 13:49:05 +00:00 by noopsec · 2 comments
Contributor

Status: ON HOLD ⚠️

Waiting for integration with avault as central secret store.

As noted by k9ert: "I think it doesn't make so much sense to implement leak-detection without a central store of secrets which you can use to create the regexes to detect the leakage."

Solution: Use avault - NIP-44 encrypted agent vault with NIP-46 remote signing.


Problem

Issue #50 describes the larger problem: Secrets are visible via Environment Variables to all plugins and subprocesses. As Phase 1 (quick win), we propose Leak Detection — a safety net that catches accidental secret leaks before they leave the agent.

Goal

Scan all outbound messages (Telegram, Nostr, API calls) on secret patterns and block/redact if found.

Central Secret Store: avault

avault provides:

  • NIP-44 encryption - XChaCha20-Poly1305 for vault files
  • NIP-46 remote signing - Phone (Amber) approves decryption
  • RAM-only secrets - Never touch disk unencrypted
  • Unix socket API - Daemon/CLI communication

avault Integration for Cobot

# cobot/plugins/vault/plugin.py
class VaultPlugin(Plugin):
    """avault integration for secret management."""
    
    meta = PluginMeta(
        id="vault",
        version="1.0.0",
        capabilities=["secrets"],
        extension_points=[
            "vault.secrets",
            "vault.secret_patterns"
        ]
    )
    
    def __init__(self):
        self._avault_socket = Path("/run/user/1000/avault.sock")
        self._secrets_cache = {}
        
    async def start(self):
        """Connect to avault daemon."""
        # avault daemon must be running
        # Started by: python3 avault.py daemon start
        # Requires: Phone approval via NIP-46
        pass
    
    def extension_point(self, name: str):
        """Provide secrets and patterns to other plugins."""
        if name == "vault.secrets":
            return self._secrets_cache
        elif name == "vault.secret_patterns":
            # Return regex patterns for leak detection
            return [re.escape(v[:20]) for v in self._secrets_cache.values() if v]
    
    def get_secret(self, name: str) -> str:
        """Get secret from avault."""
        # Query avault daemon via Unix socket
        # Returns: secret value or None
        pass

PPQ Plugin with avault

class PPQPlugin(Plugin, LLMProvider):
    meta = PluginMeta(
        id="ppq",
        version="1.0.0",
        capabilities=["llm"],
        dependencies=["config", "vault"],  # NEW: depends on vault
        implements={
            "loop.before_llm": "on_before_llm_call",
            "loop.after_llm": "on_after_llm_call",
        },
    )

    def configure(self, config: dict) -> None:
        # OLD: Direct env access
        # self._api_key = os.environ.get("PPQ_API_KEY")
        
        # NEW: Get from vault extension point
        secrets = self.registry.extension_point("vault.secrets", default={})
        self._api_key = secrets.get("PPQ_API_KEY")
        
        if not self._api_key:
            self.log_warn("PPQ_API_KEY not found in vault")

Security Plugin with avault patterns

class SecurityPlugin(Plugin):
    """Phase 1 leak detection using avault secret patterns."""
    
    meta = PluginMeta(
        id="security",
        version="1.0.0",
        capabilities=["security"],
        dependencies=["vault"],  # Needs vault for patterns
    )
    
    def on_message_outbound(self, ctx):
        """Check outgoing messages for secret leaks."""
        # Get patterns from vault extension point
        patterns = self.registry.extension_point(
            "vault.secret_patterns", 
            default=[]
        )
        
        for pattern in patterns:
            if re.search(pattern, ctx.message):
                self.log_alert(f"BLOCKED: Secret leak in {ctx.channel}")
                return False  # Block message
        
        return True  # Allow message

Files Changed

  • cobot/core/plugin.py - Add secrets to PluginMeta
  • cobot/plugins/vault/plugin.py - NEW: avault integration
  • cobot/plugins/security/plugin.py - NEW: leak detection
  • cobot/plugins/ppq/plugin.py - Update to use vault
  • cobot/plugins/telegram/plugin.py - Update to use vault
  • cobot/plugins/nostr/plugin.py - Update to use vault

Dependencies

  • #50 (Secret Isolation)
  • #51 (Vault/Target)
  • #52 (Extension Points)
  • nazim/avault (External dependency)
## Status: ON HOLD ⚠️ Waiting for integration with [avault](https://forgejo.tail593e12.ts.net/nazim/avault) as central secret store. As noted by k9ert: *"I think it doesn't make so much sense to implement leak-detection without a central store of secrets which you can use to create the regexes to detect the leakage."* **Solution:** Use [avault](https://forgejo.tail593e12.ts.net/nazim/avault) - NIP-44 encrypted agent vault with NIP-46 remote signing. --- ## Problem Issue #50 describes the larger problem: Secrets are visible via Environment Variables to all plugins and subprocesses. As **Phase 1** (quick win), we propose Leak Detection — a safety net that catches accidental secret leaks before they leave the agent. ## Goal Scan all outbound messages (Telegram, Nostr, API calls) on secret patterns and block/redact if found. ## Central Secret Store: avault [avault](https://forgejo.tail593e12.ts.net/nazim/avault) provides: - **NIP-44 encryption** - XChaCha20-Poly1305 for vault files - **NIP-46 remote signing** - Phone (Amber) approves decryption - **RAM-only secrets** - Never touch disk unencrypted - **Unix socket API** - Daemon/CLI communication ### avault Integration for Cobot ```python # cobot/plugins/vault/plugin.py class VaultPlugin(Plugin): """avault integration for secret management.""" meta = PluginMeta( id="vault", version="1.0.0", capabilities=["secrets"], extension_points=[ "vault.secrets", "vault.secret_patterns" ] ) def __init__(self): self._avault_socket = Path("/run/user/1000/avault.sock") self._secrets_cache = {} async def start(self): """Connect to avault daemon.""" # avault daemon must be running # Started by: python3 avault.py daemon start # Requires: Phone approval via NIP-46 pass def extension_point(self, name: str): """Provide secrets and patterns to other plugins.""" if name == "vault.secrets": return self._secrets_cache elif name == "vault.secret_patterns": # Return regex patterns for leak detection return [re.escape(v[:20]) for v in self._secrets_cache.values() if v] def get_secret(self, name: str) -> str: """Get secret from avault.""" # Query avault daemon via Unix socket # Returns: secret value or None pass ``` ### PPQ Plugin with avault ```python class PPQPlugin(Plugin, LLMProvider): meta = PluginMeta( id="ppq", version="1.0.0", capabilities=["llm"], dependencies=["config", "vault"], # NEW: depends on vault implements={ "loop.before_llm": "on_before_llm_call", "loop.after_llm": "on_after_llm_call", }, ) def configure(self, config: dict) -> None: # OLD: Direct env access # self._api_key = os.environ.get("PPQ_API_KEY") # NEW: Get from vault extension point secrets = self.registry.extension_point("vault.secrets", default={}) self._api_key = secrets.get("PPQ_API_KEY") if not self._api_key: self.log_warn("PPQ_API_KEY not found in vault") ``` ### Security Plugin with avault patterns ```python class SecurityPlugin(Plugin): """Phase 1 leak detection using avault secret patterns.""" meta = PluginMeta( id="security", version="1.0.0", capabilities=["security"], dependencies=["vault"], # Needs vault for patterns ) def on_message_outbound(self, ctx): """Check outgoing messages for secret leaks.""" # Get patterns from vault extension point patterns = self.registry.extension_point( "vault.secret_patterns", default=[] ) for pattern in patterns: if re.search(pattern, ctx.message): self.log_alert(f"BLOCKED: Secret leak in {ctx.channel}") return False # Block message return True # Allow message ``` ## Files Changed - `cobot/core/plugin.py` - Add secrets to PluginMeta - `cobot/plugins/vault/plugin.py` - NEW: avault integration - `cobot/plugins/security/plugin.py` - NEW: leak detection - `cobot/plugins/ppq/plugin.py` - Update to use vault - `cobot/plugins/telegram/plugin.py` - Update to use vault - `cobot/plugins/nostr/plugin.py` - Update to use vault ## Dependencies - [avault](https://forgejo.tail593e12.ts.net/nazim/avault) - NIP-44 encrypted vault - [nostr-sdk](https://pypi.org/project/nostr-sdk/) - Python Nostr library ## Related - #50 (Secret Isolation) - #51 (Vault/Target) - #52 (Extension Points) - nazim/avault (External dependency)
Author
Contributor

Updated the issue description with a concrete usage example. The key insight: Integration with existing plugins (telegram, nostr) should happen FIRST, then the infrastructure supports it.

Simplified approach:

  1. Plugin declares secrets=["TELEGRAM_BOT_TOKEN"] in meta
  2. Registry loads from env, creates patterns
  3. Security plugin scans ALL outbound messages
  4. Match = block + alert

No external vault needed for Phase 1. See updated description.

Updated the issue description with a concrete usage example. The key insight: Integration with existing plugins (telegram, nostr) should happen FIRST, then the infrastructure supports it. Simplified approach: 1. Plugin declares secrets=["TELEGRAM_BOT_TOKEN"] in meta 2. Registry loads from env, creates patterns 3. Security plugin scans ALL outbound messages 4. Match = block + alert No external vault needed for Phase 1. See updated description.
noopsec changed title from [Phase 1] Leak Detection: Scan outbound messages for accidental secret exposure to [Phase 1] Leak Detection: Scan outbound messages for accidental secret exposure [ON HOLD] 2026-02-27 18:28:03 +00:00
noopsec changed title from [Phase 1] Leak Detection: Scan outbound messages for accidental secret exposure [ON HOLD] to [Phase 1] Leak Detection: Scan outbound messages for accidental secret exposure [ON HOLD - needs avault integration] 2026-02-27 18:49:43 +00:00
Collaborator

Security/Privacy reviewer: Ben (@webdiverblue) wants to be looped in on any security and privacy findings related to secret management, avault integration, and leak detection. CC alongside @k9ert on all security-relevant updates.

**Security/Privacy reviewer:** Ben (@webdiverblue) wants to be looped in on any security and privacy findings related to secret management, avault integration, and leak detection. CC alongside @k9ert on all security-relevant updates.
Sign in to join this conversation.
No milestone
No project
No assignees
2 participants
Notifications
Due date
The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set.

Reference
ultanio/cobot#145
No description provided.