kitos/Aegis
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(phase-12): implement Red/Blue API endpoints (T-109, T-110, T-111, T-112)

Browse Source 
T-109: Rewrite tests router with full Red/Blue workflow endpoints - list with filters, create from template, Red/Blue team updates with state guards, start-execution, submit-red, submit-blue, validate-red, validate-blue, reopen, and timeline. All using workflow service from Phase 11.

T-110: Rewrite evidence router with Red/Blue separation - upload with team field, list with team filter, delete with state-based permissions. Red Team edits in draft/red_executing, Blue Team in blue_evaluating, admin bypasses all.

T-111: Create test_templates router with full CRUD - paginated list with source/platform/severity/search filters, by-technique lookup, admin-only create/update, and soft delete. Registered in main.py.

T-112: Add POST /system/import-atomic-tests endpoint to system router - admin-only trigger for Atomic Red Team import with error handling and statistics response.

Includes validation tests for all four tasks (35 checks total).
This commit is contained in:
Kitos
2026-02-09 10:45:33 +01:00
parent 7af6be10be
commit 9d7832c571
9 changed files with 1789 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

260
backend/tests/test_t110_evidence_router.py Normal file
View File

@@ -0,0 +1,260 @@
"""Validation tests for T-110: Evidence Router with Red/Blue separation.
Tests the permission logic and endpoint structure.
"""
import sys
import os
import uuid
from unittest.mock import MagicMock
from types import ModuleType
# ---------------------------------------------------------------------------
# Stubs
# ---------------------------------------------------------------------------
backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
if backend_dir not in sys.path:
sys.path.insert(0, backend_dir)
if "pydantic_settings" not in sys.modules:
pydantic_settings_mock = ModuleType("pydantic_settings")
class _BaseSettings:
def __init__(self, **kwargs): pass
def __init_subclass__(cls, **kwargs): super().__init_subclass__(**kwargs)
pydantic_settings_mock.BaseSettings = _BaseSettings
sys.modules["pydantic_settings"] = pydantic_settings_mock
if "app.config" not in sys.modules:
config_mod = ModuleType("app.config")
class _FakeSettings:
DATABASE_URL = "sqlite:///:memory:"
SECRET_KEY = "test"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
MINIO_ENDPOINT = "localhost:9000"
MINIO_ACCESS_KEY = "test"
MINIO_SECRET_KEY = "test"
MINIO_BUCKET = "test"
config_mod.settings = _FakeSettings()
sys.modules["app.config"] = config_mod
if "app.database" not in sys.modules:
db_mod = ModuleType("app.database")
db_mod.Base = type("Base", (), {"metadata": MagicMock()})
db_mod.get_db = MagicMock()
sys.modules["app.database"] = db_mod
for mod_name in [
"taxii2client", "taxii2client.v20",
"jose", "boto3", "botocore", "botocore.exceptions",
"apscheduler", "apscheduler.schedulers",
"apscheduler.schedulers.background",
"apscheduler.triggers", "apscheduler.triggers.cron",
]:
if mod_name not in sys.modules:
m = ModuleType(mod_name)
if mod_name == "taxii2client.v20": m.Server = MagicMock
elif mod_name == "jose": m.JWTError = Exception; m.jwt = MagicMock()
elif mod_name == "boto3": m.client = MagicMock()
elif mod_name == "botocore.exceptions": m.ClientError = Exception
elif mod_name == "apscheduler.schedulers.background": m.BackgroundScheduler = MagicMock
elif mod_name == "apscheduler.triggers.cron": m.CronTrigger = MagicMock
sys.modules[mod_name] = m
# ---------------------------------------------------------------------------
# Imports
# ---------------------------------------------------------------------------
from fastapi import HTTPException
from app.models.enums import TeamSide, TestState
from app.routers.evidence import (
router,
_validate_upload_permission,
_validate_delete_permission,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_test(state):
t = MagicMock()
t.id = uuid.uuid4()
t.state = state
return t
def _make_user(role):
u = MagicMock()
u.id = uuid.uuid4()
u.role = role
return u
def _make_evidence(team, uploaded_by=None, test_id=None):
e = MagicMock()
e.id = uuid.uuid4()
e.test_id = test_id or uuid.uuid4()
e.team = team
e.uploaded_by = uploaded_by or uuid.uuid4()
return e
# ---------------------------------------------------------------------------
# 1. red_tech can upload team=red in red_executing
# ---------------------------------------------------------------------------
def test_red_tech_upload_red_in_red_executing():
test = _make_test(TestState.red_executing)
user = _make_user("red_tech")
# Should not raise
_validate_upload_permission(test, TeamSide.red, user)
print(" [PASS] red_tech can upload team=red in red_executing")
# ---------------------------------------------------------------------------
# 2. red_tech can upload team=red in draft
# ---------------------------------------------------------------------------
def test_red_tech_upload_red_in_draft():
test = _make_test(TestState.draft)
user = _make_user("red_tech")
_validate_upload_permission(test, TeamSide.red, user)
print(" [PASS] red_tech can upload team=red in draft")
# ---------------------------------------------------------------------------
# 3. red_tech CANNOT upload team=blue (403)
# ---------------------------------------------------------------------------
def test_red_tech_cannot_upload_blue():
test = _make_test(TestState.red_executing)
user = _make_user("red_tech")
try:
_validate_upload_permission(test, TeamSide.blue, user)
assert False, "Should have raised HTTPException"
except HTTPException as exc:
assert exc.status_code == 403
print(" [PASS] red_tech CANNOT upload team=blue (403)")
# ---------------------------------------------------------------------------
# 4. blue_tech can upload team=blue in blue_evaluating
# ---------------------------------------------------------------------------
def test_blue_tech_upload_blue_in_blue_evaluating():
test = _make_test(TestState.blue_evaluating)
user = _make_user("blue_tech")
_validate_upload_permission(test, TeamSide.blue, user)
print(" [PASS] blue_tech can upload team=blue in blue_evaluating")
# ---------------------------------------------------------------------------
# 5. blue_tech CANNOT upload team=red (403)
# ---------------------------------------------------------------------------
def test_blue_tech_cannot_upload_red():
test = _make_test(TestState.blue_evaluating)
user = _make_user("blue_tech")
try:
_validate_upload_permission(test, TeamSide.red, user)
assert False, "Should have raised HTTPException"
except HTTPException as exc:
assert exc.status_code == 403
print(" [PASS] blue_tech CANNOT upload team=red (403)")
# ---------------------------------------------------------------------------
# 6. GET /tests/{id}/evidence?team=red — endpoint exists with team filter
# ---------------------------------------------------------------------------
def test_list_evidence_endpoint():
routes = {}
for route in router.routes:
path = getattr(route, "path", "")
methods = getattr(route, "methods", set())
for method in methods:
routes[f"{method} {path}"] = route
found = any(
"GET" in k and "/evidence" in k and "{test_id}" in k
for k in routes
)
assert found, f"GET /tests/{{test_id}}/evidence not found. Routes: {list(routes.keys())}"
print(" [PASS] GET /tests/{id}/evidence endpoint exists (filterable by team)")
# ---------------------------------------------------------------------------
# 7. DELETE in in_review → 403
# ---------------------------------------------------------------------------
def test_delete_in_review_fails():
test = _make_test(TestState.in_review)
user = _make_user("red_tech")
evidence = _make_evidence(TeamSide.red, uploaded_by=user.id)
try:
_validate_delete_permission(test, evidence, user)
assert False, "Should have raised HTTPException"
except HTTPException as exc:
assert exc.status_code == 403
print(" [PASS] DELETE in in_review -> 403")
# ---------------------------------------------------------------------------
# 8. DELETE red evidence in red_executing → allowed
# ---------------------------------------------------------------------------
def test_delete_red_evidence_in_red_executing():
test = _make_test(TestState.red_executing)
user = _make_user("red_tech")
evidence = _make_evidence(TeamSide.red, uploaded_by=user.id)
# Should not raise
_validate_delete_permission(test, evidence, user)
print(" [PASS] DELETE red evidence in red_executing -> allowed")
# ---------------------------------------------------------------------------
# 9. Admin can upload any team in any state
# ---------------------------------------------------------------------------
def test_admin_bypass():
admin = _make_user("admin")
# Red in blue_evaluating (normally blocked)
test1 = _make_test(TestState.blue_evaluating)
_validate_upload_permission(test1, TeamSide.red, admin)
# Blue in draft (normally blocked)
test2 = _make_test(TestState.draft)
_validate_upload_permission(test2, TeamSide.blue, admin)
print(" [PASS] Admin can upload any team in any state")
# ---------------------------------------------------------------------------
# Run all
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("T-110 Validation: Evidence Router with Red/Blue Separation")
print("=" * 60)
test_red_tech_upload_red_in_red_executing()
test_red_tech_upload_red_in_draft()
test_red_tech_cannot_upload_blue()
test_blue_tech_upload_blue_in_blue_evaluating()
test_blue_tech_cannot_upload_red()
test_list_evidence_endpoint()
test_delete_in_review_fails()
test_delete_red_evidence_in_red_executing()
test_admin_bypass()
print("=" * 60)
print("ALL T-110 validations PASSED!")