kitos
df49d336f7
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
_REDACTED_KEYS listed jira.api_token/jira.password/tempo.api_token, but routers/system.py actually writes admin credentials under jira.admin_api_token and tempo.admin_token. The mismatch meant every config export bundle included the live Jira and Tempo admin API tokens in plaintext instead of redacting them. Found while researching the Jira/Tempo integration for a service-account request; added a regression test.
188 lines
7.3 KiB
Python
188 lines
7.3 KiB
Python
"""Tests for the admin configuration export/import bundle (Block 4).
|
|
|
|
The feature was already implemented (GET/POST /admin/export-config,
|
|
/admin/import-config) but had zero test coverage for an admin-sensitive
|
|
data-migration endpoint. Covers: admin-only gating, secret redaction,
|
|
custom-template-only scoping, and import idempotency/user-provisioning
|
|
safety (forced password reset, no secret restoration from redacted values).
|
|
"""
|
|
|
|
from app.models.scoring_config import ScoringConfig
|
|
from app.models.sso_config import SsoConfig
|
|
from app.models.system_config import SystemConfig
|
|
from app.models.test_template import TestTemplate
|
|
from app.models.user import User
|
|
from app.models.webhook_config import WebhookConfig
|
|
|
|
|
|
def test_export_config_requires_admin(client, red_tech_headers):
|
|
resp = client.get("/api/v1/admin/export-config", headers=red_tech_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_import_config_requires_admin(client, red_tech_headers):
|
|
resp = client.post("/api/v1/admin/import-config", json={}, headers=red_tech_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_export_config_returns_expected_sections(client, auth_headers):
|
|
resp = client.get("/api/v1/admin/export-config", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
for key in ("_meta", "system_configs", "webhooks", "sso", "scoring", "custom_templates", "users"):
|
|
assert key in body
|
|
assert resp.headers["content-disposition"].startswith("attachment;")
|
|
|
|
|
|
def test_export_config_redacts_smtp_password(client, db, auth_headers):
|
|
db.add(SystemConfig(key="smtp.password", value="hunter2"))
|
|
db.add(SystemConfig(key="smtp.host", value="smtp.example.com"))
|
|
db.commit()
|
|
|
|
body = client.get("/api/v1/admin/export-config", headers=auth_headers).json()
|
|
values = {r["key"]: r["value"] for r in body["system_configs"]}
|
|
|
|
assert values["smtp.password"] == "[REDACTED]"
|
|
assert values["smtp.host"] == "smtp.example.com"
|
|
|
|
|
|
def test_export_config_redacts_live_jira_and_tempo_admin_tokens(client, db, auth_headers):
|
|
"""Regression test: the redaction key list previously named
|
|
jira.api_token/jira.password/tempo.api_token, which don't match the
|
|
actual keys routers/system.py writes (jira.admin_api_token,
|
|
tempo.admin_token) — so the live admin Jira/Tempo credentials were
|
|
exported in plaintext instead of being redacted."""
|
|
db.add(SystemConfig(key="jira.admin_api_token", value="ATATT-real-jira-token"))
|
|
db.add(SystemConfig(key="tempo.admin_token", value="real-tempo-token"))
|
|
db.commit()
|
|
|
|
body = client.get("/api/v1/admin/export-config", headers=auth_headers).json()
|
|
values = {r["key"]: r["value"] for r in body["system_configs"]}
|
|
|
|
assert values["jira.admin_api_token"] == "[REDACTED]"
|
|
assert values["tempo.admin_token"] == "[REDACTED]"
|
|
|
|
|
|
def test_export_config_never_exports_sso_private_key(client, db, auth_headers):
|
|
db.add(SsoConfig(is_enabled=True, provider_name="Okta", sp_private_key="super-secret-key"))
|
|
db.commit()
|
|
|
|
body = client.get("/api/v1/admin/export-config", headers=auth_headers).json()
|
|
|
|
assert body["sso"]["sp_private_key"] == "[REDACTED]"
|
|
|
|
|
|
def test_export_config_only_includes_custom_templates(client, db, auth_headers):
|
|
db.add(TestTemplate(mitre_technique_id="T1059", name="Atomic one", source="atomic_red_team"))
|
|
db.add(TestTemplate(mitre_technique_id="T1059", name="My custom one", source="custom"))
|
|
db.commit()
|
|
|
|
body = client.get("/api/v1/admin/export-config", headers=auth_headers).json()
|
|
names = {t["name"] for t in body["custom_templates"]}
|
|
|
|
assert "My custom one" in names
|
|
assert "Atomic one" not in names
|
|
|
|
|
|
def test_export_config_users_have_no_secrets(client, db, auth_headers, red_tech_user):
|
|
body = client.get("/api/v1/admin/export-config", headers=auth_headers).json()
|
|
exported = next(u for u in body["users"] if u["username"] == red_tech_user.username)
|
|
|
|
assert "hashed_password" not in exported
|
|
assert "jira_api_token" not in exported
|
|
assert exported["must_change_password"] is True
|
|
|
|
|
|
def test_import_config_rejects_invalid_json(client, auth_headers):
|
|
resp = client.post(
|
|
"/api/v1/admin/import-config",
|
|
data=b"not json",
|
|
headers={**auth_headers, "Content-Type": "application/json"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_import_config_creates_new_user_with_forced_reset(client, db, auth_headers):
|
|
resp = client.post(
|
|
"/api/v1/admin/import-config",
|
|
json={"users": [{"username": "imported_user", "role": "red_tech", "is_active": True}]},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["summary"]["users_created"] == 1
|
|
|
|
user = db.query(User).filter(User.username == "imported_user").first()
|
|
assert user is not None
|
|
assert user.must_change_password is True
|
|
assert user.role == "red_tech"
|
|
|
|
|
|
def test_import_config_updates_existing_user_role_only(client, db, auth_headers, red_tech_user):
|
|
original_hash = red_tech_user.hashed_password
|
|
|
|
resp = client.post(
|
|
"/api/v1/admin/import-config",
|
|
json={"users": [{"username": red_tech_user.username, "role": "red_lead", "is_active": True}]},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["summary"]["users_updated"] == 1
|
|
|
|
db.refresh(red_tech_user)
|
|
assert red_tech_user.role == "red_lead"
|
|
assert red_tech_user.hashed_password == original_hash
|
|
|
|
|
|
def test_import_config_skips_redacted_system_config_value(client, db, auth_headers):
|
|
db.add(SystemConfig(key="smtp.password", value="original-real-value"))
|
|
db.commit()
|
|
|
|
client.post(
|
|
"/api/v1/admin/import-config",
|
|
json={"system_configs": [{"key": "smtp.password", "value": "[REDACTED]"}]},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
row = db.query(SystemConfig).filter(SystemConfig.key == "smtp.password").first()
|
|
assert row.value == "original-real-value"
|
|
|
|
|
|
def test_import_config_never_restores_sso_private_key(client, db, auth_headers):
|
|
resp = client.post(
|
|
"/api/v1/admin/import-config",
|
|
json={"sso": {"is_enabled": True, "provider_name": "Okta", "sp_private_key": "[REDACTED]"}},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
row = db.query(SsoConfig).first()
|
|
assert row is not None
|
|
assert row.sp_private_key is None
|
|
|
|
|
|
def test_import_config_upserts_scoring_weights(client, db, auth_headers):
|
|
db.add(ScoringConfig(weight_tests=40.0, weight_detection_rules=25.0, weight_d3fend=15.0, weight_recency=10.0, weight_severity=10.0))
|
|
db.commit()
|
|
|
|
client.post(
|
|
"/api/v1/admin/import-config",
|
|
json={"scoring": {"weight_tests": 55.0, "weight_detection_rules": 20.0, "weight_d3fend": 10.0, "weight_recency": 10.0, "weight_severity": 5.0}},
|
|
headers=auth_headers,
|
|
)
|
|
|
|
row = db.query(ScoringConfig).first()
|
|
assert row.weight_tests == 55.0
|
|
|
|
|
|
def test_import_config_round_trip_is_idempotent(client, db, auth_headers):
|
|
db.add(WebhookConfig(name="alert-hook", url="https://example.com/hook", events=["test_validated"], is_active=True))
|
|
db.commit()
|
|
|
|
exported = client.get("/api/v1/admin/export-config", headers=auth_headers).json()
|
|
|
|
resp = client.post("/api/v1/admin/import-config", json=exported, headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
|
|
hooks = db.query(WebhookConfig).filter(WebhookConfig.name == "alert-hook").all()
|
|
assert len(hooks) == 1
|