"""Contratti di integrazione per policy tool nei diversi entry point backend. Gli endpoint sono invocati direttamente con finti loop/provider. Ogni test misura la policy effettivamente passata al loop o dimostra che un contratto letterale termina prima di provider, notifiche, planner e tool. """ from __future__ import annotations import asyncio import hashlib import hmac import json import os import time import sys import types import unittest from unittest.mock import patch from fastapi import HTTPException _BACKEND = os.path.join(os.path.dirname(__file__), "..") if _BACKEND not in sys.path: sys.path.insert(0, _BACKEND) NO_TOOL_PROMPT = "Spiega il concetto senza usare strumenti, tool, rete o file." LITERAL_PROMPT = ( "Rispondi ESCLUSIVAMENTE con TEST_E2E_OK. Non usare strumenti, tool, " "comandi shell, file, rete, servizi esterni, azioni o card." ) class _Request: def __init__(self, headers: dict[str, str] | None = None) -> None: self.headers = headers or {} class _WebhookRequest(_Request): def __init__(self, raw_body: bytes, headers: dict[str, str]) -> None: super().__init__(headers) self._raw_body = raw_body async def body(self) -> bytes: return self._raw_body class _CallbackRequest(_Request): async def json(self) -> dict[str, object]: raise AssertionError("bad secret must not parse payload") class _RecordingLoop: calls: list[dict] = [] def __init__(self, **_kwargs: object) -> None: pass async def run(self, **kwargs: object) -> dict[str, object]: self.__class__.calls.append(dict(kwargs)) return {"success": True, "output": "safe textual result", "steps": []} class _FakeAIClient: pass def _fake_loop_modules() -> dict[str, types.ModuleType]: agents = types.ModuleType("agents") agents.__path__ = [] # type: ignore[attr-defined] unified = types.ModuleType("agents.unified_loop") unified.UnifiedAgentLoop = _RecordingLoop # type: ignore[attr-defined] models = types.ModuleType("models") models.__path__ = [] # type: ignore[attr-defined] ai_client = types.ModuleType("models.ai_client") ai_client.AIClient = _FakeAIClient # type: ignore[attr-defined] return { "agents": agents, "agents.unified_loop": unified, "models": models, "models.ai_client": ai_client, } async def _noop(*_args: object, **_kwargs: object) -> None: return None def _signed_webhook_request( payload: dict[str, object], *, event_id: str = "evt-000000000001", timestamp: int | None = None, signature: str | None = None, ) -> _WebhookRequest: raw = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") ts = str(int(time.time()) if timestamp is None else timestamp) signing = b"v1." + ts.encode("ascii") + b"." + event_id.encode("ascii") + b"." + raw expected = hmac.new(b"h" * 32, signing, hashlib.sha256).hexdigest() return _WebhookRequest(raw, { "x-webhook-id": event_id, "x-webhook-timestamp": ts, "x-webhook-signature": signature or f"sha256={expected}", }) class EntryPointPolicyContracts(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self) -> None: from api.webhook_security import WebhookDeliveryStore _RecordingLoop.calls.clear() await WebhookDeliveryStore.reset_memory_for_test() async def test_telegram_callback_fails_closed_before_payload_or_outbound_work(self) -> None: from api import webhook with patch.dict(os.environ, {"TELEGRAM_WEBHOOK_SECRET": ""}, clear=False): result = await webhook.telegram_callback(_CallbackRequest()) self.assertEqual(result, {"ok": True, "ignored": "bad_secret"}) async def test_webhook_literal_contract_returns_before_loop_or_notification(self) -> None: from api import webhook import api.state as state request = _signed_webhook_request({"goal": LITERAL_PROMPT}) env = { "WEBHOOK_TOKEN": "test-webhook-token", "WEBHOOK_HMAC_SECRET": "h" * 32, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false", } with patch.dict(os.environ, env, clear=False), \ patch.object(state, "_sb", None), \ patch.object(webhook, "_tg_start", side_effect=AssertionError("notification forbidden")), \ patch.object(webhook, "_get_mem_manager", side_effect=AssertionError("provider forbidden")): result = await webhook.inbound_webhook("test-webhook-token", request) self.assertEqual(result["output"], "TEST_E2E_OK") self.assertEqual(result["steps"], 0) self.assertEqual(_RecordingLoop.calls, []) async def test_public_chat_literal_contract_returns_before_loop_or_notification(self) -> None: from api import webhook payload = webhook.PublicChatPayload(message=LITERAL_PROMPT, conversation_id="conv-safe") with patch.dict(os.environ, {"PUBLIC_API_TOKEN": "public-test-token"}, clear=False), \ patch.object(webhook, "_tg_start", side_effect=AssertionError("notification forbidden")), \ patch.object(webhook, "_get_mem_manager", side_effect=AssertionError("provider forbidden")): result = await webhook.public_chat( payload, _Request({"authorization": "Bearer public-test-token"}), ) self.assertEqual(result["response"], "TEST_E2E_OK") self.assertEqual(result["conversation_id"], "conv-safe") self.assertEqual(_RecordingLoop.calls, []) async def test_webhook_and_public_chat_propagate_no_tool_policy_to_loop(self) -> None: from api import webhook import api.state as state fake_modules = _fake_loop_modules() env = { "WEBHOOK_TOKEN": "test-webhook-token", "PUBLIC_API_TOKEN": "public-test-token", "WEBHOOK_HMAC_SECRET": "h" * 32, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false", } with patch.dict(sys.modules, fake_modules), \ patch.dict(os.environ, env, clear=False), \ patch.object(state, "_sb", None), \ patch.object(webhook, "_get_mem_manager", return_value=object()), \ patch.object(webhook, "_get_executor", return_value=object()), \ patch.object(webhook, "_get_planner", return_value=object()), \ patch.object(webhook, "_tg_start", _noop), \ patch.object(webhook, "_tg_done", _noop): webhook_result = await webhook.inbound_webhook( "test-webhook-token", _signed_webhook_request({"goal": NO_TOOL_PROMPT}) ) public_result = await webhook.public_chat( webhook.PublicChatPayload(message=NO_TOOL_PROMPT), _Request({"authorization": "Bearer public-test-token"}), ) self.assertTrue(webhook_result["ok"]) self.assertTrue(public_result["ok"]) self.assertEqual(len(_RecordingLoop.calls), 2) self.assertTrue(all(call["allow_tools"] is False for call in _RecordingLoop.calls)) async def test_webhook_replay_conflict_and_expired_timestamp_are_rejected_before_dispatch(self) -> None: from api import webhook import api.state as state env = { "WEBHOOK_TOKEN": "test-webhook-token", "WEBHOOK_HMAC_SECRET": "h" * 32, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false", "WEBHOOK_MAX_AGE_SECONDS": "300", } first = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-replay-000001") duplicate = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-replay-000001") conflict = _signed_webhook_request({"goal": LITERAL_PROMPT, "context": [{"content": "different"}]}, event_id="evt-replay-000001") expired = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-expired-0001", timestamp=int(time.time()) - 301) with patch.dict(os.environ, env, clear=False), patch.object(state, "_sb", None): first_response = await webhook.inbound_webhook("test-webhook-token", first) duplicate_response = await webhook.inbound_webhook("test-webhook-token", duplicate) with self.assertRaises(HTTPException) as conflict_error: await webhook.inbound_webhook("test-webhook-token", conflict) with self.assertRaises(HTTPException) as expired_error: await webhook.inbound_webhook("test-webhook-token", expired) self.assertEqual(first_response, duplicate_response) self.assertEqual(conflict_error.exception.status_code, 409) self.assertEqual(expired_error.exception.status_code, 401) async def test_webhook_bad_signature_is_rejected_before_pydantic_parse(self) -> None: from api import webhook request = _signed_webhook_request({"goal": LITERAL_PROMPT}, signature="sha256=" + "0" * 64) env = {"WEBHOOK_TOKEN": "test-webhook-token", "WEBHOOK_HMAC_SECRET": "h" * 32} with patch.dict(os.environ, env, clear=False), \ patch.object(webhook.WebhookPayload, "model_validate_json", side_effect=AssertionError("must not parse")): with self.assertRaises(HTTPException) as signature_error: await webhook.inbound_webhook("test-webhook-token", request) self.assertEqual(signature_error.exception.status_code, 401) async def test_scheduler_literal_contract_skips_provider_initialization(self) -> None: from api import scheduler with patch.object(scheduler, "_get_ai_client", side_effect=AssertionError("provider forbidden"), create=True): result = await scheduler._run_goal(LITERAL_PROMPT) self.assertEqual(result, "TEST_E2E_OK") self.assertEqual(_RecordingLoop.calls, []) async def test_scheduler_propagates_no_tool_policy_to_loop(self) -> None: from api import scheduler import api.state as state fake_modules = _fake_loop_modules() async def fake_memory() -> object: return object() with patch.dict(sys.modules, fake_modules), \ patch.object(state, "_get_ai_client", return_value=object()), \ patch.object(state, "_get_mem_manager_async", fake_memory), \ patch.object(state, "_get_executor", return_value=object()), \ patch.object(state, "_get_planner", return_value=object()): result = await scheduler._run_goal(NO_TOOL_PROMPT, risk="safe") self.assertEqual(result, "safe textual result") self.assertEqual(len(_RecordingLoop.calls), 1) self.assertIs(_RecordingLoop.calls[0]["allow_tools"], False) async def test_restored_sse_task_rebuilds_literal_policy_before_task_start(self) -> None: from api import agent task_id = "resume-literal-policy" original = dict(agent._agent_tasks) agent._agent_tasks.clear() agent._agent_tasks[task_id] = {"id": task_id, "goal": LITERAL_PROMPT, "status": "QUEUED"} try: with patch.object(agent, "sb_update_status", _noop): response = await agent.stream_agent_task(task_id, _Request()) payload = b"".join([ chunk.encode() if isinstance(chunk, str) else chunk async for chunk in response.body_iterator ]).decode() finally: agent._agent_tasks.clear() agent._agent_tasks.update(original) self.assertIn('"event": "task_done"', payload) self.assertIn("TEST_E2E_OK", payload) self.assertNotIn("task_start", payload) self.assertTrue(agent._agent_tasks == original or task_id not in agent._agent_tasks) if __name__ == "__main__": unittest.main()