from __future__ import annotations import importlib import secrets import pytest from flask import Flask, jsonify from sqlalchemy import create_engine from app.api.agent_group import agent_group_bp from app.api.auth import auth_bp from app.api.template import template_bp from app.db import Base, create_session_factory from app.services.identity import IdentityRepository, PasswordService from app.utils.api_errors import ApiError from app.utils.locale import t agent_group_module = importlib.import_module("app.api.agent_group") template_module = importlib.import_module("app.api.template") TEST_EMAIL = "auxiliary-security@example.com" TEST_AUTH_INPUT = "local-only-auth-input" def make_app(): engine = create_engine("sqlite+pysqlite:///:memory:") Base.metadata.create_all(engine) session_factory = create_session_factory(engine) app = Flask(__name__) app.config.update( TESTING=True, SESSION_COOKIE_SECURE=False, ) app.config["SECRET_KEY"] = secrets.token_hex(32) app.extensions["crowdsight_session_factory"] = session_factory @app.errorhandler(ApiError) def handle_api_error(error: ApiError): return jsonify(error.to_payload(t)), error.status_code app.register_blueprint(auth_bp, url_prefix="/api/auth") app.register_blueprint(template_bp, url_prefix="/api/template") app.register_blueprint(agent_group_bp, url_prefix="/api/agent-group") with session_factory() as session: repo = IdentityRepository(session) organization = repo.create_organization(name="Auxiliary Org", slug="auxiliary-org") user = repo.create_user( email=TEST_EMAIL, password_hash=PasswordService.hash_password(TEST_AUTH_INPUT), ) repo.create_membership(user.id, organization.id, "admin") session.commit() return app, engine def login(client): response = client.post( "/api/auth/login", json={"email": TEST_EMAIL, "password": TEST_AUTH_INPUT}, ) assert response.status_code == 200 return client.get_cookie("crowdsight_csrf").value def test_template_list_requires_authentication(): app, engine = make_app() try: response = app.test_client().get("/api/template/list") assert response.status_code == 401 assert response.get_json()["error_code"] == "unauthorized" finally: engine.dispose() def test_agent_group_categorize_requires_authentication(): app, engine = make_app() try: response = app.test_client().post( "/api/agent-group/categorize", json={"agents": [{"name": "Alice"}]}, ) assert response.status_code == 401 assert response.get_json()["error_code"] == "unauthorized" finally: engine.dispose() def test_authenticated_auxiliary_reads_and_pure_filter_remain_available(): app, engine = make_app() try: client = app.test_client() csrf = login(client) templates = client.get("/api/template/list") assert templates.status_code == 200 assert templates.get_json()["success"] is True filtered = client.post( "/api/agent-group/filter", json={ "agents": [{"agent_id": 1}], "groups": [{"group_id": "all", "agent_indices": [0]}], "selected_group_ids": ["all"], }, headers={"X-CSRF-Token": csrf}, ) assert filtered.status_code == 200 assert filtered.get_json()["selected_agent_ids"] == [0] finally: engine.dispose() @pytest.mark.parametrize( ("path", "payload"), [ ("/api/template/auto-select", {"text": "Alice founded Orbit."}), ("/api/agent-group/categorize", {"agents": [{"name": "Alice"}]}), ], ) def test_auxiliary_llm_mutations_require_csrf_and_idempotency(path, payload): app, engine = make_app() try: client = app.test_client() csrf = login(client) missing_csrf = client.post( path, json=payload, headers={"Idempotency-Key": "auxiliary-mutation-1"}, ) assert missing_csrf.status_code == 403 assert missing_csrf.get_json()["error_code"] == "csrf_failed" missing_idempotency = client.post( path, json=payload, headers={"X-CSRF-Token": csrf}, ) assert missing_idempotency.status_code == 400 assert missing_idempotency.get_json()["error_code"] == "idempotency_required" finally: engine.dispose() def test_every_auxiliary_route_requires_authentication(): app, engine = make_app() try: client = app.test_client() routes = [ ("GET", "/api/template/list", None), ("GET", "/api/template/news_event/filter-rules", None), ("POST", "/api/template/auto-select", {"text": "Alice founded Orbit."}), ("POST", "/api/agent-group/filter", {"agents": [], "groups": []}), ("POST", "/api/agent-group/categorize", {"agents": [{"name": "Alice"}]}), ] for method, path, payload in routes: response = client.open(path, method=method, json=payload) assert response.status_code == 401, (method, path, response.get_json()) finally: engine.dispose() def test_llm_auxiliary_errors_are_safe(monkeypatch): class ExplodingLLM: def chat_json(self, **_kwargs): raise RuntimeError("sensitive backend detail") monkeypatch.setattr(template_module, "LLMClient", ExplodingLLM) monkeypatch.setattr(agent_group_module, "LLMClient", ExplodingLLM) app, engine = make_app() try: client = app.test_client() csrf = login(client) responses = [ client.post( "/api/template/auto-select", json={"text": "Alice founded Orbit."}, headers={"X-CSRF-Token": csrf, "Idempotency-Key": "safe-template-error"}, ), client.post( "/api/agent-group/categorize", json={"agents": [{"name": "Alice"}]}, headers={"X-CSRF-Token": csrf, "Idempotency-Key": "safe-agent-error"}, ), ] for response in responses: body = response.get_json() assert response.status_code == 500 assert body["error_code"] == "internal_error" assert "sensitive backend detail" not in response.get_data(as_text=True) finally: engine.dispose() def test_llm_auxiliary_mutations_replay_completed_responses(monkeypatch): class FakeTemplateLLM: def chat_json(self, **_kwargs): return { "template_id": "news_event", "prompt": "Alice founded Orbit.", "confidence": 0.9, "reasoning": "The input describes a news event.", } class FakeAgentLLM: def chat_json(self, **_kwargs): return { "groups": [ { "group_id": "audience", "group_name": "Audience", "default_enabled": True, "agent_indices": [0], } ] } monkeypatch.setattr(template_module, "LLMClient", FakeTemplateLLM) monkeypatch.setattr(agent_group_module, "LLMClient", FakeAgentLLM) app, engine = make_app() try: client = app.test_client() csrf = login(client) cases = [ ( "/api/template/auto-select", {"text": "Alice founded Orbit."}, "template-replay", ), ( "/api/agent-group/categorize", {"agents": [{"name": "Alice"}]}, "agent-replay", ), ] for path, payload, key in cases: headers = {"X-CSRF-Token": csrf, "Idempotency-Key": key} first = client.post(path, json=payload, headers=headers) second = client.post(path, json=payload, headers=headers) assert first.status_code == 200 assert second.status_code == 200 assert second.get_json() == first.get_json() finally: engine.dispose()