"""Exposure preferences suppress ACL diagnostics without granting protection.""" from fastapi.testclient import TestClient import pytest from sqlmodel import Session, select from backend.app.db.models import AppSetting, AuditEvent, GatewayUser from backend.tests.test_nginx_proxy_manager_connector import _reset_db, _proxy_hosts, _certificates, _set_npm_config, _restore_config def _stored_preference(engine, enabled): with Session(engine) as db: db.add(AppSetting(key="exposure:preferences", value_json={"nginxAccessListChecksEnabled": enabled}, is_secret=False)) db.commit() def test_disabling_acl_checks_removes_only_acl_connector_warnings(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.connectors.nginx_proxy_manager import NginxProxyManagerConnector _stored_preference(session.engine, False) previous = _set_npm_config() calls = [] def request(_connector, method, path, **_kwargs): calls.append((method, path)) return {"/nginx/proxy-hosts": _proxy_hosts(), "/nginx/certificates": _certificates(), "/nginx/access-lists": []}[path] monkeypatch.setattr(NginxProxyManagerConnector, "_request", request) try: connector = NginxProxyManagerConnector() warnings = connector.security_warnings()["items"] summary = connector.exposure_summary() assert not any(item["type"] == "public_host_without_access_list" for item in warnings) assert not any(item["type"] == "public_host_without_access_list" for item in summary["warnings"]) assert {"host_without_certificate", "ssl_not_forced", "unknown_proxy_target"} <= {item["type"] for item in warnings} assert summary["hostsWithoutAccessList"] == 1 # Metadata remains truthful. assert ("GET", "/nginx/access-lists") not in calls assert all(method == "GET" for method, _path in calls) finally: _restore_config(previous) def test_disabled_acl_checks_do_not_mark_public_hosts_protected(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.api import routes_observability from backend.app.connectors.nginx_proxy_manager import NginxProxyManagerConnector _stored_preference(session.engine, False) monkeypatch.setattr(NginxProxyManagerConnector, "proxy_hosts", lambda _self: {"success": True, "items": [ {"id": 1, "domainNames": ["open.example.test"], "enabled": True, "certificateId": 1, "sslForced": True, "accessListId": None}, {"id": 2, "domainNames": ["sso.example.test"], "enabled": True, "certificateId": 1, "sslForced": True, "accessListId": None}, {"id": 3, "domainNames": ["acl.example.test"], "enabled": True, "certificateId": 1, "sslForced": True, "accessListId": 2}, ]}) monkeypatch.setattr(NginxProxyManagerConnector, "certificates", lambda _self: {"success": True, "items": [{"id": 1, "isExpired": False, "isExpiringSoon": False}]}) monkeypatch.setattr(routes_observability, "_auth_coverage_by_host", lambda: {"2": {"coverageStatus": "protected"}}) rows = routes_observability._build_network_exposure() assert rows[0]["protected"] is False assert rows[0]["risk"] == "high" assert all("access list" not in reason.lower() for reason in rows[0]["riskReasons"]) assert rows[1]["protected"] is True assert rows[1]["protectionSource"] == "authentik" assert rows[2]["protected"] is True assert rows[2]["protectionSource"] == "npm_access_list" def test_exposure_preferences_admin_write_is_strict_persistent_and_audited(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.main import app with TestClient(app) as client: assert client.get("/api/settings/exposure").json() == {"nginxAccessListChecksEnabled": True} for invalid in ({"nginxAccessListChecksEnabled": "false"}, {"nginxAccessListChecksEnabled": 0}, {"nginxAccessListChecksEnabled": False, "other": 1}, {}): assert client.patch("/api/settings/exposure", json=invalid).status_code == 422 response = client.patch("/api/settings/exposure", json={"nginxAccessListChecksEnabled": False}) assert response.status_code == 200 assert response.json() == {"nginxAccessListChecksEnabled": False} assert client.get("/api/settings/exposure").json() == response.json() with Session(session.engine) as db: row = db.get(AppSetting, "exposure:preferences") assert row and row.is_secret is False assert row.value_json == {"nginxAccessListChecksEnabled": False} events = db.exec(select(AuditEvent).where(AuditEvent.action_id == "settings.exposure.update")).all() assert len(events) == 1 assert events[0].before_state == {"nginxAccessListChecksEnabled": True} assert events[0].after_state == {"nginxAccessListChecksEnabled": False} def test_non_admin_with_settings_permission_cannot_change_exposure_preferences(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.main import app from backend.app.services import user_access user = GatewayUser(id="custom-settings", provider="authentik", issuer="test", subject="custom", role="custom", permissions=["settings.manage"]) monkeypatch.setattr(user_access, "current_user", lambda _request: user) with TestClient(app) as client: assert client.get("/api/settings/exposure").status_code == 200 assert client.patch("/api/settings/exposure", json={"nginxAccessListChecksEnabled": False}).status_code == 403 def test_corrupt_exposure_preference_keeps_acl_checks_enabled(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.services.exposure_preferences import get_exposure_preferences _stored_preference(session.engine, "false") with Session(session.engine) as db: assert get_exposure_preferences(db).nginxAccessListChecksEnabled is True def test_safety_signature_tracks_effective_preference_without_timestamp_change(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.services import safety_center_service _stored_preference(session.engine, True) with Session(session.engine) as db: before = safety_center_service._source_signature(db) row = db.get(AppSetting, "exposure:preferences") row.value_json = {"nginxAccessListChecksEnabled": False} db.add(row) db.commit() assert safety_center_service._source_signature(db) != before def test_safety_does_not_reuse_warnings_from_different_preferences_when_refresh_fails(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.services import safety_center_service monkeypatch.setattr(safety_center_service, "_dashboard_cache", { "signature": "previous-preferences", "expiresAt": "2099-01-01T00:00:00Z", "generatedAt": "2026-01-01T00:00:00Z", "payload": {"warnings": [{"type": "public_host_without_access_list"}]}, "generationDurationMs": 0, }) def unavailable(_session): raise RuntimeError("snapshot unavailable") monkeypatch.setattr(safety_center_service, "_payload", unavailable) with Session(session.engine) as db, pytest.raises(RuntimeError, match="snapshot unavailable"): safety_center_service.get_dashboard(db, force_refresh=True) def test_disabled_acl_checks_drop_stale_reviewer_acl_findings_only(tmp_path, monkeypatch): _reset_db(tmp_path, monkeypatch) from backend.app.db import session from backend.app.services import ai_security_reviewer _stored_preference(session.engine, False) data = {"npmExposure": {"success": True, "status": "ok", "warnings": [ {"type": "public_host_without_access_list", "title": "No ACL", "evidence": ["open.example.test"]}, {"type": "certificate_expired", "title": "Certificate expired", "evidence": ["open.example.test"]}, ]}, "authCoverage": {"status": "missing_config"}} with Session(session.engine) as db: findings = ai_security_reviewer._exposure_findings(db, data) assert [finding.id for finding in findings] == ["ai-sec-npm-certificate-expired"]