fix(security): F09 (Astra P1) — CRM-/MCP-Tools delegieren mit HMAC-Token statt toter Header
Check Cross-Plugin Imports / check (push) Has been cancelled

Vorher: Zwei generische CRM-API-Tools (ai_assistant/crm_api_tool,
mcp_server/tool_definitions) sendeten X-Internal-Call/X-Tenant-Id/
X-User-Id-Header — die geschuetzte API akzeptiert diese nicht als
Authentisierung (Astra-Repro: "Not authenticated"). Der vorhandene
Delegationstoken-Code (app/core/delegation_token.py, HMAC-SHA256,
max 60s) war komplett unverbunden (0 Aufrufer). Im Worker zeigte der
lokale Default-Host zudem auf den Worker selbst.

Fix:
- get_current_user akzeptiert X-Delegation-Token: HMAC-verifiziert,
  baut den User-Kontext mit den ECHTEN Rechten des Users auf
  (get_cached_permissions + RLS-Kontext) — keine Sonderrechte
- CSRF-Middleware skippt Delegations-Header (browsers never attach
  them cross-site — gleiche Begruendung wie Bearer)
- Gemeinsamer Helper _make_internal_api_request in crm_api_tool:
  erstellt pro Request ein 60s-Delegationstoken, sendet es als
  X-Delegation-Token; MCP-Tool delegiert an denselben Helper
  (Astra: beide Implementierungen konsolidieren)
- _get_base_url: INTERNAL_API_URL-Override — Compose setzt fuer den
  Worker http://crm_app:8000 (127.0.0.1 zeigte im Worker auf sich
  selbst)

Abnahme (Astra): Dieselbe Fachaktion ist fuer denselben Benutzer ueber
UI und Agent gleichermaassen erlaubt oder gesperrt — die Tools laufen
jetzt mit den echten User-Rechten durch denselben Auth-Pfad. Das
Audit-Naming (delegated_by) folgt mit dem transparency-Update.

Verifikation: test_api_tokens (inkl. 6 Delegations-Tests) +
test_agent_loop + test_s1_security_guards 49/49, Syntax + ruff clean.
This commit is contained in:
Agent Zero
2026-09-18 12:12:18 +02:00
parent fdc4e36d14
commit 421726700b
5 changed files with 120 additions and 46 deletions
+55
View File
@@ -74,7 +74,62 @@ async def get_current_user(
Returns session data dict with user_id, tenant_id, email, name, role,
and resolved permissions from Redis cache.
F09 (Astra P1): also accepts a short-lived HMAC-signed delegation
token (X-Delegation-Token header) for INTERNAL calls made on behalf
of a user — e.g. the generic CRM-API tool used by AI agents and the
MCP server. Previously those tools sent unauthenticated
X-Internal-Call headers that the protected API never accepted.
"""
# F09: internal delegation path — HMAC-signed, max 60 seconds
delegation_header = request.headers.get("X-Delegation-Token", "")
if delegation_header:
from app.core.delegation_token import verify_delegation_token
payload = verify_delegation_token(delegation_header)
if payload is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"detail": "Invalid or expired delegation token", "code": "delegation_invalid"},
)
deleg_user_id = str(payload.get("user_id", ""))
deleg_tenant_id = str(payload.get("tenant_id", ""))
if not deleg_user_id or not deleg_tenant_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"detail": "Delegation token missing user/tenant", "code": "delegation_invalid"},
)
tenant_id = uuid.UUID(deleg_tenant_id)
user_id = uuid.UUID(deleg_user_id)
await set_tenant_context(db, tenant_id)
from app.core.permissions import get_cached_permissions
resolved = await get_cached_permissions(db, redis, user_id, tenant_id)
from sqlalchemy import select as _select
from app.models.group import UserGroup
groups_q = await db.execute(
_select(UserGroup.group_id)
.where(UserGroup.user_id == user_id)
.where(UserGroup.tenant_id == tenant_id)
)
group_ids = [row[0] for row in groups_q]
is_admin = bool(resolved.get("is_system_admin", False))
await set_user_context(db, user_id, group_ids, is_admin)
return {
"user_id": deleg_user_id,
"tenant_id": deleg_tenant_id,
"email": "", # not needed for permission decisions
"name": "delegated",
"role": "",
"permissions": resolved.get("permissions", []),
"denied_permissions": resolved.get("denied", []),
"field_permissions": resolved.get("field_permissions", {}),
"is_system_admin": is_admin,
"delegated_by": payload.get("agent_id", ""),
}
settings = get_settings()
session_id = request.cookies.get(settings.session_cookie_name)