fix(permissions): admin can no longer operate tests — start/submit/edit/create, view only
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled

Admin still bypassed require_any_role (non-strict) on the pure
operator actions: start-execution, submit-red, start-blue-work,
submit-blue, pause/resume-timer, and the red/blue field-edit + general
test create/update/remediation/import-rt/template endpoints all used
the loose dependency even where admin wasn't in the role tuple.
Switched every one of these to require_any_role_strict, matching the
review/validate/dispute/hold/assign endpoints already fixed earlier.

Frontend: removed every remaining role === "admin" disjunct gating
Start Execution, Submit to Blue Team, blue evaluating actions, timer
control, red/blue field editing, test creation, and template creation.
Team-blindness visibility (isBlind) deliberately keeps the admin
bypass — admin still sees both sides unmasked for oversight, since
that's visibility, not operating.

Updated ~20 tests whose fixtures used the admin account as a
convenience shortcut to create/drive tests through the workflow —
switched them to a red_lead account, fixing an incidental
fixture-resolution-order cookie bug along the way (client's cookie
jar prefers the last-logged-in role's cookie over any Bearer header
explicitly passed to a later request).
This commit is contained in:
kitos
2026-07-13 09:27:43 +02:00
parent 337faf824d
commit a3f86c7b31
13 changed files with 110 additions and 72 deletions
+8 -8
View File
@@ -37,8 +37,8 @@ from sqlalchemy.orm import Session
# Import get_db from app.database
from app.database import get_db
# Import get_current_user, require_any_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role
# Import get_current_user, require_any_role_strict from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role_strict
# Import UnitOfWork from app.domain.unit_of_work
from app.domain.unit_of_work import UnitOfWork
@@ -167,7 +167,7 @@ def template_stats(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> dict:
"""Return catalog statistics: active, by_source, by_platform.
@@ -195,7 +195,7 @@ def bulk_activate_templates(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> dict:
"""Set all templates to active or inactive.
@@ -317,7 +317,7 @@ def create_template(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestTemplateOut:
"""Create a custom test template.
@@ -386,7 +386,7 @@ def update_template(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestTemplateOut:
"""Update fields of an existing test template.
@@ -439,7 +439,7 @@ def toggle_template_active(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestTemplateOut:
"""Toggle a template between active and inactive (is_active = not is_active).
@@ -491,7 +491,7 @@ def delete_template(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> dict:
"""Soft-delete a test template by setting ``is_active=False``.
+17 -16
View File
@@ -42,7 +42,7 @@ from sqlalchemy.orm import Session
from app.database import get_db
# Import get_current_user, require_any_role from app.dependencies.auth
from app.dependencies.auth import get_current_user, require_any_role, require_any_role_strict
from app.dependencies.auth import get_current_user, require_any_role_strict
# Import UnitOfWork from app.domain.unit_of_work
from app.domain.unit_of_work import UnitOfWork
@@ -331,7 +331,7 @@ def create_test(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestOut:
"""Create a new test linked to an existing technique.
@@ -411,7 +411,7 @@ def create_test_from_template(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestOut:
"""Instantiate a real Test from an existing TestTemplate.
@@ -527,11 +527,12 @@ def update_test(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestOut:
"""Update one or more fields of an existing test.
Only leads or admins can update general test fields.
Only leads can update general test fields — admin administers the
site, not test content.
The test must be in ``draft`` or ``rejected`` state.
Args:
@@ -658,7 +659,7 @@ def update_test_red(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_tech", "red_lead")),
current_user: User = Depends(require_any_role_strict("red_tech", "red_lead")),
) -> TestOut:
"""Red Team updates their fields (allowed in ``draft`` and ``red_executing``).
@@ -724,7 +725,7 @@ def update_test_blue(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("blue_tech", "blue_lead")),
current_user: User = Depends(require_any_role_strict("blue_tech", "blue_lead")),
) -> TestOut:
"""Blue Team updates their fields (allowed only in ``blue_evaluating``).
@@ -788,7 +789,7 @@ def start_execution(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_tech", "red_lead")),
current_user: User = Depends(require_any_role_strict("red_tech", "red_lead")),
) -> TestOut:
"""Move a test from ``draft`` to ``red_executing``.
@@ -839,7 +840,7 @@ def submit_red(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_tech", "red_lead")),
current_user: User = Depends(require_any_role_strict("red_tech", "red_lead")),
) -> TestOut:
"""Red Team finalises — move from ``red_executing`` to ``blue_evaluating``.
@@ -887,7 +888,7 @@ def submit_blue(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("blue_tech", "blue_lead")),
current_user: User = Depends(require_any_role_strict("blue_tech", "blue_lead")),
) -> TestOut:
"""Blue Team finalises — move from ``blue_evaluating`` to ``in_review``.
@@ -931,7 +932,7 @@ def submit_blue(
def start_blue_work(
test_id: uuid.UUID,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("blue_tech", "blue_lead")),
current_user: User = Depends(require_any_role_strict("blue_tech", "blue_lead")),
):
"""Blue tech picks up the test to start evaluating. Sets the Tempo timer start."""
test = crud_get_test_or_raise(db, test_id)
@@ -1029,7 +1030,7 @@ def pause_timer(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_tech", "blue_tech", "red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_tech", "blue_tech", "red_lead", "blue_lead")),
) -> TestOut:
"""Pause the running timer for the current phase (red_executing or blue_evaluating).
@@ -1068,7 +1069,7 @@ def resume_timer(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_tech", "blue_tech", "red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_tech", "blue_tech", "red_lead", "blue_lead")),
) -> TestOut:
"""Resume the paused timer for the current phase.
@@ -1446,7 +1447,7 @@ def update_remediation(
# Entry: db
db: Session = Depends(get_db),
# Entry: current_user
current_user: User = Depends(require_any_role("red_lead", "blue_lead")),
current_user: User = Depends(require_any_role_strict("red_lead", "blue_lead")),
) -> TestOut:
"""Update remediation fields on a test.
@@ -1800,13 +1801,13 @@ class RTImportPayload(BaseModel):
def import_rt(
payload: RTImportPayload,
db: Session = Depends(get_db),
current_user: User = Depends(require_any_role("red_lead")),
current_user: User = Depends(require_any_role_strict("red_lead")),
):
"""Import results from a real Red Team engagement.
Creates one Test record per technique in ``validated`` state (bypassing
the normal Red/Blue workflow) and immediately recalculates coverage metrics.
Requires ``red_lead`` or ``admin`` role.
Requires ``red_lead`` — admin administers the site, not test content.
"""
# Pre-validate: every technique must include at least one evidence image
for entry in payload.techniques:
@@ -26,8 +26,39 @@ DUMMY_ID = str(uuid.uuid4())
("post", f"/api/v1/tests/{DUMMY_ID}/resume", None),
("post", f"/api/v1/tests/{DUMMY_ID}/assign", {"red_tech_assignee": DUMMY_ID}),
("patch", f"/api/v1/tests/{DUMMY_ID}/classification", {"data_classification": "pii"}),
("post", f"/api/v1/tests/{DUMMY_ID}/start-execution", None),
("post", f"/api/v1/tests/{DUMMY_ID}/submit-red", None),
("post", f"/api/v1/tests/{DUMMY_ID}/start-blue-work", None),
("post", f"/api/v1/tests/{DUMMY_ID}/submit-blue", None),
("post", f"/api/v1/tests/{DUMMY_ID}/pause-timer", None),
("post", f"/api/v1/tests/{DUMMY_ID}/resume-timer", None),
("patch", f"/api/v1/tests/{DUMMY_ID}/red", {"procedure_text": "x"}),
("patch", f"/api/v1/tests/{DUMMY_ID}/blue", {"detection_result": "detected"}),
("patch", f"/api/v1/tests/{DUMMY_ID}", {"name": "x"}),
("patch", f"/api/v1/tests/{DUMMY_ID}/remediation", {"remediation_status": "completed"}),
("post", "/api/v1/tests", {"technique_id": DUMMY_ID, "name": "x"}),
("post", "/api/v1/tests/from-template", {"template_id": DUMMY_ID, "technique_id": DUMMY_ID}),
("post", "/api/v1/tests/import-rt", {"name": "x", "techniques": []}),
],
)
def test_admin_forbidden(client, auth_headers, method, path, payload):
resp = getattr(client, method)(path, json=payload, headers=auth_headers)
assert resp.status_code == 403, f"{method.upper()} {path} -> {resp.status_code}: {resp.text}"
@pytest.mark.parametrize(
"method,path,payload",
[
("post", "/api/v1/test-templates", {"name": "x", "mitre_technique_id": "T1059"}),
("patch", f"/api/v1/test-templates/{DUMMY_ID}", {"name": "x"}),
("patch", f"/api/v1/test-templates/{DUMMY_ID}/toggle-active", None),
("delete", f"/api/v1/test-templates/{DUMMY_ID}", None),
("patch", "/api/v1/test-templates/bulk-activate", {"template_ids": [DUMMY_ID], "is_active": True}),
],
)
def test_admin_forbidden_templates(client, auth_headers, method, path, payload):
kwargs = {"headers": auth_headers}
if payload is not None:
kwargs["json"] = payload
resp = getattr(client, method)(path, **kwargs)
assert resp.status_code == 403, f"{method.upper()} {path} -> {resp.status_code}: {resp.text}"
+2 -2
View File
@@ -46,10 +46,10 @@ def technique(api, auth_headers):
@pytest.fixture
def blind_test(client, db, api, auth_headers, red_tech_headers, technique):
def blind_test(client, db, api, red_lead_headers, red_lead_user, red_tech_headers, technique):
"""A test in red_executing with red-side content already filled in."""
resp = api(
"post", "/api/v1/tests", auth_headers,
"post", "/api/v1/tests", red_lead_headers,
json={"technique_id": technique, "name": "Blind visibility test"},
)
test_id = resp.json()["id"]
+2 -2
View File
@@ -49,10 +49,10 @@ def test_new_test_defaults_to_internal_use_only_via_db_default(db, red_lead_user
assert test.data_classification == "internal_use_only"
def test_create_test_endpoint_uses_tactic_heuristic(client, db, api, auth_headers, technique):
def test_create_test_endpoint_uses_tactic_heuristic(client, db, api, red_lead_headers, red_lead_user, technique):
"""Going through the real create-test flow applies the tactic-based heuristic."""
resp = api(
"post", "/api/v1/tests", auth_headers,
"post", "/api/v1/tests", red_lead_headers,
json={"technique_id": technique["id"], "name": "Heuristic test"},
)
assert resp.status_code == 201, resp.text
+1 -1
View File
@@ -36,7 +36,7 @@ def _reach_disputed(client, db, api, auth_headers, red_tech_headers, red_lead_he
blue_tech_headers, blue_lead_headers, technique):
"""Drive a fresh test all the way to disputed: red approves, blue rejects."""
resp = api(
"post", "/api/v1/tests", auth_headers,
"post", "/api/v1/tests", red_lead_headers,
json={"technique_id": technique, "name": "Dispute test"},
)
test_id = resp.json()["id"]
+12 -12
View File
@@ -86,7 +86,7 @@ def test_review_red_forbidden_for_non_assigned_lead(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
"""A red_lead who isn't the assigned reviewer gets 403."""
test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)
test_id = _reach_red_review(client, db, api, red_lead_headers, red_tech_headers, technique)
# red_lead_user IS the only red_lead fixture, so it WILL be auto-assigned
# as reviewer (single-candidate load balancing). To exercise the 403
# path we need a second, non-assigned red_lead.
@@ -112,7 +112,7 @@ def test_review_red_forbidden_for_non_assigned_lead(
def test_review_red_approve_moves_to_blue_evaluating(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)
test_id = _reach_red_review(client, db, api, red_lead_headers, red_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "approve", "notes": "LGTM"})
assert resp.status_code == 200, resp.text
@@ -125,7 +125,7 @@ def test_review_red_approve_moves_to_blue_evaluating(
def test_review_red_reopen_requires_notes(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)
test_id = _reach_red_review(client, db, api, red_lead_headers, red_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "reopen"})
assert resp.status_code == 400
@@ -134,7 +134,7 @@ def test_review_red_reopen_requires_notes(
def test_review_red_reopen_moves_to_red_executing(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)
test_id = _reach_red_review(client, db, api, red_lead_headers, red_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-red", red_lead_headers, json={"decision": "reopen", "notes": "add more detail"})
assert resp.status_code == 200, resp.text
@@ -144,11 +144,11 @@ def test_review_red_reopen_moves_to_red_executing(
def test_review_red_forbidden_for_admin(
client, db, api, auth_headers, red_tech_headers, red_lead_user, technique
client, db, api, auth_headers, red_tech_headers, red_lead_headers, red_lead_user, technique
):
"""Admin administers the site, not the test workflow — reviewing is a
red_lead-only action now, with no admin override."""
test_id = _reach_red_review(client, db, api, auth_headers, red_tech_headers, technique)
test_id = _reach_red_review(client, db, api, red_lead_headers, red_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-red", auth_headers, json={"decision": "approve"})
assert resp.status_code == 403
@@ -163,7 +163,7 @@ def test_review_blue_forbidden_for_non_assigned_lead(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
red_lead_user, blue_lead_user, technique,
):
test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
test_id = _reach_blue_review(client, db, api, red_lead_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
from app.auth import hash_password
from app.models.user import User
@@ -187,7 +187,7 @@ def test_review_blue_approve_moves_to_in_review(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
red_lead_user, blue_lead_user, technique,
):
test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
test_id = _reach_blue_review(client, db, api, red_lead_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "approve"})
assert resp.status_code == 200, resp.text
@@ -198,7 +198,7 @@ def test_review_blue_reopen_requires_notes(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
red_lead_user, blue_lead_user, technique,
):
test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
test_id = _reach_blue_review(client, db, api, red_lead_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "reopen"})
assert resp.status_code == 400
@@ -208,7 +208,7 @@ def test_review_blue_reopen_moves_to_blue_evaluating(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
red_lead_user, blue_lead_user, technique,
):
test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
test_id = _reach_blue_review(client, db, api, red_lead_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "reopen", "notes": "redo detection"})
assert resp.status_code == 200, resp.text
@@ -221,7 +221,7 @@ def test_review_blue_gap_requires_system_gaps(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
red_lead_user, blue_lead_user, technique,
):
test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
test_id = _reach_blue_review(client, db, api, red_lead_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
resp = api("post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers, json={"decision": "gap"})
assert resp.status_code == 400
@@ -231,7 +231,7 @@ def test_review_blue_gap_moves_to_in_review_with_system_gaps(
client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, blue_lead_headers,
red_lead_user, blue_lead_user, technique,
):
test_id = _reach_blue_review(client, db, api, auth_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
test_id = _reach_blue_review(client, db, api, red_lead_headers, red_tech_headers, red_lead_headers, blue_tech_headers, technique)
resp = api(
"post", f"/api/v1/tests/{test_id}/review-blue", blue_lead_headers,
+16 -10
View File
@@ -31,8 +31,12 @@ def test_create_test_requires_auth(client):
assert response.status_code in (401, 403)
def test_create_test_success(client, auth_headers, technique):
"""Admin can create a test via POST /tests."""
def test_create_test_success(client, red_lead_headers, red_lead_user, technique):
"""Lead can create a test via POST /tests (admin administers the site, not test content)."""
# The `technique` fixture logs in as admin (its own auth_headers
# dependency), leaving a cookie that would otherwise outrank the
# red_lead Bearer header below — get_current_user prefers the cookie.
client.cookies.clear()
response = client.post(
"/api/v1/tests",
json={
@@ -41,7 +45,7 @@ def test_create_test_success(client, auth_headers, technique):
"description": "Test description",
"platform": "windows",
},
headers=auth_headers,
headers=red_lead_headers,
)
assert response.status_code == 201
data = response.json()
@@ -50,7 +54,7 @@ def test_create_test_success(client, auth_headers, technique):
assert data["technique_id"] == technique["id"]
def test_create_test_nonexistent_technique(client, auth_headers):
def test_create_test_nonexistent_technique(client, red_lead_headers, red_lead_user):
"""Creating a test with non-existent technique fails."""
response = client.post(
"/api/v1/tests",
@@ -58,27 +62,29 @@ def test_create_test_nonexistent_technique(client, auth_headers):
"technique_id": "00000000-0000-0000-0000-000000000000",
"name": "Test",
},
headers=auth_headers,
headers=red_lead_headers,
)
assert response.status_code == 404
def test_get_test_by_id(client, auth_headers, technique):
"""GET /tests/{id} returns the test."""
def test_get_test_by_id(client, auth_headers, technique, red_lead_headers, red_lead_user):
"""GET /tests/{id} returns the test. Admin can still view (just not create)."""
client.cookies.clear()
create_response = client.post(
"/api/v1/tests",
json={"technique_id": technique["id"], "name": "Test"},
headers=auth_headers,
headers=red_lead_headers,
)
test_id = create_response.json()["id"]
client.cookies.clear()
response = client.get(f"/api/v1/tests/{test_id}", headers=auth_headers)
assert response.status_code == 200
assert response.json()["id"] == test_id
def test_start_execution_twice_returns_invalid_transition(
client, auth_headers, technique, red_tech_user
client, red_lead_headers, red_lead_user, technique, red_tech_user
):
"""Invalid workflow transition surfaces domain error JSON (FASE 0.4).
@@ -89,7 +95,7 @@ def test_start_execution_twice_returns_invalid_transition(
create_response = client.post(
"/api/v1/tests",
json={"technique_id": technique["id"], "name": "Workflow dup start"},
headers=auth_headers,
headers=red_lead_headers,
)
assert create_response.status_code == 201
test_id = create_response.json()["id"]