Secrets & Permissions

apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: ai-agent-read-only
  namespace: production
rules:
- apiGroups: [""]
  resources: ["pods", "services", "configmaps", "events"]
  verbs: ["get", "list", "watch"]
- apiGroups: ["apps"]
  resources: ["deployments", "replicasets"]
  verbs: ["get", "list", "watch"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: ai-agent-binding
  namespace: production
subjects:
- kind: ServiceAccount
  name: ai-agent
  namespace: production
roleRef:
  kind: Role
  name: ai-agent-read-only
  apiGroup: rbac.authorization.k8s.io

import time

class CredentialManager:
    def __init__(self, vault_client):
        self.vault = vault_client
        self.cache = {}

    def get_credential(self, service, ttl_seconds=300):
        """Get a short-lived credential from the secret store."""
        cache_key = f"{service}:{int(time.time() // ttl_seconds)}"

        if cache_key not in self.cache:
            # Fetch a new token with short TTL
            token = self.vault.read(
                path=f"creds/{service}",
                ttl=ttl_seconds
            )
            self.cache[cache_key] = {
                "token": token["client_token"],
                "expires": time.time() + ttl_seconds
            }

        return self.cache[cache_key]["token"]

    def invalidate(self, service):
        """Immediately invalidate cached credentials."""
        keys_to_remove = [k for k in self.cache if k.startswith(service)]
        for key in keys_to_remove:
            del self.cache[key]

import re

class SecretRedactor:
    def __init__(self):
        self.patterns = [
            (r"(?:api[_-]?key|token|secret|password)[=:\s]+[\w-]+", "[REDACTED_SECRET]"),
            (r"Bearer\s+[\w-]+", "Bearer [REDACTED]"),
            (r"\b[A-Za-z0-9]{32,}\b", "[REDACTED_TOKEN]"),
        ]

    def redact(self, text):
        """Remove secrets from text before logging."""
        redacted = text
        for pattern, replacement in self.patterns:
            redacted = re.sub(pattern, replacement, redacted, flags=re.IGNORECASE)
        return redacted

    def redact_tool_call(self, tool_name, arguments):
        """Redact secrets from tool call logs."""
        redacted_args = {}
        for key, value in arguments.items():
            if isinstance(value, str):
                redacted_args[key] = self.redact(value)
            else:
                redacted_args[key] = value
        return {"tool": tool_name, "arguments": redacted_args}

class AuditLog:
    def __init__(self, log_store):
        self.store = log_store

    def log_action(self, agent_id, action, result, user):
        """Log every action with full context."""
        entry = {
            "agent_id": agent_id,
            "action": action["tool"],
            "arguments": self.redact_arguments(action["arguments"]),
            "result_status": "success" if result["success"] else "failure",
            "user": user,
            "timestamp": time.time(),
            "checksum": self.compute_checksum(action, result),
        }
        self.store.append_immutable(entry)

    def redact_arguments(self, args):
        """Redact secrets from arguments."""
        redactor = SecretRedactor()
        return {k: redactor.redact(str(v)) for k, v in args.items()}

    def compute_checksum(self, action, result):
        """Compute a checksum for tamper detection."""
        import hashlib
        data = json.dumps({"action": action, "result": result}, sort_keys=True)
        return hashlib.sha256(data.encode()).hexdigest()