test(admin): cover config export/import bundle (Block 4)
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
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
The "Export Configuration" admin feature (GET/POST /admin/export-config, /admin/import-config) was already fully implemented — admin-gated, redacts secrets, scoped to genuine config (settings/webhooks/SSO/scoring/ custom templates/users) rather than bulk data — but had zero test coverage for an admin-sensitive data-migration endpoint. Adds coverage for: admin-only gating on both endpoints, secret redaction (SMTP password, SSO private key), custom-template-only scoping, safe user import (forced password reset, no secret restoration from "[REDACTED]" placeholders), and round-trip import idempotency. All pass against the existing implementation — no bugs found.
This commit is contained in:
@@ -0,0 +1,170 @@
|
||||
"""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_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
|
||||
Reference in New Issue
Block a user