kitos
504dfc52f5
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
Login is now by email, not username. username still exists internally (JWT sub claim, audit logs, Jira actor attribution, SSO provisioning all still key off it) but is now always kept equal to email everywhere a user is created or their email changes — never a separately-chosen value. - User.email is now unique + NOT NULL (migration b067 backfills any missing/blank email from username first, so existing rows — notably the seeded admin, which historically had none — never violate it). - /auth/login and the (unused but updated for consistency) authenticate_user() now query by email. - create_user (legacy, unreferenced but kept) and create_user_without_password both derive username from email. - update_user keeps username in sync when email changes, and rejects duplicate emails. - seed.py reads ADMIN_EMAIL (new env var, wired through install.sh and docker-compose.prod.yml) for the initial admin; falls back to an email-shaped ADMIN_USERNAME or a placeholder that's flagged for the operator to fix. - admin_config.py's import bundle now matches/creates users by email, skipping (not crashing on) entries with no email. - sso_service.py always sets username = email for SSO-provisioned users. - LoginPage/auth.ts updated to email input/copy (wire field name stays 'username' — that's the OAuth2PasswordRequestForm spec, not the value).
199 lines
7.8 KiB
Python
199 lines
7.8 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", "email": "imported_user@test.com", "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.email == "imported_user@test.com").first()
|
|
assert user is not None
|
|
assert user.must_change_password is True
|
|
assert user.role == "red_tech"
|
|
|
|
|
|
def test_import_config_skips_user_with_no_email(client, db, auth_headers):
|
|
resp = client.post(
|
|
"/api/v1/admin/import-config",
|
|
json={"users": [{"username": "no_email_user", "role": "red_tech", "is_active": True}]},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["summary"]["users_skipped_no_email"] == 1
|
|
assert db.query(User).filter(User.username == "no_email_user").first() is None
|
|
|
|
|
|
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, "email": red_tech_user.email, "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
|