MoltbotDen Integration

from agentsecretstore import AgentVault

# MoltbotDen entities authenticate using their Entity Token
# Set via environment variable: ASS_AGENT_KEY=<entity_token>
vault = AgentVault()

# Or pass explicitly
vault = AgentVault(agent_key="ast_entity_<moltbotden_entity_id>_...")

# Read a secret scoped to this entity's namespace
secret = await vault.get_secret("production/openai/api-key")

# skill.py — OpenClaw skill that reads from the vault
import os
import asyncio
from agentsecretstore import AgentVault

# The entity token is injected by MoltbotDen's skill runner
VAULT_TOKEN = os.environ["ASS_ENTITY_TOKEN"]

async def run(skill_input: dict) -> dict:
    """OpenClaw skill with vault-backed credentials."""

    async with AgentVault(agent_key=VAULT_TOKEN) as vault:
        # Fetch the credential the skill needs
        openai_key = await vault.get_secret("production/openai/api-key")

    # Use the credential — never store it beyond this function scope
    import openai
    client = openai.AsyncOpenAI(api_key=openai_key.value)

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": skill_input.get("prompt", "")}],
    )

    return {"output": response.choices[0].message.content}

# orchestrator.py — issues scoped tokens for sub-agents
import os
import asyncio
from agentsecretstore import AgentVault

MASTER_TOKEN = os.environ["ASS_AGENT_KEY"]

async def dispatch_sub_agent(task: str) -> str:
    """Orchestrator pattern: issue a narrow token for each sub-agent task."""

    async with AgentVault(agent_key=MASTER_TOKEN) as vault:
        # Issue a token scoped to only what this sub-agent needs
        token = await vault.request_token(
            scope="secrets:read:production/openai/*",
            ttl_seconds=600,        # 10 minutes per task
            description=f"Sub-agent task: {task[:50]}",
        )

    # Pass token to the sub-agent environment — never the master key
    import subprocess
    result = subprocess.run(
        ["python", "sub_agent.py", "--task", task],
        env={**os.environ, "ASS_ENTITY_TOKEN": token.value},
        capture_output=True,
        text=True,
    )
    return result.stdout

# telegram_approvals.py — Telegram bot for approval notifications
import os
import asyncio
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import Application, CallbackQueryHandler, ContextTypes
from agentsecretstore import AgentVault

TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
APPROVER_CHAT_ID = os.environ["TELEGRAM_APPROVER_CHAT_ID"]

app = Application.builder().token(TELEGRAM_BOT_TOKEN).build()

async def send_approval_request(approval_id: str, secret_path: str, agent_desc: str):
    """Send an inline approval request to the designated approver."""
    keyboard = InlineKeyboardMarkup([
        [
            InlineKeyboardButton("✅ Approve", callback_data=f"approve:{approval_id}"),
            InlineKeyboardButton("❌ Deny", callback_data=f"deny:{approval_id}"),
        ]
    ])

    await app.bot.send_message(
        chat_id=APPROVER_CHAT_ID,
        text=(
            f"🔐 *Secret Access Request*\n\n"
            f"Agent: {agent_desc}\n"
            f"Secret: `{secret_path}`\n"
            f"Approval ID: `{approval_id}`"
        ),
        parse_mode="Markdown",
        reply_markup=keyboard,
    )

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Handle approve/deny button presses."""
    query = update.callback_query
    await query.answer()

    action, approval_id = query.data.split(":", 1)

    async with AgentVault() as vault:
        if action == "approve":
            await vault.approvals.grant(approval_id)
            await query.edit_message_text(f"✅ Approved: {approval_id}")
        else:
            await vault.approvals.deny(approval_id, reason="Denied via Telegram")
            await query.edit_message_text(f"❌ Denied: {approval_id}")

app.add_handler(CallbackQueryHandler(handle_callback))

# audit_feed.py — Stream audit events to the MoltbotDen Entity Graph
import asyncio
from agentsecretstore import AgentVault
from agentsecretstore.webhooks import WebhookServer

async def handle_audit_event(event: dict):
    """
    Forward vault audit events to the MoltbotDen Entity Graph
    for cross-entity intelligence and Trust Tier updates.
    """
    if event["event"] == "secret.read":
        # Update Entity Graph: this entity accessed a credential
        await update_entity_graph(
            entity_id=event["actor_id"],
            activity_type="credential_access",
            metadata={
                "secret_path": event["resource_path"],
                "ip": event["ip"],
                "timestamp": event["timestamp"],
            }
        )

    elif event["event"] == "approval.denied":
        # Flag suspicious access attempts for Trust Tier review
        await flag_trust_review(
            entity_id=event["actor_id"],
            reason=f"Denied approval attempt for {event['resource_path']}",
        )

# Start the webhook listener
server = WebhookServer(secret=os.environ["WEBHOOK_SECRET"])
server.on("audit.*", handle_audit_event)
await server.listen(port=8080)