Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
93 lines
3.0 KiB
Python
93 lines
3.0 KiB
Python
"""Delegation Token Service — HMAC-signed short-lived tokens for internal AI calls.
|
|
|
|
Tokens are signed with the app SECRET_KEY using HMAC-SHA256.
|
|
Max lifetime: 60 seconds. No persistent storage — stateless verification.
|
|
|
|
Token format: base64(payload).base64(signature)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from app.config import get_settings
|
|
|
|
DELEGATION_AUDIENCE = "internal-ai-delegation"
|
|
MAX_TOKEN_LIFETIME = 60 # seconds
|
|
|
|
|
|
def _get_secret() -> bytes:
|
|
"""Get the signing secret from app settings."""
|
|
return get_settings().secret_key.encode()
|
|
|
|
|
|
def _sign(payload: dict) -> str:
|
|
"""Sign payload with HMAC-SHA256 and return base64(payload).base64(sig)."""
|
|
payload_bytes = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
|
|
sig = hmac.new(_get_secret(), payload_bytes, hashlib.sha256).digest()
|
|
return f"{base64.b64encode(payload_bytes).decode()}.{base64.b64encode(sig).decode()}"
|
|
|
|
|
|
def _verify(token: str) -> dict | None:
|
|
"""Verify a delegation token. Returns payload dict or None."""
|
|
try:
|
|
payload_b64, sig_b64 = token.rsplit(".", 1)
|
|
payload_bytes = base64.b64decode(payload_b64)
|
|
expected_sig = hmac.new(_get_secret(), payload_bytes, hashlib.sha256).digest()
|
|
actual_sig = base64.b64decode(sig_b64)
|
|
if not hmac.compare_digest(expected_sig, actual_sig):
|
|
return None
|
|
payload = json.loads(payload_bytes)
|
|
# Check expiry
|
|
now = datetime.now(UTC)
|
|
expires_at = datetime.fromisoformat(payload["expires_at"])
|
|
if expires_at.tzinfo is None:
|
|
expires_at = expires_at.replace(tzinfo=UTC)
|
|
if now > expires_at:
|
|
return None
|
|
# Check audience
|
|
if payload.get("audience") != DELEGATION_AUDIENCE:
|
|
return None
|
|
return payload
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def create_delegation_token(
|
|
user_id: str,
|
|
tenant_id: str,
|
|
agent_id: str = "ai-copilot",
|
|
lifetime_seconds: int = MAX_TOKEN_LIFETIME,
|
|
) -> str:
|
|
"""Create a short-lived delegation token for an internal AI call.
|
|
|
|
The token contains:
|
|
- user_id, tenant_id: who the AI acts on behalf of
|
|
- agent_id: which agent/service is calling
|
|
- audience: fixed to internal-ai-delegation
|
|
- expires_at: max 60 seconds from now
|
|
- token_id: unique ID for audit tracing
|
|
"""
|
|
now = datetime.now(UTC)
|
|
expires_at = now + timedelta(seconds=min(lifetime_seconds, MAX_TOKEN_LIFETIME))
|
|
payload = {
|
|
"user_id": user_id,
|
|
"tenant_id": tenant_id,
|
|
"agent_id": agent_id,
|
|
"audience": DELEGATION_AUDIENCE,
|
|
"expires_at": expires_at.isoformat(),
|
|
"token_id": str(uuid.uuid4()),
|
|
}
|
|
return _sign(payload)
|
|
|
|
|
|
def verify_delegation_token(token: str) -> dict[str, Any] | None:
|
|
"""Verify a delegation token. Returns payload or None if invalid/expired."""
|
|
return _verify(token)
|