From 6ce04770b830a2a27332ab1393c0df26a825dd9f Mon Sep 17 00:00:00 2001 From: Danny Burrow Date: Fri, 10 Apr 2026 10:38:59 +0100 Subject: [PATCH] feat: add fluent-bit support for Loki log ingestion Loki log lines wrapped in a fluent-bit JSON envelope ({"log": "..."}) are now automatically detected and unwrapped. Server-side line filters use format-agnostic regex patterns so they match both promtail and fluent-bit escaped formats. Default --loki-query changed to {app_kubernetes_io_name="cilium-agent"} (fluent-bit); promtail users can override with --loki-query '{container="cilium-agent"}'. --- CHANGELOG.md | 11 +++ README.md | 6 +- hubble_audit2policy.py | 62 ++++++++++++-- pyproject.toml | 2 +- tests/test_flow_parsing.py | 165 ++++++++++++++++++++++++++++++++++--- 5 files changed, 221 insertions(+), 25 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a7f85ee..279a994 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,17 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.17.0] - 2026-04-10 + +### Added + +- fluent-bit support: Loki log lines wrapped in a `{"log": "..."}` JSON envelope are now automatically detected and unwrapped, so flows ingested via fluent-bit work out of the box alongside promtail. +- Default `--loki-query` changed from `{container="cilium-agent"}` (promtail) to `{app_kubernetes_io_name="cilium-agent"}` (fluent-bit). Promtail users can override with `--loki-query '{container="cilium-agent"}'`. + +### Fixed + +- Loki server-side line filters now use format-agnostic regex patterns (`verdict.{0,5}:.{0,5}AUDIT`) instead of literal JSON patterns (`"verdict":"AUDIT"`), so they match both promtail (plain quotes) and fluent-bit (escaped quotes) log formats. + ## [0.13.0] - 2026-04-02 ### Added diff --git a/README.md b/README.md index e611538..95d7f4d 100644 --- a/README.md +++ b/README.md @@ -88,10 +88,10 @@ hubble-audit2policy flows.json --report-only Query a Grafana Loki instance directly -- ideal when Hubble flows are already being shipped to Loki via fluentd, promtail, or another collector. -The default LogQL selector is `{container="cilium-agent"}`, which matches the standard Cilium agent container label. Override it with `--loki-query` if your setup uses different labels. +The default LogQL selector is `{app_kubernetes_io_name="cilium-agent"}`, which matches the standard fluent-bit Kubernetes label for the Cilium agent. For promtail setups use `--loki-query '{container="cilium-agent"}'`. Override with `--loki-query` if your setup uses different labels. ```bash -# All flows from the last hour (uses default query {container="cilium-agent"}): +# All flows from the last hour (uses default query {app_kubernetes_io_name="cilium-agent"}): hubble-audit2policy --from loki --loki-url http://loki:3100 --dry-run # Scoped to a namespace with a custom time window: @@ -239,7 +239,7 @@ hubble-audit2policy [-h] [-o OUTPUT_DIR] [-n NAMESPACE] | `--no-enrich` | Skip live cluster enrichment via Cilium endpoints | | `--from {file,loki}` | Flow source backend (default: `file`) | | `--loki-url URL` | Loki base URL, e.g. `http://loki:3100` | -| `--loki-query LOGQL` | LogQL stream selector (default: `{container="cilium-agent"}`) | +| `--loki-query LOGQL` | LogQL stream selector (default: `{app_kubernetes_io_name="cilium-agent"}`; for promtail: `{container="cilium-agent"}`) | | `--since DURATION` | How far back to query, e.g. `30m`, `2h`, `1d` (default: `1h`) | | `--until DURATION` | End of query window as duration before now (default: `0s` = now) | | `--loki-limit N` | Max entries per Loki request batch (default: `5000`) | diff --git a/hubble_audit2policy.py b/hubble_audit2policy.py index 1f09bd8..6862342 100755 --- a/hubble_audit2policy.py +++ b/hubble_audit2policy.py @@ -8,7 +8,7 @@ from __future__ import annotations -__version__ = "0.16.0" +__version__ = "0.17.0" __author__ = "noexecstack" __license__ = "Apache-2.0" @@ -453,26 +453,32 @@ def _loki_enrich_query( filters lines server-side. Multiple namespaces/verdicts are OR-joined into a single regex filter. + + The regex patterns use ``.{0,5}`` gaps between JSON keys, colons, and + values so they match both promtail-style lines (where the log text + contains ``"verdict":"AUDIT"``) and fluent-bit-style lines (where the + flow JSON is escaped inside a wrapper object, producing + ``\"verdict\":\"AUDIT\"``). """ # Verdict filter -- push as specific a match as possible so Loki # discards non-matching lines before sending them to us. verdict_list = sorted(set(verdicts)) if len(verdict_list) == 1: - query += r' |= "\"verdict\":\"' + verdict_list[0] + r'\""' + query += ' |~ "verdict.{0,5}:.{0,5}' + re.escape(verdict_list[0]) + '"' elif len(verdict_list) > 1: vpat = "|".join(re.escape(v) for v in verdict_list) - query += r' |~ "\"verdict\":\"(' + vpat + r')\""' + query += ' |~ "verdict.{0,5}:.{0,5}(' + vpat + ')"' else: # No specific verdict -- still filter to flow records only. - query += r' |= "\"verdict\":"' + query += ' |~ "verdict.{0,5}:"' ns_list = sorted(set(namespaces)) if not ns_list: return query if len(ns_list) == 1: - return query + r' |= "\"namespace\":\"' + ns_list[0] + r'\""' + return query + ' |~ "namespace.{0,5}:.{0,5}' + re.escape(ns_list[0]) + '"' pattern = "|".join(re.escape(ns) for ns in ns_list) - return query + r' |~ "\"namespace\":\"(' + pattern + r')\""' + return query + ' |~ "namespace.{0,5}:.{0,5}(' + pattern + ')"' class _LokiProgress: @@ -558,6 +564,40 @@ class _ChunkStats: failed: bool = False +_FLUENT_BIT_LOG_PREFIX = re.compile( + r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\s+\S+\s+\S+\s+" +) + + +def _extract_flow_from_loki_line(parsed: Any) -> dict[str, Any] | None: + """Return a Hubble flow dict from a parsed Loki log line. + + Handles two formats: + + * **Raw / promtail**: the log line is the Hubble JSON directly, e.g. + ``{"flow": {...}, ...}`` or ``{"time": ..., "verdict": ...}``. + * **fluent-bit envelope**: the log line is a JSON object with a ``"log"`` + key whose value is the original container stdout, e.g. + ``{"log": "2026-... stdout F {\"flow\":{...}}", "kubernetes": {...}}``. + The inner text is stripped of the timestamp/stream prefix and then + parsed as JSON. + + Returns ``None`` if no flow can be extracted. + """ + if not isinstance(parsed, dict): + return None + # fluent-bit format: outer envelope with a "log" string field. + if "log" in parsed and isinstance(parsed["log"], str): + inner = _FLUENT_BIT_LOG_PREFIX.sub("", parsed["log"], count=1).strip() + try: + parsed = json.loads(inner) + except (json.JSONDecodeError, ValueError): + return None + if not isinstance(parsed, dict): + return None + return cast("dict[str, Any]", parsed) + + def _loki_fetch_chunk( base_url: str, query: str, @@ -657,7 +697,10 @@ def _loki_fetch_chunk( continue ts_line_hashes.setdefault(ts, set()).add(line_h) try: - entries.append((ts, json.loads(line))) + parsed = json.loads(line) + flow = _extract_flow_from_loki_line(parsed) + if flow is not None: + entries.append((ts, flow)) except json.JSONDecodeError: pass # Non-JSON lines (e.g. cilium agent logs) are expected. @@ -2400,9 +2443,10 @@ def _build_parser() -> argparse.ArgumentParser: ) loki_group.add_argument( "--loki-query", - default='{container="cilium-agent"}', + default='{app_kubernetes_io_name="cilium-agent"}', metavar="LOGQL", - help='LogQL stream selector (default: {container="cilium-agent"})', + help='LogQL stream selector (default: {app_kubernetes_io_name="cilium-agent"})' + '; for promtail use {container="cilium-agent"}', ) loki_group.add_argument( "--since", diff --git a/pyproject.toml b/pyproject.toml index c1c2dfe..41a793d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "hubble-audit2policy" -version = "0.16.0" +version = "0.17.0" description = "Generate least-privilege CiliumNetworkPolicy YAML from Hubble flow logs." readme = "README.md" license = "Apache-2.0" diff --git a/tests/test_flow_parsing.py b/tests/test_flow_parsing.py index 6a434ed..76ca12a 100644 --- a/tests/test_flow_parsing.py +++ b/tests/test_flow_parsing.py @@ -690,6 +690,119 @@ def test_loki_flows_produce_policies(self) -> None: assert "db" in apps +class TestExtractFlowFromLokiLine: + """Tests for _extract_flow_from_loki_line (promtail vs fluent-bit).""" + + def test_raw_flow_dict(self) -> None: + """promtail-style: the parsed line IS the flow dict.""" + flow = _make_flow(port=80) + assert h._extract_flow_from_loki_line(flow) is flow + + def test_flow_envelope(self) -> None: + """promtail-style with a {flow: ...} wrapper.""" + inner = _make_flow(port=443) + # parse_flows handles the {"flow": ...} unwrap; _extract_flow_from_loki_line + # just returns the outer dict. + envelope = {"flow": inner, "time": "2026-01-01T00:00:00Z"} + result = h._extract_flow_from_loki_line(envelope) + assert result is envelope + + def test_fluentbit_envelope(self) -> None: + """fluent-bit-style: flow JSON is escaped inside a 'log' string.""" + inner = _make_flow(port=8080) + log_value = "2026-04-10T09:08:39.418Z stdout F " + json.dumps(inner) + line = {"log": log_value, "kubernetes": {"namespace_name": "kube-system"}} + result = h._extract_flow_from_loki_line(line) + assert result is not None + assert result["l4"]["TCP"]["destination_port"] == 8080 + assert result["verdict"] == "AUDIT" + + def test_fluentbit_envelope_with_flow_wrapper(self) -> None: + """fluent-bit log value contains {flow: {}}.""" + inner = _make_flow(port=9090) + wrapper = {"flow": inner, "time": "2026-04-10T09:00:00Z"} + log_value = "2026-04-10T09:00:00.000Z stdout F " + json.dumps(wrapper) + line = {"log": log_value, "kubernetes": {}} + result = h._extract_flow_from_loki_line(line) + assert result is not None + assert result["flow"]["l4"]["TCP"]["destination_port"] == 9090 + + def test_fluentbit_non_json_log(self) -> None: + """fluent-bit line whose 'log' is not parseable JSON returns None.""" + line = {"log": "2026-04-10T09:00:00Z stdout F level=info msg=starting"} + assert h._extract_flow_from_loki_line(line) is None + + def test_fluentbit_log_not_dict(self) -> None: + """fluent-bit line whose inner JSON is a list returns None.""" + line = {"log": "2026-04-10T09:00:00Z stdout F [1, 2, 3]"} + assert h._extract_flow_from_loki_line(line) is None + + def test_non_dict_input(self) -> None: + assert h._extract_flow_from_loki_line("not a dict") is None + assert h._extract_flow_from_loki_line(42) is None + assert h._extract_flow_from_loki_line(None) is None + + +class TestFluentBitLokiEndToEnd: + """End-to-end: fluent-bit Loki response -> parse_flows -> policies.""" + + _EMPTY_RESPONSE = json.dumps({"status": "success", "data": {"result": []}}).encode() + + @staticmethod + def _fluentbit_loki_response(flows: list[dict[str, Any]]) -> bytes: + """Build a Loki response where each line is a fluent-bit envelope.""" + values = [] + for i, flow in enumerate(flows, 1): + ts = str(i * 1_000_000_000) + log_text = f"2026-04-10T09:08:{i:02d}.000Z stdout F " + json.dumps(flow) + envelope = {"log": log_text, "kubernetes": {"namespace_name": "kube-system"}} + values.append([ts, json.dumps(envelope)]) + body = { + "status": "success", + "data": {"result": [{"stream": {}, "values": values}]}, + } + return json.dumps(body).encode() + + def test_fluentbit_flows_produce_policies(self) -> None: + flows = [ + _make_flow(src_app="web", dst_app="api", port=8080), + _make_flow(src_app="api", dst_app="db", port=5432), + ] + resp_bytes = self._fluentbit_loki_response(flows) + + empty_resp = mock.MagicMock() + empty_resp.read.return_value = self._EMPTY_RESPONSE + empty_resp.__enter__ = mock.Mock(return_value=empty_resp) + empty_resp.__exit__ = mock.Mock(return_value=False) + + data_resp = mock.MagicMock() + data_resp.read.return_value = resp_bytes + data_resp.__enter__ = mock.Mock(return_value=data_resp) + data_resp.__exit__ = mock.Mock(return_value=False) + + with mock.patch("hubble_audit2policy.urllib.request.urlopen") as mock_open: + mock_open.side_effect = [data_resp, empty_resp] + + loki_result = h._read_flows_loki( + "http://loki:3100", + '{app="hubble"}', + 3600, + 0, + threads=1, + chunk_seconds=7200, + ) + assert len(loki_result.flows) == 2 + policies, _, total, matched, _ = h.parse_flows( + "", LABEL_KEYS, set(), set(), flow_iter=iter(loki_result.flows) + ) + assert total == 2 + assert matched == 2 + apps = {app for _, app in policies} + assert "web" in apps + assert "api" in apps + assert "db" in apps + + class TestLokiAuth: """Test Loki authentication modes in _read_flows_loki.""" @@ -809,7 +922,7 @@ def test_tls_ca_with_bearer(self) -> None: def test_enrich_query_no_namespaces(self) -> None: q = '{container="cilium-agent"}' - verdict = r' |= "\"verdict\":"' + verdict = ' |~ "verdict.{0,5}:"' assert h._loki_enrich_query(q, []) == q + verdict assert h._loki_enrich_query(q, set()) == q + verdict @@ -817,48 +930,76 @@ def test_enrich_query_single_namespace(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, ["argocd"]) assert result == ( - r'{container="cilium-agent"} |= "\"verdict\":"' - r' |= "\"namespace\":\"argocd\""' + '{container="cilium-agent"} |~ "verdict.{0,5}:" |~ "namespace.{0,5}:.{0,5}argocd"' ) def test_enrich_query_multiple_namespaces(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, ["monitoring", "argocd"]) assert result == ( - r'{container="cilium-agent"} |= "\"verdict\":"' - r' |~ "\"namespace\":\"(argocd|monitoring)\""' + '{container="cilium-agent"} |~ "verdict.{0,5}:"' + ' |~ "namespace.{0,5}:.{0,5}(argocd|monitoring)"' ) def test_enrich_query_deduplicates(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, ["argocd", "argocd"]) assert result == ( - r'{container="cilium-agent"} |= "\"verdict\":"' - r' |= "\"namespace\":\"argocd\""' + '{container="cilium-agent"} |~ "verdict.{0,5}:" |~ "namespace.{0,5}:.{0,5}argocd"' ) def test_enrich_query_single_verdict(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, [], verdicts=["AUDIT"]) - assert result == (r'{container="cilium-agent"} |= "\"verdict\":\"AUDIT\""') + assert result == '{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}AUDIT"' def test_enrich_query_multiple_verdicts(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, [], verdicts=["DROPPED", "AUDIT"]) - assert result == (r'{container="cilium-agent"} |~ "\"verdict\":\"(AUDIT|DROPPED)\""') + assert result == ('{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}(AUDIT|DROPPED)"') def test_enrich_query_verdict_deduplicates(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, [], verdicts=["AUDIT", "AUDIT"]) - assert result == (r'{container="cilium-agent"} |= "\"verdict\":\"AUDIT\""') + assert result == '{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}AUDIT"' def test_enrich_query_verdict_and_namespace(self) -> None: q = '{container="cilium-agent"}' result = h._loki_enrich_query(q, ["argocd"], verdicts=["AUDIT"]) assert result == ( - r'{container="cilium-agent"} |= "\"verdict\":\"AUDIT\""' - r' |= "\"namespace\":\"argocd\""' + '{container="cilium-agent"} |~ "verdict.{0,5}:.{0,5}AUDIT"' + ' |~ "namespace.{0,5}:.{0,5}argocd"' + ) + + def test_enrich_query_filters_match_promtail_format(self) -> None: + """Verify generated regex matches promtail-style log text.""" + import re + + q = '{app="cilium-agent"}' + result = h._loki_enrich_query(q, ["loki"], verdicts=["AUDIT"]) + # Extract the two regex patterns from the LogQL filters. + patterns = re.findall(r'\|~ "([^"]+)"', result) + assert len(patterns) == 2 + # promtail text has plain JSON: "verdict":"AUDIT" "namespace":"loki" + promtail_line = '{"verdict":"AUDIT","source":{"namespace":"loki"}}' + for pat in patterns: + assert re.search(pat, promtail_line), f"pattern {pat!r} should match promtail" + + def test_enrich_query_filters_match_fluentbit_format(self) -> None: + """Verify generated regex matches fluent-bit-style log text.""" + import re + + q = '{app="cilium-agent"}' + result = h._loki_enrich_query(q, ["loki"], verdicts=["AUDIT"]) + patterns = re.findall(r'\|~ "([^"]+)"', result) + assert len(patterns) == 2 + # fluent-bit text has escaped JSON inside a "log" wrapper: + fluentbit_line = ( + r'{"log":"2026-04-10T09:08:39Z stdout F ' + r'{\"verdict\":\"AUDIT\",\"source\":{\"namespace\":\"loki\"}}"}' ) + for pat in patterns: + assert re.search(pat, fluentbit_line), f"pattern {pat!r} should match fluent-bit" def test_build_loki_ssl_context_none(self) -> None: assert h._build_loki_ssl_context(None) is None