fix(security): F01 (Astra P0) — KI-Tool-Ausführung ohne Freigabe verhindern

Vorher: _execute_tool (agent_loop.py) und der KI-Chat-Loop
(stream_chat_comm) führten JEDES im Registry registrierte Tool aus, wenn
das LLM dessen Namen lieferte — ohne Abgleich mit der angebotenen Liste,
ohne required_permission-Check. Reproduktion (Astra): Nur audit_allowed
angeboten, Modell nannte audit_restricted (system:admin) → Handler lief.

Fix (fail-closed, an ALLEN Ausfuehrungspfaden):
- _check_tool_access: (1) Allowlist — nur Tools die dem LLM angeboten
  wurden duerfen laufen; (2) required_permission gegen die AKTUELLEN
  User-Rechte (deny-first, Rechteentzug wirkt sofort, ohne Kontext =
  Ablehnung). Guard vor dry-run/approval/execute-Pfaden.
- stream_chat_comm: gleicher Allowlist-Guard vor execute_tool_call.
- run_react_loop/agent_runner/agent_stream/agent_routes reichen
  user_permissions durch (perm_ctx bzw. Session-User).
- check_permission: Session-Kontexte tragen denied_permissions statt
  denied — beide Keys werden gelesen, Deny-Liste wird nie mehr ignoriert.

Tests: test_agent_loop.py 18/18 (7 neue F01-Tests nach Astra-Abnahme:
nicht angeboten → Handler null; fehlende Permission → abgewiesen;
Fail-closed ohne Kontext; Deny-Liste session-shape; Rechteentzug
mitten im Lauf wirkt auf naechste Aktion; dry-run guardet auch).
ruff clean. Pre-existing-Beweis: permission_system_live-Failures
reproduzieren sich ohne diesen Patch identisch (Plugin-Aktivierung in
ephemeraler Test-DB, bekanntes Vorbestands-Finding).

tests/test_agent_loop.py

@@ -16,7 +16,6 @@ import pytest
 
 from app.ai.agent_loop import ReActResult, ReActStep, run_react_loop
 
-
 # ──────────────────────────────────────────────────────────────────────────
 # Test helpers
 # ──────────────────────────────────────────────────────────────────────────
@@ -416,6 +415,306 @@ async def test_tool_not_found(agent_def, tenant_id, user_id, mock_db):
     assert "not found" in (result.steps[0].observation or "").lower()
 
 
