1055 lines
46 KiB
Python
1055 lines
46 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import SQLModel, Session, create_engine
|
|
|
|
|
|
def _reset_db(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "ai_security_test.db"
|
|
monkeypatch.setenv("APPOPS_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("APP_OPS_VOICE_TOKEN", "test-voice-token")
|
|
|
|
from backend.app.core.config import settings
|
|
from backend.app.db import session
|
|
from backend.app.db.init_db import init_db
|
|
|
|
previous = {
|
|
"apis": dict(settings.apis),
|
|
"environment": settings.environment,
|
|
"secret_key": settings.secret_key,
|
|
"secret_key_source": settings.secret_key_source,
|
|
}
|
|
for key in settings.apis:
|
|
settings.apis[key] = None if key.endswith(("_URL", "_API_KEY", "_TOKEN", "_SECRET", "_PASSWORD")) else settings.apis[key]
|
|
settings.environment = "development"
|
|
settings.secret_key = "test-ai-security-key"
|
|
settings.secret_key_source = "env"
|
|
|
|
engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False})
|
|
session.engine = engine
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
init_db()
|
|
return previous
|
|
|
|
|
|
def _restore_config(previous):
|
|
from backend.app.core.config import settings
|
|
|
|
settings.apis.clear()
|
|
settings.apis.update(previous["apis"])
|
|
settings.environment = previous["environment"]
|
|
settings.secret_key = previous["secret_key"]
|
|
settings.secret_key_source = previous["secret_key_source"]
|
|
|
|
|
|
def _client(tmp_path, monkeypatch):
|
|
previous = _reset_db(tmp_path, monkeypatch)
|
|
from backend.app.main import app
|
|
|
|
return TestClient(app), previous
|
|
|
|
|
|
def _npm_exposure():
|
|
return {
|
|
"success": True,
|
|
"status": "ok",
|
|
"totalProxyHosts": 2,
|
|
"enabledProxyHosts": 2,
|
|
"publicHosts": 2,
|
|
"hostsWithoutCertificate": 1,
|
|
"hostsWithExpiredCertificate": 0,
|
|
"hostsWithExpiringCertificate": 1,
|
|
"hostsWithoutAccessList": 1,
|
|
"hostsWithoutSslForce": 1,
|
|
"hostsWithWebsocket": 1,
|
|
"warnings": [
|
|
{"type": "host_without_certificate", "severity": "high", "title": "open.example.com has no certificate", "evidence": ["open.example.com"], "suggestedAction": "review_certificate"},
|
|
{"type": "certificate_expiring_soon", "severity": "warning", "title": "plex.example.com certificate expires soon", "evidence": ["12"], "suggestedAction": "review_certificate_in_npm"},
|
|
],
|
|
}
|
|
|
|
|
|
def _npm_expiring_certificate_exposure():
|
|
return {
|
|
**_npm_exposure(),
|
|
"warnings": [
|
|
{
|
|
"type": "certificate_expiring_soon",
|
|
"severity": "warning",
|
|
"title": "auth.example.com certificate expires soon",
|
|
"evidence": ["auth.example.com", "30 days", "2026-08-07 07:13:42"],
|
|
"suggestedAction": "review_certificate_in_npm",
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def _auth_coverage():
|
|
return {
|
|
"success": True,
|
|
"status": "ok",
|
|
"summary": {"totalPublicHosts": 2, "protectedHosts": 1, "likelyProtectedHosts": 0, "unprotectedHosts": 1, "npmStatus": "ok"},
|
|
"publicHosts": [
|
|
{"hostId": 1, "domains": ["plex.example.com"], "target": "http://plex:32400", "coverageStatus": "protected", "authentikMatches": ["plex.example.com"]},
|
|
{"hostId": 2, "domains": ["open.example.com"], "target": "http://unknown:8080", "coverageStatus": "unprotected", "authentikMatches": []},
|
|
],
|
|
}
|
|
|
|
|
|
def _media_snapshot_with_stalled_download():
|
|
return {
|
|
"sonarrHealth": {"success": True, "status": "connected"},
|
|
"sonarrWanted": {"success": True, "status": "connected", "total": 0, "items": []},
|
|
"sonarrQueue": {"success": True, "status": "connected", "total": 0, "items": []},
|
|
"prowlarrHealth": {"success": True, "status": "connected"},
|
|
"prowlarrIndexers": {"success": True, "status": "ok", "total": 1, "enabled": 1, "failing": 0, "items": []},
|
|
"prowlarrApplications": {"success": True, "status": "ok", "total": 1, "items": []},
|
|
"prowlarrSync": {"success": True, "status": "ok", "indexersFailing": 0},
|
|
"delugeHealth": {"success": True, "status": "connected"},
|
|
"delugeSession": {"success": True, "status": "ok", "active": 0, "paused": 0, "stalled": 2},
|
|
"delugeActive": {"success": True, "status": "ok", "total": 0, "items": []},
|
|
"delugeStalled": {"success": True, "status": "ok", "total": 2, "items": [{"id": "stalled-1", "name": "Example"}]},
|
|
"delugePaused": {"success": True, "status": "ok", "total": 0, "items": []},
|
|
"delugeSpeeds": {"success": True, "status": "ok", "active": 0, "paused": 0, "stalled": 2},
|
|
"plexHealth": {"success": True, "status": "connected"},
|
|
"plexLibraries": {"success": True, "status": "ok", "total": 1, "libraries": []},
|
|
"plexSessions": {"success": True, "status": "ok", "total": 0, "sessions": []},
|
|
"plexRecent": {"success": True, "status": "ok", "total": 0, "items": []},
|
|
"plexStreams": {"success": True, "status": "ok", "total": 0, "transcoding": 0},
|
|
"tautulliHealth": {"success": True, "status": "connected"},
|
|
"tautulliActivity": {"success": True, "status": "ok", "streamCount": 0, "sessions": []},
|
|
"tautulliHistory": {"success": True, "status": "ok", "total": 0, "items": []},
|
|
"tautulliStats": {"success": True, "status": "ok", "totalPlays": 0},
|
|
"tautulliStreams": {"success": True, "status": "ok", "activeStreams": 0, "transcoding": 0},
|
|
"tautulliLibraries": {"success": True, "status": "ok", "total": 1, "items": []},
|
|
"widefrogHealth": {"success": True, "status": "connected"},
|
|
"widefrogQueue": {"success": True, "status": "ok", "total": 0, "items": []},
|
|
"widefrogFailed": {"success": True, "status": "ok", "failed": 0, "items": []},
|
|
"widefrogActivity": {"success": True, "status": "ok", "recent": []},
|
|
"widefrogDownloads": {"success": True, "status": "ok", "active": 0, "items": []},
|
|
}
|
|
|
|
|
|
def test_ai_security_summary_generated_with_no_data_is_partial(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
with client:
|
|
payload = client.get("/api/ai/security/summary").json()
|
|
|
|
assert payload["status"] in {"partial", "warning"}
|
|
assert 0 <= payload["score"] <= 100
|
|
assert payload["dataCompleteness"]["percentage"] < 100
|
|
assert "secret_storage" in payload["dataCompleteness"]["connectedSources"]
|
|
assert payload["criticalFindings"] == 0
|
|
assert payload["scoreBreakdown"]["finalScore"] == payload["score"]
|
|
assert set(payload["scoreBreakdown"]["missingSources"]) >= {"npm", "authentik", "unraid"}
|
|
assert payload["scoreBreakdown"]["coverageStatus"] in {"partial", "warning"}
|
|
assert payload["scoreBreakdown"]["coverageSources"]["secret_storage"]["status"] == "connected"
|
|
assert payload["scoreBreakdown"]["coverageSources"]["npm"]["status"] == "missing_config"
|
|
assert payload["scoreBreakdown"]["coverageSources"]["authentik"]["status"] == "missing_config"
|
|
assert payload["scoreBreakdown"]["coverageSources"]["unraid"]["status"] == "missing_config"
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_missing_npm_and_auth_does_not_infer_unprotected_public_hosts(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
summary = client.get("/api/ai/security/summary").json()
|
|
|
|
finding_ids = {finding["id"] for finding in findings}
|
|
assert "ai-sec-authentik-missing-public-hosts" not in finding_ids
|
|
assert "ai-sec-auth-public-host-without-authentik" not in finding_ids
|
|
assert summary["status"] in {"partial", "warning"}
|
|
assert summary["criticalFindings"] == 0
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_npm_and_auth_coverage_findings(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary", lambda self: _npm_exposure(), raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.proxy_hosts", lambda self: {"success": True, "items": [{"id": 2, "domainNames": ["open.example.com"], "enabled": True}]}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.auth_coverage", lambda self: _auth_coverage(), raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.security_warnings", lambda self: {"success": True, "items": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.outpost_health", lambda self: {"success": True, "available": False, "embeddedExpected": True, "items": []}, raising=True)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
types = {finding["id"] for finding in findings}
|
|
assert "ai-sec-npm-host-without-certificate" in types
|
|
assert "ai-sec-npm-certificate-expiring-soon" in types
|
|
assert "ai-sec-auth-public-host-without-authentik" in types
|
|
assert all(action["executable"] is False for finding in findings for action in finding["suggestedActions"])
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_npm_certificate_context_stays_evidence_not_affected(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary", lambda self: _npm_expiring_certificate_exposure(), raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.proxy_hosts", lambda self: {"success": True, "items": [{"id": 2, "domainNames": ["auth.example.com"], "enabled": True}]}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.auth_coverage", lambda self: {**_auth_coverage(), "publicHosts": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.security_warnings", lambda self: {"success": True, "items": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.outpost_health", lambda self: {"success": True, "available": True, "embeddedExpected": True, "items": []}, raising=True)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
finding = next(item for item in findings if item["id"] == "ai-sec-npm-certificate-expiring-soon")
|
|
assert finding["affectedResources"] == ["auth.example.com"]
|
|
assert finding["evidence"][0]["value"] == ["auth.example.com", "30 days", "2026-08-07 07:13:42"]
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_snapshot_runs_external_reads_in_parallel(monkeypatch, tmp_path):
|
|
from threading import Barrier
|
|
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
barrier = Barrier(3, timeout=1)
|
|
|
|
def wait_payload(payload):
|
|
def _inner(self):
|
|
barrier.wait()
|
|
return payload
|
|
|
|
return _inner
|
|
|
|
monkeypatch.setattr("backend.app.services.ai_security_reviewer._REVIEW_CACHE", {}, raising=False)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary", wait_payload(_npm_exposure()), raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.proxy_hosts", lambda self: {"success": True, "items": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.auth_coverage", wait_payload(_auth_coverage()), raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.security_warnings", lambda self: {"success": True, "items": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.outpost_health", lambda self: {"success": True, "available": True, "embeddedExpected": True, "items": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.unraid.UnraidConnector.test_connection", wait_payload({"success": True, "status": "connected"}), raising=True)
|
|
|
|
with client:
|
|
payload = client.get("/api/ai/security/summary").json()
|
|
|
|
assert payload["scoreBreakdown"]["coverageSources"]["npm"]["status"] == "connected"
|
|
assert payload["scoreBreakdown"]["coverageSources"]["authentik"]["status"] == "connected"
|
|
assert payload["scoreBreakdown"]["coverageSources"]["unraid"]["status"] == "connected"
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_snapshot_reuses_npm_and_auth_instances(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.connectors.authentik import AuthentikConnector
|
|
from backend.app.connectors.nginx_proxy_manager import NginxProxyManagerConnector
|
|
from backend.app.core.config import settings
|
|
from backend.app.services import ai_security_reviewer
|
|
|
|
settings.apis["NPM_URL"] = "https://npm.example.test"
|
|
settings.apis["NPM_IDENTITY"] = "admin@example.test"
|
|
settings.apis["NPM_SECRET"] = "NPM_SECRET_VALUE"
|
|
settings.apis["AUTHENTIK_URL"] = "https://auth.example.test"
|
|
settings.apis["AUTHENTIK_TOKEN"] = "AUTHENTIK_TOKEN_VALUE"
|
|
settings.apis["AUTHENTIK_EXPECT_EMBEDDED_OUTPOST"] = True
|
|
|
|
npm_calls = []
|
|
auth_calls = []
|
|
|
|
def npm_request(self, method, path, **kwargs):
|
|
npm_calls.append(path)
|
|
return {
|
|
"/nginx/proxy-hosts": [
|
|
{
|
|
"id": 1,
|
|
"domain_names": ["media.example.com"],
|
|
"forward_host": "plex",
|
|
"forward_port": 32400,
|
|
"forward_scheme": "http",
|
|
"enabled": True,
|
|
"ssl_forced": True,
|
|
"hsts_enabled": True,
|
|
"certificate_id": 10,
|
|
"access_list_id": 2,
|
|
},
|
|
],
|
|
"/nginx/certificates": [{"id": 10, "domain_names": ["media.example.com"], "expires_on": "2099-01-01T00:00:00Z"}],
|
|
"/nginx/access-lists": [{"id": 2, "name": "Authentik"}],
|
|
}[path]
|
|
|
|
def auth_paged(self, path, params=None):
|
|
auth_calls.append(path)
|
|
return {
|
|
"/core/applications/": [{"pk": "app-media", "name": "Media", "slug": "media", "launch_url": "https://media.example.com", "provider": 101}],
|
|
"/core/providers/": [{"pk": 101, "name": "Media proxy", "type": "authentik_providers_proxy.proxyprovider", "assigned_application_slug": "media"}],
|
|
"/providers/proxy/": [{"pk": 101, "name": "Media proxy", "external_host": "https://media.example.com", "assigned_application_slug": "media"}],
|
|
"/providers/oauth2/": [],
|
|
"/outposts/instances/": [{"pk": "outpost-1", "name": "Embedded Outpost", "health": "healthy", "providers": [101]}],
|
|
"/events/events/": [{"pk": "evt-1", "action": "login", "created": "2026-01-01T10:00:00Z"}],
|
|
}[path]
|
|
|
|
monkeypatch.setattr(NginxProxyManagerConnector, "_login", lambda self: "runtime-token", raising=True)
|
|
monkeypatch.setattr(NginxProxyManagerConnector, "_request", npm_request, raising=True)
|
|
monkeypatch.setattr(AuthentikConnector, "_paged", auth_paged, raising=True)
|
|
ai_security_reviewer._clear_review_cache()
|
|
|
|
with client:
|
|
payload = client.get("/api/ai/security/summary").json()
|
|
|
|
assert payload["scoreBreakdown"]["coverageSources"]["npm"]["status"] == "connected"
|
|
assert payload["scoreBreakdown"]["coverageSources"]["authentik"]["status"] == "connected"
|
|
assert npm_calls.count("/nginx/proxy-hosts") <= 2
|
|
assert npm_calls.count("/nginx/certificates") == 1
|
|
assert npm_calls.count("/nginx/access-lists") == 1
|
|
assert auth_calls.count("/core/applications/") == 1
|
|
assert auth_calls.count("/core/providers/") == 1
|
|
assert auth_calls.count("/providers/proxy/") == 1
|
|
assert auth_calls.count("/outposts/instances/") == 1
|
|
assert auth_calls.count("/events/events/") == 1
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_reuses_persistent_review_after_memory_cache_loss(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
with client:
|
|
first = client.get("/api/ai/security/summary").json()
|
|
|
|
from backend.app.services import ai_security_reviewer
|
|
|
|
ai_security_reviewer._clear_review_cache()
|
|
|
|
def fail_snapshot(session):
|
|
raise AssertionError("persistent AI Security cache should avoid a cold snapshot")
|
|
|
|
monkeypatch.setattr(ai_security_reviewer, "_snapshot", fail_snapshot)
|
|
|
|
with client:
|
|
second = client.get("/api/ai/security/summary").json()
|
|
|
|
assert second["generatedAt"] == first["generatedAt"]
|
|
assert second["score"] == first["score"]
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_persistent_review_ignores_other_dashboard_cache_rows(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
with client:
|
|
first = client.get("/api/ai/security/summary").json()
|
|
safety = client.get("/api/safety/dashboard").json()
|
|
|
|
assert safety["cache"]["source"] in {"fresh", "cache"}
|
|
|
|
from backend.app.services import ai_security_reviewer
|
|
|
|
ai_security_reviewer._clear_review_cache()
|
|
|
|
def fail_snapshot(session):
|
|
raise AssertionError("dashboard cache rows should not invalidate AI Security persistent cache")
|
|
|
|
monkeypatch.setattr(ai_security_reviewer, "_snapshot", fail_snapshot)
|
|
|
|
with client:
|
|
second = client.get("/api/ai/security/summary").json()
|
|
|
|
assert second["generatedAt"] == first["generatedAt"]
|
|
assert second["score"] == first["score"]
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_uses_persistent_media_summary_cache(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.session import engine
|
|
from backend.app.services import ai_security_reviewer, media_pipeline_service
|
|
|
|
media_pipeline_service._SNAPSHOT_CACHE.clear()
|
|
monkeypatch.setattr(media_pipeline_service, "_build_snapshot", lambda: _media_snapshot_with_stalled_download())
|
|
with Session(engine) as session:
|
|
cached = media_pipeline_service.summary(session, force_refresh=True)
|
|
|
|
assert cached["status"] == "warning"
|
|
assert cached["counts"]["stalledDownloads"] == 2
|
|
|
|
media_pipeline_service._SNAPSHOT_CACHE.clear()
|
|
ai_security_reviewer._clear_review_cache()
|
|
|
|
def fail_media_snapshot():
|
|
raise AssertionError("AI Security should reuse the persistent Media Pipeline summary cache")
|
|
|
|
monkeypatch.setattr(media_pipeline_service, "_build_snapshot", fail_media_snapshot)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=media").json()
|
|
|
|
media_finding = next(item for item in findings if item["id"] == "ai-sec-media-pipeline-issue")
|
|
assert media_finding["affectedResources"] == ["deluge"]
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_access_list_warning_honors_authentik_coverage(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary",
|
|
lambda self: {
|
|
"success": True,
|
|
"status": "ok",
|
|
"warnings": [
|
|
{
|
|
"type": "public_host_without_access_list",
|
|
"severity": "high",
|
|
"title": "protected.example.com has no access list",
|
|
"evidence": ["protected.example.com", "http://protected-app:8080"],
|
|
"suggestedAction": "review_access_protection",
|
|
},
|
|
{
|
|
"type": "public_host_without_access_list",
|
|
"severity": "high",
|
|
"title": "open.example.com has no access list",
|
|
"evidence": ["open.example.com", "http://open-app:8081"],
|
|
"suggestedAction": "review_access_protection",
|
|
},
|
|
],
|
|
},
|
|
raising=True,
|
|
)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.proxy_hosts", lambda self: {"success": True, "items": []}, raising=True)
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.authentik.AuthentikConnector.auth_coverage",
|
|
lambda self: {
|
|
"success": True,
|
|
"status": "ok",
|
|
"publicHosts": [
|
|
{"hostId": 1, "domains": ["protected.example.com"], "coverageStatus": "protected"},
|
|
{"hostId": 2, "domains": ["open.example.com"], "coverageStatus": "unprotected"},
|
|
],
|
|
},
|
|
raising=True,
|
|
)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=exposure").json()
|
|
|
|
assert not any(finding["id"] == "ai-sec-npm-host-without-access-list" for finding in findings)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_authentik_missing_config_with_public_hosts(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.proxy_hosts", lambda self: {"success": True, "items": [{"id": 1, "domainNames": ["public.example.com"], "enabled": True}]}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary", lambda self: {**_npm_exposure(), "warnings": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.auth_coverage", lambda self: {"success": False, "status": "missing_config", "publicHosts": []}, raising=True)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=auth").json()
|
|
|
|
assert any(finding["id"] == "ai-sec-authentik-missing-public-hosts" for finding in findings)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_connected_npm_public_hosts_with_missing_auth_is_evidence_based(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.proxy_hosts", lambda self: {"success": True, "status": "ok", "items": [{"id": 1, "domainNames": ["public.example.com"], "enabled": True}]}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary", lambda self: {**_npm_exposure(), "warnings": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.auth_coverage", lambda self: {"success": False, "status": "missing_config", "publicHosts": []}, raising=True)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=auth").json()
|
|
|
|
finding = next(item for item in findings if item["id"] == "ai-sec-authentik-missing-public-hosts")
|
|
assert finding["severity"] in {"medium", "high"}
|
|
assert any(evidence["source"] == "npm" and evidence["label"] == "publicHosts" and evidence["value"] == 1 for evidence in finding["evidence"])
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_reported_embedded_outpost_with_unknown_health_is_not_unavailable(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.auth_coverage", lambda self: {"success": True, "status": "ok", "publicHosts": []}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.security_warnings", lambda self: {"success": True, "status": "warning", "items": [{"type": "embedded_outpost_unavailable", "severity": "warning", "title": "Expected embedded outpost is unavailable or unknown", "evidence": ["warning"]}]}, raising=True)
|
|
monkeypatch.setattr("backend.app.connectors.authentik.AuthentikConnector.outpost_health", lambda self: {"success": True, "status": "warning", "available": False, "embeddedExpected": True, "items": [{"id": "embedded", "health": "unknown"}]}, raising=True)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=auth").json()
|
|
|
|
assert not any(finding["id"] == "ai-sec-auth-outpost-unavailable" for finding in findings)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_critical_status_requires_critical_or_high_risk_evidence(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import Policy
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
policy = session.get(Policy, "policy-volume-delete")
|
|
policy.manual = "allowed"
|
|
session.add(policy)
|
|
session.commit()
|
|
|
|
with client:
|
|
payload = client.get("/api/ai/security/summary").json()
|
|
|
|
assert payload["status"] == "critical"
|
|
assert payload["criticalFindings"] >= 1
|
|
assert any(item.get("findingId") == "ai-sec-policy-volume-delete-not-blocked" for item in payload["scoreBreakdown"]["penalties"])
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_ignores_stale_changed_discovery_when_latest_is_managed(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import App, AppContainerMapping, DiscoveredContainer, DiscoveryRun
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
app = session.get(App, "nginx_proxy_manager")
|
|
if not app:
|
|
app = App(id="nginx_proxy_manager", name="Nginx Proxy Manager", category="Network")
|
|
app.criticality = "critical"
|
|
app.protected = True
|
|
app.needs_review = False
|
|
app.discovery_status = "managed"
|
|
session.add(app)
|
|
old_run = DiscoveryRun(
|
|
id="disc-old-npm",
|
|
status="complete",
|
|
started_at=datetime.now(UTC).replace(tzinfo=None) - timedelta(hours=1),
|
|
finished_at=datetime.now(UTC).replace(tzinfo=None) - timedelta(hours=1),
|
|
)
|
|
new_run = DiscoveryRun(
|
|
id="disc-new-npm",
|
|
status="complete",
|
|
started_at=datetime.now(UTC).replace(tzinfo=None),
|
|
finished_at=datetime.now(UTC).replace(tzinfo=None),
|
|
)
|
|
session.add(old_run)
|
|
session.add(new_run)
|
|
session.add(
|
|
DiscoveredContainer(
|
|
id="dc-old-npm",
|
|
discovery_run_id=old_run.id,
|
|
container_id="npm-container",
|
|
name="NginxProxyManager",
|
|
image="jlesage/nginx-proxy-manager",
|
|
state="running",
|
|
status="healthy",
|
|
suggested_app_type="nginx_proxy_manager",
|
|
suggested_category="network",
|
|
suggested_criticality="critical",
|
|
suggested_connector="nginx_proxy_manager",
|
|
suggested_policy_profile="reverse_proxy",
|
|
import_status="changed",
|
|
matched_app_id="nginx_proxy_manager",
|
|
)
|
|
)
|
|
session.add(
|
|
DiscoveredContainer(
|
|
id="dc-new-npm",
|
|
discovery_run_id=new_run.id,
|
|
container_id="npm-container",
|
|
name="NginxProxyManager",
|
|
image="jlesage/nginx-proxy-manager",
|
|
state="running",
|
|
status="healthy",
|
|
suggested_app_type="nginx_proxy_manager",
|
|
suggested_category="network",
|
|
suggested_criticality="critical",
|
|
suggested_connector="nginx_proxy_manager",
|
|
suggested_policy_profile="reverse_proxy",
|
|
import_status="managed",
|
|
matched_app_id="nginx_proxy_manager",
|
|
)
|
|
)
|
|
session.add(
|
|
AppContainerMapping(
|
|
id="map-npm-container",
|
|
app_id="nginx_proxy_manager",
|
|
container_id="npm-container",
|
|
container_name="NginxProxyManager",
|
|
source="discovery",
|
|
is_primary=True,
|
|
last_seen_run_id=new_run.id,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
assert not any(finding["id"] == "ai-sec-inventory-critical-review-dc-old-npm" for finding in findings)
|
|
assert not any(
|
|
finding["category"] == "inventory" and "NginxProxyManager" in finding["title"]
|
|
for finding in findings
|
|
)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_ignores_appops_deployment_discovery_artifacts(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import DiscoveredContainer, DiscoveryRun
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
run = DiscoveryRun(
|
|
id="disc-appops-deploy-artifacts",
|
|
status="complete",
|
|
started_at=datetime.now(UTC).replace(tzinfo=None),
|
|
finished_at=datetime.now(UTC).replace(tzinfo=None),
|
|
)
|
|
session.add(run)
|
|
for suffix, name in (
|
|
("current", "unraid-appops-gateway-1216"),
|
|
("rollback", "unraid-appops-gateway-1216-pre-20260712T212606Z"),
|
|
):
|
|
session.add(DiscoveredContainer(
|
|
id=f"dc-appops-{suffix}",
|
|
discovery_run_id=run.id,
|
|
container_id=f"appops-{suffix}",
|
|
name=name,
|
|
image="unraid-appops-gateway:test",
|
|
state="exited" if suffix == "rollback" else "running",
|
|
status="healthy",
|
|
suggested_app_type="unraid_appops_gateway",
|
|
suggested_category="infrastructure",
|
|
suggested_criticality="critical",
|
|
suggested_connector="none",
|
|
suggested_policy_profile="protected_service",
|
|
import_status="discovered",
|
|
matched_app_id="unraid_appops_gateway_1216",
|
|
))
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
assert not any(
|
|
finding["category"] == "inventory" and "unraid-appops-gateway" in finding["title"].lower()
|
|
for finding in findings
|
|
)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_ignores_changed_discovery_for_protected_mapped_app(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import App, DiscoveredContainer, DiscoveryRun
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
app = session.get(App, "nginx_proxy_manager")
|
|
if not app:
|
|
app = App(id="nginx_proxy_manager", name="Nginx Proxy Manager", category="Network")
|
|
app.criticality = "critical"
|
|
app.protected = True
|
|
app.needs_review = False
|
|
app.discovery_status = "managed"
|
|
run = DiscoveryRun(id="disc-current-npm", status="complete", started_at=datetime.now(UTC).replace(tzinfo=None), finished_at=datetime.now(UTC).replace(tzinfo=None))
|
|
session.add(app)
|
|
session.add(run)
|
|
session.add(
|
|
DiscoveredContainer(
|
|
id="dc-current-npm",
|
|
discovery_run_id=run.id,
|
|
container_id="npm-container",
|
|
name="NginxProxyManager",
|
|
image="jlesage/nginx-proxy-manager",
|
|
state="running",
|
|
status="healthy",
|
|
suggested_app_type="nginx_proxy_manager",
|
|
suggested_category="network",
|
|
suggested_criticality="critical",
|
|
suggested_connector="nginx_proxy_manager",
|
|
suggested_policy_profile="reverse_proxy",
|
|
import_status="changed",
|
|
matched_app_id="nginx_proxy_manager",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
assert not any(
|
|
finding["category"] == "inventory" and "NginxProxyManager" in finding["title"]
|
|
for finding in findings
|
|
)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_critical_discovery_links_to_matched_app_detail(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import App, DiscoveredContainer, DiscoveryRun
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
app = session.get(App, "homeassistant")
|
|
if not app:
|
|
app = App(id="homeassistant", name="Home Assistant", category="Automation")
|
|
app.criticality = "critical"
|
|
app.protected = False
|
|
app.needs_review = True
|
|
session.add(app)
|
|
run = DiscoveryRun(id="disc-ha-review", status="complete", started_at=datetime.now(UTC).replace(tzinfo=None), finished_at=datetime.now(UTC).replace(tzinfo=None))
|
|
session.add(run)
|
|
session.add(
|
|
DiscoveredContainer(
|
|
id="dc-ha-review",
|
|
discovery_run_id=run.id,
|
|
container_id="ha-container",
|
|
name="homeassistant",
|
|
image="ghcr.io/home-assistant/home-assistant",
|
|
state="running",
|
|
status="healthy",
|
|
suggested_app_type="home_assistant",
|
|
suggested_category="automation",
|
|
suggested_criticality="critical",
|
|
suggested_connector="home_assistant",
|
|
suggested_policy_profile="home_automation",
|
|
import_status="changed",
|
|
matched_app_id="homeassistant",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
finding = next(item for item in findings if item["id"] == "ai-sec-inventory-critical-review-dc-ha-review")
|
|
hrefs = {action["href"] for action in finding["suggestedActions"]}
|
|
assert "/apps/homeassistant" in hrefs
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_ignores_discovered_row_when_reviewed_canonical_app_exists(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import App, DiscoveredContainer, DiscoveryRun
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
app = session.get(App, "home_assistant")
|
|
if not app:
|
|
app = App(id="home_assistant", name="Home Assistant", category="Infrastructure")
|
|
app.criticality = "medium"
|
|
app.protected = False
|
|
app.needs_review = False
|
|
app.discovery_status = "managed"
|
|
app.connector_status = "connected"
|
|
session.add(app)
|
|
run = DiscoveryRun(id="disc-ha-stale", status="complete", started_at=datetime.now(UTC).replace(tzinfo=None), finished_at=datetime.now(UTC).replace(tzinfo=None))
|
|
session.add(run)
|
|
session.add(
|
|
DiscoveredContainer(
|
|
id="dc-ha-stale",
|
|
discovery_run_id=run.id,
|
|
container_id="ha-container",
|
|
name="homeassistant",
|
|
image="ghcr.io/home-assistant/home-assistant",
|
|
state="running",
|
|
status="healthy",
|
|
suggested_app_type="home_assistant",
|
|
suggested_category="automation",
|
|
suggested_criticality="critical",
|
|
suggested_connector="home_assistant",
|
|
suggested_policy_profile="home_automation",
|
|
import_status="discovered",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
assert not any(finding["id"] == "ai-sec-inventory-critical-review-dc-ha-stale" for finding in findings)
|
|
assert not any(
|
|
finding["category"] == "inventory" and "homeassistant" in finding["title"].lower()
|
|
for finding in findings
|
|
)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_review_refresh_replaces_cached_findings(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import App, DiscoveredContainer, DiscoveryRun
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
app = session.get(App, "home_assistant")
|
|
if not app:
|
|
app = App(id="home_assistant", name="Home Assistant", category="Infrastructure")
|
|
app.criticality = "medium"
|
|
app.needs_review = True
|
|
app.discovery_status = "managed"
|
|
session.add(app)
|
|
run = DiscoveryRun(id="disc-ha-cache", status="complete", started_at=datetime.now(UTC).replace(tzinfo=None), finished_at=datetime.now(UTC).replace(tzinfo=None))
|
|
session.add(run)
|
|
session.add(
|
|
DiscoveredContainer(
|
|
id="dc-ha-cache",
|
|
discovery_run_id=run.id,
|
|
container_id="ha-cache-container",
|
|
name="homeassistant",
|
|
image="ghcr.io/home-assistant/home-assistant",
|
|
state="running",
|
|
status="healthy",
|
|
suggested_app_type="home_assistant",
|
|
suggested_category="automation",
|
|
suggested_criticality="critical",
|
|
suggested_connector="home_assistant",
|
|
suggested_policy_profile="home_automation",
|
|
import_status="discovered",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with client:
|
|
before = client.get("/api/ai/security/findings").json()
|
|
|
|
assert any(finding["id"] == "ai-sec-inventory-critical-review-dc-ha-cache" for finding in before)
|
|
|
|
with Session(engine) as session:
|
|
app = session.get(App, "home_assistant")
|
|
app.needs_review = False
|
|
session.add(app)
|
|
session.commit()
|
|
|
|
with client:
|
|
refreshed = client.post("/api/ai/security/review").json()
|
|
after = client.get("/api/ai/security/findings").json()
|
|
|
|
assert not any(finding["id"] == "ai-sec-inventory-critical-review-dc-ha-cache" for finding in refreshed["findings"])
|
|
assert not any(finding["id"] == "ai-sec-inventory-critical-review-dc-ha-cache" for finding in after)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_secret_policy_audit_and_media_findings(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import AuditEvent, Policy
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
policy = session.get(Policy, "policy-unraid-mover")
|
|
policy.voice = "allowed"
|
|
session.add(policy)
|
|
session.add(AuditEvent(id="audit-block-1", source="voice", identity="ha", action_id="container.restart", target="authentik", risk="high", policy_decision="blocked", result="failed"))
|
|
session.add(AuditEvent(id="audit-block-2", source="voice", identity="ha", action_id="container.restart", target="authentik", risk="high", policy_decision="blocked", result="failed"))
|
|
session.add(AuditEvent(id="audit-block-3", source="voice", identity="ha", action_id="container.restart", target="authentik", risk="high", policy_decision="blocked", result="failed"))
|
|
session.commit()
|
|
|
|
monkeypatch.setattr("backend.app.services.media_pipeline_service.summary", lambda session: {"success": True, "status": "warning", "issues": [{"id": "deluge-stalled", "severity": "warning", "title": "Deluge stalled downloads", "description": "2 stalled downloads", "affected": ["deluge"]}], "connectors": {"deluge": {"status": "connected"}}}, raising=False)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
recommendations = client.get("/api/ai/security/recommendations").json()
|
|
|
|
assert any(finding["id"] == "ai-sec-policy-unraid-mover-voice-allowed" for finding in findings)
|
|
assert any(finding["id"] == "ai-sec-audit-repeated-blocked-attempts" for finding in findings)
|
|
assert any(finding["category"] == "connector_config" and "missing" in finding["title"].lower() for finding in findings)
|
|
media_finding = next(finding for finding in findings if finding["id"] == "ai-sec-media-pipeline-issue")
|
|
assert "Deluge stalled downloads" in media_finding["title"]
|
|
assert media_finding["affectedResources"] == ["deluge"]
|
|
assert any(recommendation["id"] == "rec-ai-sec-media-pipeline-issue" for recommendation in recommendations)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_media_finding_uses_summary_counts_when_issues_missing(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.services.media_pipeline_service.summary",
|
|
lambda session: {
|
|
"success": True,
|
|
"status": "warning",
|
|
"message": "Media pipeline has items requiring attention.",
|
|
"counts": {"stalledDownloads": 4, "failedWidefrogJobs": 0, "failingIndexers": 0, "transcodes": 0},
|
|
"risk": {"level": "medium", "reasons": ["Deluge has stalled downloads"]},
|
|
"connectors": {"deluge": {"status": "connected"}},
|
|
},
|
|
raising=False,
|
|
)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings").json()
|
|
|
|
media_finding = next(finding for finding in findings if finding["id"] == "ai-sec-media-pipeline-issue")
|
|
assert media_finding["affectedResources"] == ["deluge"]
|
|
assert media_finding["evidence"][1]["value"] == ["Deluge has stalled downloads"]
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_missing_secret_actions_deep_link_to_connector_setup(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=secrets").json()
|
|
|
|
finding = next(item for item in findings if item["id"] == "ai-sec-missing-secret-portainer_api_key")
|
|
hrefs = {action["href"] for action in finding["suggestedActions"]}
|
|
assert "/setup?step=connectors&connector=portainer" in hrefs
|
|
assert "/connectors" not in hrefs
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_ignores_stale_setup_failures_after_later_success(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.db.models import AuditEvent
|
|
from backend.app.db.session import engine
|
|
|
|
with Session(engine) as session:
|
|
base = datetime.now(UTC).replace(tzinfo=None)
|
|
for index in range(3):
|
|
session.add(
|
|
AuditEvent(
|
|
id=f"audit-old-portainer-fail-{index}",
|
|
source="manual",
|
|
identity="setup",
|
|
action_id="setup.test.portainer",
|
|
target="portainer",
|
|
risk="low",
|
|
policy_decision="allowed",
|
|
result="failed",
|
|
timestamp=base - timedelta(minutes=10 - index),
|
|
)
|
|
)
|
|
session.add(
|
|
AuditEvent(
|
|
id="audit-new-portainer-success",
|
|
source="manual",
|
|
identity="setup",
|
|
action_id="setup.test.portainer",
|
|
target="portainer",
|
|
risk="low",
|
|
policy_decision="allowed",
|
|
result="success",
|
|
timestamp=base,
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=audit").json()
|
|
|
|
assert not any(finding["id"] == "ai-sec-audit-repeated-blocked-attempts" for finding in findings)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_lifecycle_routes_and_accept_does_not_execute(tmp_path, monkeypatch):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
with client:
|
|
review = client.post("/api/ai/security/review").json()
|
|
finding_id = review["findings"][0]["id"]
|
|
recommendation_id = review["recommendations"][0]["id"]
|
|
acknowledged = client.post(f"/api/ai/security/findings/{finding_id}/acknowledge").json()
|
|
dismissed = client.post(f"/api/ai/security/findings/{finding_id}/dismiss").json()
|
|
accepted = client.post(f"/api/ai/security/recommendations/{recommendation_id}/accept").json()
|
|
|
|
assert acknowledged["status"] == "acknowledged"
|
|
assert dismissed["status"] == "dismissed"
|
|
assert accepted["status"] == "accepted"
|
|
assert accepted["applied"] is False
|
|
assert accepted["recommendation"]["canAutoApply"] is False
|
|
assert all(action["executable"] is False for action in accepted["recommendation"]["suggestedActions"])
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_redacts_secret_values(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
secret = "AUTHENTIK_TOKEN_VALUE"
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.authentik.AuthentikConnector.security_warnings",
|
|
lambda self: {"success": True, "items": [{"type": "token_leak", "severity": "critical", "title": f"leaked {secret}", "evidence": [secret], "suggestedAction": "rotate"}]},
|
|
raising=True,
|
|
)
|
|
|
|
with client:
|
|
payload = client.get("/api/ai/security/findings").json()
|
|
|
|
assert secret not in str(payload)
|
|
assert "[REDACTED]" in str(payload)
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_does_not_require_optional_widefrog_api_key_when_connected(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
|
|
from backend.app.core.config import settings
|
|
from backend.app.db.models import Connector
|
|
from backend.app.db.session import engine
|
|
|
|
settings.apis["WIDEFROG_URL"] = "http://widefrog.local"
|
|
settings.apis["WIDEFROG_API_KEY"] = None
|
|
with Session(engine) as session:
|
|
connector = session.get(Connector, "widefrog")
|
|
connector.status = "connected"
|
|
connector.secret_ref = "WIDEFROG_API_KEY"
|
|
session.add(connector)
|
|
session.commit()
|
|
|
|
with client:
|
|
findings = client.get("/api/ai/security/findings?scope=secrets").json()
|
|
|
|
assert all(
|
|
finding["id"] != "ai-sec-missing-secret-widefrog_api_key"
|
|
for finding in findings
|
|
)
|
|
assert "WIDEFROG_API_KEY" not in str([finding["title"] for finding in findings])
|
|
|
|
_restore_config(previous)
|
|
|
|
|
|
def test_ai_security_score_breakdown_is_secret_safe(monkeypatch, tmp_path):
|
|
client, previous = _client(tmp_path, monkeypatch)
|
|
secret = "SUPER_SECRET_API_KEY_VALUE"
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.connectors.nginx_proxy_manager.NginxProxyManagerConnector.exposure_summary",
|
|
lambda self: {"success": False, "status": "missing_config", "message": f"missing {secret}"},
|
|
raising=True,
|
|
)
|
|
|
|
with client:
|
|
payload = client.get("/api/ai/security/summary").json()
|
|
|
|
assert "scoreBreakdown" in payload
|
|
assert secret not in str(payload["scoreBreakdown"])
|
|
assert "API_KEY_VALUE" not in str(payload["scoreBreakdown"])
|
|
|
|
_restore_config(previous)
|