from __future__ import annotations import base64 import contextlib import io import tempfile import unittest import urllib.error import urllib.request from dataclasses import replace from datetime import datetime, timezone from pathlib import Path from app.config import Config from app.metrics import PrometheusMetrics from app.state import RuntimeStats from app.store import AlertStore from app.webui import WebServer class _Status: def __init__(self, **values): self.values = values def status(self): return dict(self.values) class PrometheusMetricsTests(unittest.TestCase): def test_render_exports_raw_runtime_suricata_and_in_memory_component_state(self): stats = RuntimeStats() stats.inc("tzsp_datagrams", 5) stats.inc("frames_injected", 4) stats.stamp("last_packet_at") stats.update_suricata( { "capture": {"kernel_packets": 123, "kernel_drops": 2}, "detect": {"alert_queue_overflow": 1}, }, "2026-08-16T07:00:00+00:00", ) metrics = PrometheusMetrics( stats, version="1.2.3", mode="full", started_at=datetime(2026, 8, 16, 6, 0, tzinfo=timezone.utc), state_provider=lambda: { "components": {"suricata": True, "tzsp": True}, "features": {"auto_block": False}, }, flow_tracker=_Status( active_flows=3, max_flows=20000, published_updates=7, evicted_flows=1, parse_errors=2, throughput_samples=9, update_interval_seconds=2.0, traffic_counters={ "bytes_total": 10000, "bytes_in": 6000, "bytes_out": 3000, "bytes_internal": 750, "bytes_external": 250, "packets_total": 100, "packets_in": 60, "packets_out": 30, "packets_internal": 7, "packets_external": 3, }, ), ) rendered = metrics.render() self.assertIn('mikrosuricata_build_info{mode="full",version="1.2.3"} 1', rendered) self.assertIn("mikrosuricata_tzsp_datagrams_total 5", rendered) self.assertIn("mikrosuricata_frames_injected_total 4", rendered) self.assertIn("mikrosuricata_suricata_capture_kernel_packets 123", rendered) self.assertIn("mikrosuricata_suricata_capture_kernel_drops 2", rendered) self.assertIn('mikrosuricata_component_up{component="suricata"} 1', rendered) self.assertIn("mikrosuricata_flow_tracker_active_flows 3", rendered) self.assertIn('mikrosuricata_traffic_bytes_total{direction="total"} 10000', rendered) self.assertIn('mikrosuricata_traffic_bytes_total{direction="inbound"} 6000', rendered) self.assertIn('mikrosuricata_traffic_packets_total{direction="external"} 3', rendered) self.assertEqual(rendered.count("# TYPE mikrosuricata_traffic_bytes_total counter"), 1) self.assertNotIn("block_success_rate", rendered) self.assertNotIn("tzsp_to_tap_loss", rendered) self.assertEqual(rendered.count("# TYPE mikrosuricata_component_up gauge"), 1) def test_runtime_metrics_snapshot_has_no_derived_values(self): stats = RuntimeStats() stats.inc("tzsp_datagrams", 10) stats.inc("frames_injected", 9) raw = stats.metrics_snapshot() self.assertNotIn("tzsp_to_tap_loss", raw) self.assertNotIn("block_success_rate", raw) def test_metrics_endpoint_allows_default_loopback_acl_and_does_not_call_health_provider(self): with tempfile.TemporaryDirectory() as tmp: db_path = str(Path(tmp) / "ids.db") cfg = replace( Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path, admin_password="test-password", ) store = AlertStore(db_path) def forbidden_health(): raise AssertionError("/metrics must not call the health provider") web = WebServer( cfg, store, forbidden_health, metrics_provider=lambda: "# TYPE mikrosuricata_test gauge\nmikrosuricata_test 1\n", ) try: with contextlib.redirect_stdout(io.StringIO()): web.start() port = web.server.server_address[1] with urllib.request.urlopen(f"http://127.0.0.1:{port}/metrics", timeout=2) as response: body = response.read().decode("utf-8") content_type = response.headers.get("Content-Type", "") self.assertEqual(200, response.status) self.assertIn("version=0.0.4", content_type) self.assertIn("mikrosuricata_test 1", body) finally: web.stop() store.close() def test_metrics_ip_acl_denies_before_rendering_metrics(self): with tempfile.TemporaryDirectory() as tmp: db_path = str(Path(tmp) / "ids.db") cfg = replace( Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path, metrics_allowed_ips="192.0.2.10/32", ) store = AlertStore(db_path) calls = [] web = WebServer( cfg, store, lambda: {}, metrics_provider=lambda: calls.append(True) or "mikrosuricata_test 1\n", ) try: with contextlib.redirect_stdout(io.StringIO()): web.start() port = web.server.server_address[1] with self.assertRaises(urllib.error.HTTPError) as caught: urllib.request.urlopen(f"http://127.0.0.1:{port}/metrics", timeout=2) self.assertEqual(403, caught.exception.code) self.assertEqual([], calls) finally: web.stop() store.close() def test_metrics_basic_auth_requires_valid_credentials_after_ip_acl(self): with tempfile.TemporaryDirectory() as tmp: db_path = str(Path(tmp) / "ids.db") cfg = replace( Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path, metrics_allowed_ips="127.0.0.1", metrics_basic_auth_username="prometheus", metrics_basic_auth_password="strong-secret", ) store = AlertStore(db_path) calls = [] web = WebServer( cfg, store, lambda: {}, metrics_provider=lambda: calls.append(True) or "mikrosuricata_test 1\n", ) try: with contextlib.redirect_stdout(io.StringIO()): web.start() port = web.server.server_address[1] url = f"http://127.0.0.1:{port}/metrics" with self.assertRaises(urllib.error.HTTPError) as caught: urllib.request.urlopen(url, timeout=2) self.assertEqual(401, caught.exception.code) self.assertIn("Basic", caught.exception.headers.get("WWW-Authenticate", "")) self.assertEqual([], calls) bad = urllib.request.Request(url, headers={"Authorization": "Basic !!!"}) with self.assertRaises(urllib.error.HTTPError) as caught: urllib.request.urlopen(bad, timeout=2) self.assertEqual(401, caught.exception.code) self.assertEqual([], calls) token = base64.b64encode(b"prometheus:strong-secret").decode("ascii") request = urllib.request.Request(url, headers={"Authorization": f"Basic {token}"}) with urllib.request.urlopen(request, timeout=2) as response: body = response.read().decode("utf-8") self.assertEqual(200, response.status) self.assertIn("mikrosuricata_test 1", body) self.assertEqual([True], calls) finally: web.stop() store.close() def test_metrics_basic_auth_configuration_must_be_complete(self): with tempfile.TemporaryDirectory() as tmp: db_path = str(Path(tmp) / "ids.db") cfg = replace( Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path, metrics_basic_auth_username="prometheus", metrics_basic_auth_password="", ) store = AlertStore(db_path) try: with self.assertRaisesRegex(ValueError, "must either both be set"): WebServer(cfg, store, lambda: {}, metrics_provider=lambda: "") finally: store.close() def test_metrics_acl_rejects_invalid_network_configuration(self): with tempfile.TemporaryDirectory() as tmp: db_path = str(Path(tmp) / "ids.db") cfg = replace( Config.from_env(), web_bind="127.0.0.1", web_port=0, db_path=db_path, metrics_allowed_ips="not-an-ip", ) store = AlertStore(db_path) try: with self.assertRaisesRegex(ValueError, "invalid METRICS_ALLOWED_IPS"): WebServer(cfg, store, lambda: {}, metrics_provider=lambda: "") finally: store.close() if __name__ == "__main__": unittest.main()