+def _make_guard_registry(tools: dict[str, Any]) -> MagicMock:
+    """Build a mock ToolRegistry with REAL attribute values.
+
+    Unlike ``_make_tool_registry`` (whose MagicMock tools auto-create any
+    attribute), this sets ``required_permission`` explicitly so the F01
+    guard behaves like production AITool objects.
+    """
+    registry = MagicMock()
+
+    def _get(name: str) -> Any:
+        return tools.get(name)
+
+    registry.get = _get
+    return registry
+
+
+def _guard_tool(name: str, handler: AsyncMock, required_permission: str | None = None) -> MagicMock:
+    t = MagicMock()
+    t.name = name
+    t.handler = handler
+    t.required_permission = required_permission
+    return t
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# F01 (Astra P0): allowlist + required_permission enforced at execution time
+# ──────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_f01_registered_but_not_offered_tool_rejected(agent_def, tenant_id, user_id, mock_db):
+    """F01 acceptance: a tool registered in the registry but NOT offered
+    to the LLM is rejected — the handler must never run (stays null)."""
+    handler = AsyncMock(return_value="secret audit data")
+    registry = _make_guard_registry({
+        "audit_restricted": _guard_tool("audit_restricted", handler, required_permission="system:admin"),
+    })
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = [
+            _make_llm_response(
+                content="calling restricted tool",
+                tool_calls=[_make_tool_call(name="audit_restricted")],
+                cost_usd=0.001,
+            ),
+            _make_llm_response(content="understood", cost_usd=0.001),
+        ]
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "run audit_restricted"}],
+            tools=[{"type": "function", "function": {"name": "audit_allowed"}}],  # only audit_allowed offered
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            user_permissions={"permissions": ["system:admin"], "denied": [], "is_system_admin": True},
+        )
+
+    assert result.status == "completed"
+    handler.assert_not_awaited()  # core F01 acceptance: handler stays null
+    assert "not available" in (result.steps[0].observation or "")
+
+
+@pytest.mark.asyncio
+async def test_f01_offered_tool_user_lacks_permission_rejected(agent_def, tenant_id, user_id, mock_db):
+    """Offered tool with required_permission: user WITHOUT the permission
+    is rejected even though the tool was offered to the LLM."""
+    handler = AsyncMock(return_value="mail sent")
+    registry = _make_guard_registry({
+        "send_mail": _guard_tool("send_mail", handler, required_permission="mail:write"),
+    })
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = [
+            _make_llm_response(
+                content="sending",
+                tool_calls=[_make_tool_call(name="send_mail")],
+                cost_usd=0.001,
+            ),
+            _make_llm_response(content="ok", cost_usd=0.001),
+        ]
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "send a mail"}],
+            tools=[{"type": "function", "function": {"name": "send_mail"}}],
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            user_permissions={"permissions": ["mail:read"], "denied": [], "is_system_admin": False},
+        )
+
+    assert result.status == "completed"
+    handler.assert_not_awaited()
+    assert "Permission 'mail:write' required" in (result.steps[0].observation or "")
+
+
+@pytest.mark.asyncio
+async def test_f01_offered_tool_with_permission_executes(agent_def, tenant_id, user_id, mock_db):
+    """Offered tool + user HAS the required permission → executes normally."""
+    handler = AsyncMock(return_value="mail sent")
+    registry = _make_guard_registry({
+        "send_mail": _guard_tool("send_mail", handler, required_permission="mail:write"),
+    })
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = [
+            _make_llm_response(
+                content="sending",
+                tool_calls=[_make_tool_call(name="send_mail")],
+                cost_usd=0.001,
+            ),
+            _make_llm_response(content="done", cost_usd=0.001),
+        ]
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "send a mail"}],
+            tools=[{"type": "function", "function": {"name": "send_mail"}}],
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            user_permissions={"permissions": ["mail:write", "mail:read"], "denied": [], "is_system_admin": False},
+        )
+
+    assert result.status == "completed"
+    handler.assert_awaited_once()
+    assert result.steps[0].observation == "mail sent"
+
+
+@pytest.mark.asyncio
+async def test_f01_fail_closed_without_permission_context(agent_def, tenant_id, user_id, mock_db):
+    """Tool with required_permission but NO permission context → rejected
+    (fail-closed: missing context is not implicit access)."""
+    handler = AsyncMock(return_value="boom")
+    registry = _make_guard_registry({
+        "dangerous": _guard_tool("dangerous", handler, required_permission="system:admin"),
+    })
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = [
+            _make_llm_response(
+                content="calling",
+                tool_calls=[_make_tool_call(name="dangerous")],
+                cost_usd=0.001,
+            ),
+            _make_llm_response(content="ok", cost_usd=0.001),
+        ]
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "call it"}],
+            tools=[{"type": "function", "function": {"name": "dangerous"}}],
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            user_permissions=None,  # no permission context at all
+        )
+
+    assert result.status == "completed"
+    handler.assert_not_awaited()
+    assert "Permission 'system:admin' required" in (result.steps[0].observation or "")
+
+
+@pytest.mark.asyncio
+async def test_f01_deny_list_session_shape_respected(agent_def, tenant_id, user_id, mock_db):
+    """Session user contexts carry ``denied_permissions`` (not ``denied``).
+    An explicit deny must reject even when the permission is also granted."""
+    handler = AsyncMock(return_value="mail sent")
+    registry = _make_guard_registry({
+        "send_mail": _guard_tool("send_mail", handler, required_permission="mail:write"),
+    })
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = [
+            _make_llm_response(
+                content="sending",
+                tool_calls=[_make_tool_call(name="send_mail")],
+                cost_usd=0.001,
+            ),
+            _make_llm_response(content="ok", cost_usd=0.001),
+        ]
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "send"}],
+            tools=[{"type": "function", "function": {"name": "send_mail"}}],
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            # session shape: granted AND denied via denied_permissions key
+            user_permissions={
+                "permissions": ["mail:write"],
+                "denied_permissions": ["mail:write"],
+                "is_system_admin": False,
+            },
+        )
+
+    assert result.status == "completed"
+    handler.assert_not_awaited()
+    assert "Permission 'mail:write' required" in (result.steps[0].observation or "")
+
+
+@pytest.mark.asyncio
+async def test_f01_permission_revoked_during_run_takes_effect_next_action(agent_def, tenant_id, user_id, mock_db):
+    """F01 acceptance: revoking permissions DURING a run takes effect on
+    the next action — the guard reads the CURRENT context every call."""
+    handler = AsyncMock(return_value="mail sent")
+    registry = _make_guard_registry({
+        "send_mail": _guard_tool("send_mail", handler, required_permission="mail:write"),
+    })
+
+    perms: dict[str, Any] = {
+        "permissions": ["mail:write"],
+        "denied": [],
+        "is_system_admin": False,
+    }
+
+    call_count = 0
+
+    async def llm_with_revocation(**kwargs: Any) -> dict[str, Any]:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return _make_llm_response(
+                content="first send",
+                tool_calls=[_make_tool_call(name="send_mail")],
+                cost_usd=0.001,
+            )
+        # Revoke the permission before the second LLM response
+        if call_count == 2:
+            perms["permissions"] = []  # permission revoked mid-run
+            return _make_llm_response(
+                content="second send",
+                tool_calls=[_make_tool_call(name="send_mail")],
+                cost_usd=0.001,
+            )
+        return _make_llm_response(content="finished", cost_usd=0.001)
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = llm_with_revocation
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "send twice"}],
+            tools=[{"type": "function", "function": {"name": "send_mail"}}],
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            user_permissions=perms,
+        )
+
+    assert result.status == "completed"
+    handler.assert_awaited_once()  # first call ran, second was rejected
+    assert result.steps[0].observation == "mail sent"
+    assert "Permission 'mail:write' required" in (result.steps[1].observation or "")
+
+
+@pytest.mark.asyncio
+async def test_f01_dry_run_still_guards_non_offered_tools(agent_def, tenant_id, user_id, mock_db):
+    """Dry-run must not even SIMULATE a non-offered tool — the guard runs
+    before the dry-run path."""
+    handler = AsyncMock(return_value="x")
+    registry = _make_guard_registry({
+        "secret_tool": _guard_tool("secret_tool", handler),
+    })
+
+    with patch("app.ai.agent_loop.llm_complete", new_callable=AsyncMock) as mock_llm:
+        mock_llm.side_effect = [
+            _make_llm_response(
+                content="trying",
+                tool_calls=[_make_tool_call(name="secret_tool")],
+                cost_usd=0.001,
+            ),
+            _make_llm_response(content="done", cost_usd=0.001),
+        ]
+
+        result = await run_react_loop(
+            agent_definition=agent_def,
+            messages=[{"role": "user", "content": "try it"}],
+            tools=[],  # nothing offered
+            tool_registry=registry,
+            db=mock_db,
+            tenant_id=tenant_id,
+            user_id=user_id,
+            dry_run=True,
+        )
+
+    assert result.status == "completed"
+    obs = result.steps[0].observation or ""
+    assert "not available" in obs
+    assert "would_execute" not in obs  # guard fired, not the dry-run simulation
+
+
 @pytest.mark.asyncio
 async def test_multiple_tool_calls_per_step(agent_def, tenant_id, user_id, mock_db):
     """Multiple tool calls in a single LLM response are all executed."""