Vault MCP Bridge

from fastapi import Depends, Request
from .utils import Router as APIRouter
import logging
from ..models import SSHPassOp, SSHSignOp, Principal
from ..security import require_scopes
from ..vault import client_for_principal

router = APIRouter(prefix="/ssh", tags=["ssh"])

@router.post("/otp")
async def ssh_otp(op: SSHPassOp, request: Request, p: Principal = Depends(require_scopes(["write"]))):
    client = client_for_principal(p)
    res = client.secrets.ssh.generate_credential(name=op.role, username=op.username, ip=op.ip, port=op.port)
    data = res.get("data", {})
    out = {
        "ip": data.get("ip"),
        "username": data.get("username"),
        "port": data.get("port"),
        "otp": data.get("key") or data.get("otp"),
        "lease_id": res.get("lease_id"),
        "lease_duration": res.get("lease_duration"),
    }
    try:
        logging.getLogger("vault_mcp.response").info(
            "ssh_otp",
            extra={"extra": {"subject": p.subject, "role": op.role, "ip": op.ip, "user": op.username, "request_id": request.headers.get("x-request-id")}},
        )
    except Exception:
        pass
    return out

@router.post("/sign")
async def ssh_sign(op: SSHSignOp, request: Request, p: Principal = Depends(require_scopes(["write"]))):
    client = client_for_principal(p)
    res = client.secrets.ssh.sign_key(name=op.role, public_key=op.public_key, cert_type=op.cert_type, valid_principals=op.valid_principals, ttl=op.ttl)
    data = res.get("data", {})
    cert = data.get("signed_key") or data.get("signed_key_pem") or data.get("ssh_signature")
    try:
        logging.getLogger("vault_mcp.response").info(
            "ssh_sign",
            extra={"extra": {"subject": p.subject, "role": op.role, "cert_len": len(cert) if cert else 0, "request_id": request.headers.get("x-request-id")}},
        )
    except Exception:
        pass
    return {"certificate": cert}