"""Tests for GET /api/engagements//c2/callbacks//history.""" from __future__ import annotations import pytest from cryptography.fernet import Fernet from flask.testing import FlaskClient from backend.app.services.c2.adapter import C2Error from backend.tests.conftest import auth_headers as _h _FERNET_KEY = Fernet.generate_key().decode() @pytest.fixture(autouse=True) def set_encryption_key(monkeypatch): monkeypatch.setenv("MIMIC_ENCRYPTION_KEY", _FERNET_KEY) @pytest.fixture(autouse=True) def use_fake_adapter(monkeypatch): monkeypatch.setenv("MIMIC_C2_ADAPTER", "fake") # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_engagement(client: FlaskClient, token: str) -> dict: resp = client.post( "/api/engagements", headers=_h(token), json={"name": "Op Alpha", "start_date": "2026-06-10"}, ) assert resp.status_code == 201 return resp.get_json() def _put_config(client: FlaskClient, token: str, eid: int) -> None: resp = client.put( f"/api/engagements/{eid}/c2-config", headers=_h(token), json={"url": "https://c2.internal:7443", "api_token": "s3cr3t", "verify_tls": True}, ) assert resp.status_code == 200 def _history(client: FlaskClient, token: str, eid: int, cid: int, **params): return client.get( f"/api/engagements/{eid}/c2/callbacks/{cid}/history", headers=_h(token), query_string=params, ) # --------------------------------------------------------------------------- # Happy path # --------------------------------------------------------------------------- class TestHistoryHappyPath: def test_returns_200(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) assert resp.status_code == 200 def test_response_shape(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) body = resp.get_json() assert "tasks" in body assert "total" in body assert "page" in body assert "page_size" in body def test_task_shape(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) task = resp.get_json()["tasks"][0] for field in ("display_id", "command", "params", "status", "completed", "timestamp"): assert field in task, f"missing field: {field}" def test_default_page_is_1(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) assert resp.get_json()["page"] == 1 def test_default_page_size_is_25(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) assert resp.get_json()["page_size"] == 25 def test_callback_1_has_12_total(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) assert resp.get_json()["total"] == 12 def test_callback_2_has_0_tasks(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 2) body = resp.get_json() assert body["total"] == 0 assert body["tasks"] == [] def test_pagination_page_size_applied(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1, page=1, page_size=5) body = resp.get_json() assert len(body["tasks"]) == 5 assert body["page_size"] == 5 def test_redteam_can_view_history( self, client: FlaskClient, admin_token: str, redteam_token: str ) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, redteam_token, eng["id"], 1) assert resp.status_code == 200 # --------------------------------------------------------------------------- # Validation errors # --------------------------------------------------------------------------- class TestHistoryValidation: def test_400_page_size_too_large(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1, page_size=101) assert resp.status_code == 400 def test_400_page_zero(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1, page=0) assert resp.status_code == 400 def test_400_page_size_zero(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1, page_size=0) assert resp.status_code == 400 def test_400_page_negative(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1, page=-1) assert resp.status_code == 400 def test_400_page_size_100_is_ok(self, client: FlaskClient, admin_token: str) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1, page_size=100) assert resp.status_code == 200 # --------------------------------------------------------------------------- # Authorization / error cases # --------------------------------------------------------------------------- class TestHistoryErrors: def test_403_soc( self, client: FlaskClient, admin_token: str, soc_token: str ) -> None: eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, soc_token, eng["id"], 1) assert resp.status_code == 403 def test_503_no_key( self, monkeypatch, client: FlaskClient, admin_token: str ) -> None: monkeypatch.delenv("MIMIC_ENCRYPTION_KEY", raising=False) eng = _make_engagement(client, admin_token) resp = _history(client, admin_token, eng["id"], 1) assert resp.status_code == 503 def test_404_engagement_not_found( self, client: FlaskClient, admin_token: str ) -> None: resp = _history(client, admin_token, 9999, 1) assert resp.status_code == 404 def test_404_no_c2_config( self, client: FlaskClient, admin_token: str ) -> None: eng = _make_engagement(client, admin_token) resp = _history(client, admin_token, eng["id"], 1) assert resp.status_code == 404 def test_502_adapter_error( self, monkeypatch, client: FlaskClient, admin_token: str ) -> None: from backend.app.services.c2 import fake as fake_mod def _boom(self, callback_display_id, page=1, page_size=25): raise C2Error("upstream error") monkeypatch.setattr(fake_mod.FakeAdapter, "list_callback_tasks", _boom) eng = _make_engagement(client, admin_token) _put_config(client, admin_token, eng["id"]) resp = _history(client, admin_token, eng["id"], 1) assert resp.status_code == 502 assert "upstream error" in resp.get_json().get("error", "")