-
Notifications
You must be signed in to change notification settings - Fork 0
feat: readable firewall rule expressions #5
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| """Render an nftables JSON rule expression into readable nft-like syntax. | ||
|
|
||
| `nft -j list ruleset` emits each rule's body as a list of statement objects. | ||
| Shown raw, that JSON is unreadable; this turns it into something close to what | ||
| `nft list ruleset` (non-JSON) prints, e.g. `tcp dport 22 accept` or | ||
| `ip saddr @allowlist accept`. | ||
|
|
||
| Pure and defensive: any shape it doesn't recognise falls back to a compact | ||
| form rather than raising, so a novel rule never breaks the Firewall page. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import json | ||
| from typing import Any | ||
|
|
||
| # meta keys that nft prints without the leading "meta" (others show as "meta <k>"). | ||
| _BARE_META = {"iifname", "oifname", "iif", "oif"} | ||
| _VERDICTS = {"accept", "drop", "reject", "return", "continue", "masquerade"} | ||
|
|
||
|
|
||
| def render_expr(expr: Any) -> str: | ||
| """Render a rule expression (JSON string or parsed list) to nft-like text.""" | ||
| if isinstance(expr, str): | ||
| try: | ||
| items = json.loads(expr) | ||
| except ValueError: | ||
| return expr | ||
| else: | ||
| items = expr | ||
| if not isinstance(items, list): | ||
| return str(items) | ||
| parts = [] | ||
| for s in items: | ||
| try: | ||
| parts.append(_stmt(s)) | ||
| except Exception: # a novel/malformed shape must never break the page | ||
| parts.append(json.dumps(s, ensure_ascii=False, separators=(",", ":"))) | ||
| return " ".join(p for p in parts if p) | ||
|
|
||
|
|
||
| def _stmt(stmt: Any) -> str: | ||
| if not isinstance(stmt, dict) or len(stmt) != 1: | ||
| return _value(stmt) | ||
| key, val = next(iter(stmt.items())) | ||
|
|
||
| if key in _VERDICTS: | ||
| return key | ||
| if key == "match": | ||
| return _match(val) | ||
| if key in ("jump", "goto"): | ||
| target = val.get("target") if isinstance(val, dict) else val | ||
| return f"{key} {target}" | ||
| if key == "counter": | ||
| return "counter" | ||
| if key == "log": | ||
| prefix = val.get("prefix") if isinstance(val, dict) else None | ||
| return f'log prefix "{prefix}"' if prefix else "log" | ||
| if key in ("dnat", "snat"): | ||
| return f"{key} to {_dnat_target(val)}" | ||
| if key == "mangle": | ||
| # {"mangle": {"key": <left>, "value": <v>}} -> "<left> set <v>" | ||
| if isinstance(val, dict): | ||
| return f"{_value(val.get('key'))} set {_value(val.get('value'))}".strip() | ||
| return f"mangle {_value(val)}".strip() | ||
| if key == "limit": | ||
| return "limit" | ||
| if key == "queue": | ||
| return "queue" | ||
| if key == "reject": | ||
| return "reject" | ||
| if key == "xt": | ||
| # iptables-nft compatibility match/target (e.g. UFW's `-m conntrack`). | ||
| # nft can't express it natively; surface the extension name. | ||
| name = val.get("name") if isinstance(val, dict) else None | ||
| return name or "xt" | ||
| # Unknown statement: show its key (and value if scalar). | ||
| if val in (None, {}, []): | ||
| return key | ||
| return f"{key} {_value(val)}" | ||
|
|
||
|
|
||
| def _match(val: Any) -> str: | ||
| if not isinstance(val, dict): | ||
| return _value(val) | ||
| op = val.get("op", "==") | ||
| left = _value(val.get("left")) | ||
| right = _value(val.get("right")) | ||
| # nft prints set membership implicitly (`ct state { ... }`), so "==" and "in" | ||
| # carry no visible operator. | ||
| if op in ("==", "in"): | ||
| return f"{left} {right}".strip() | ||
| return f"{left} {op} {right}".strip() | ||
|
|
||
|
|
||
| def _dnat_target(val: Any) -> str: | ||
| if isinstance(val, dict): | ||
| addr = _value(val.get("addr")) if val.get("addr") is not None else "" | ||
| port = val.get("port") | ||
| return f"{addr}:{port}" if port is not None else addr | ||
| return _value(val) | ||
|
|
||
|
|
||
| def _value(v: Any) -> str: | ||
| if v is None: | ||
| return "" | ||
| if isinstance(v, bool): | ||
| return "true" if v else "false" | ||
| if isinstance(v, (int, float)): | ||
| return str(v) | ||
| if isinstance(v, str): | ||
| return v | ||
| if isinstance(v, list): | ||
| return " . ".join(_value(x) for x in v) | ||
| if isinstance(v, dict): | ||
| return _dict_value(v) | ||
| return str(v) | ||
|
|
||
|
|
||
| def _dict_value(v: dict) -> str: | ||
| if "payload" in v: # {"payload": {"protocol": "tcp", "field": "dport"}} | ||
| p = v["payload"] | ||
| return f"{p.get('protocol', '')} {p.get('field', '')}".strip() or "payload" | ||
| if "meta" in v: # {"meta": {"key": "iifname"}} | ||
| key = v["meta"].get("key", "") | ||
| return key if key in _BARE_META else f"meta {key}".strip() | ||
| if "ct" in v: # {"ct": {"key": "state"}} | ||
| return f"ct {v['ct'].get('key', '')}".strip() | ||
| if "set" in v: # anonymous set {"set": ["a", "b"]} or named "@name" | ||
| elems = v["set"] | ||
| if isinstance(elems, list): | ||
| return "{ " + ", ".join(_value(e) for e in elems) + " }" | ||
| return _value(elems) | ||
| if "prefix" in v: # {"prefix": {"addr": "10.0.0.0", "len": 8}} | ||
| p = v["prefix"] | ||
| return f"{_value(p.get('addr'))}/{p.get('len')}" | ||
| if "range" in v: # {"range": [1, 1024]} | ||
| r = v["range"] | ||
| return f"{_value(r[0])}-{_value(r[1])}" if isinstance(r, list) and len(r) == 2 else _value(r) | ||
| if "concat" in v: | ||
| return " . ".join(_value(x) for x in v["concat"]) | ||
| # Unknown object: compact JSON so nothing is lost. | ||
| return json.dumps(v, ensure_ascii=False, separators=(",", ":")) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| import json | ||
|
|
||
| from bastion.services.nftexpr import render_expr | ||
|
|
||
|
|
||
| def test_simple_dport_accept(): | ||
| expr = [ | ||
| {"match": {"op": "==", "left": {"payload": {"protocol": "tcp", "field": "dport"}}, "right": 22}}, | ||
| {"accept": None}, | ||
| ] | ||
| assert render_expr(expr) == "tcp dport 22 accept" | ||
|
|
||
|
|
||
| def test_saddr_named_set_accept(): | ||
| expr = [ | ||
| {"match": {"op": "==", "left": {"payload": {"protocol": "ip", "field": "saddr"}}, "right": "@bastion_allow"}}, | ||
| {"accept": None}, | ||
| ] | ||
| assert render_expr(expr) == "ip saddr @bastion_allow accept" | ||
|
|
||
|
|
||
| def test_not_equal_operator(): | ||
| expr = [ | ||
| {"match": {"op": "!=", "left": {"payload": {"protocol": "ip", "field": "daddr"}}, "right": "@nozapret"}}, | ||
| ] | ||
| assert render_expr(expr) == "ip daddr != @nozapret" | ||
|
|
||
|
|
||
| def test_anonymous_set_dport(): | ||
| expr = [ | ||
| {"match": {"op": "==", "left": {"payload": {"protocol": "tcp", "field": "dport"}}, | ||
| "right": {"set": [80, 443]}}}, | ||
| {"accept": None}, | ||
| ] | ||
| assert render_expr(expr) == "tcp dport { 80, 443 } accept" | ||
|
|
||
|
|
||
| def test_meta_iifname_bare(): | ||
| expr = [{"match": {"op": "==", "left": {"meta": {"key": "iifname"}}, "right": "enp4s0"}}] | ||
| assert render_expr(expr) == "iifname enp4s0" | ||
|
|
||
|
|
||
| def test_ct_state(): | ||
| # nft prints set membership implicitly — no visible "in" operator. | ||
| expr = [ | ||
| {"match": {"op": "in", "left": {"ct": {"key": "state"}}, "right": {"set": ["established", "related"]}}}, | ||
| {"accept": None}, | ||
| ] | ||
| assert render_expr(expr) == "ct state { established, related } accept" | ||
|
|
||
|
|
||
| def test_meta_mark_keeps_meta_prefix(): | ||
| # Only iifname/oifname/iif/oif are bare; other meta keys show "meta <k>". | ||
| expr = [{"match": {"op": "==", "left": {"meta": {"key": "mark"}}, "right": 1}}] | ||
| assert render_expr(expr) == "meta mark 1" | ||
|
|
||
|
|
||
| def test_mangle_non_dict_does_not_raise(): | ||
| assert render_expr([{"mangle": None}]) # must not raise; returns some string | ||
|
|
||
|
|
||
| def test_malformed_nested_shape_falls_back(): | ||
| # payload as a bare string (not the expected object) must not raise. | ||
| out = render_expr([{"match": {"op": "==", "left": {"payload": "weird"}, "right": 1}}]) | ||
| assert isinstance(out, str) and out | ||
|
|
||
|
|
||
| def test_jump_and_counter(): | ||
| expr = [{"counter": {"packets": 5, "bytes": 300}}, {"jump": {"target": "ufw-user-input"}}] | ||
| assert render_expr(expr) == "counter jump ufw-user-input" | ||
|
|
||
|
|
||
| def test_prefix_and_range(): | ||
| expr = [ | ||
| {"match": {"op": "==", "left": {"payload": {"protocol": "ip", "field": "saddr"}}, | ||
| "right": {"prefix": {"addr": "10.0.0.0", "len": 8}}}}, | ||
| ] | ||
| assert render_expr(expr) == "ip saddr 10.0.0.0/8" | ||
|
|
||
|
|
||
| def test_log_with_prefix(): | ||
| expr = [{"log": {"prefix": "[UFW BLOCK] "}}] | ||
| assert render_expr(expr) == 'log prefix "[UFW BLOCK] "' | ||
|
|
||
|
|
||
| def test_accepts_json_string_input(): | ||
| expr_str = json.dumps([{"drop": None}]) | ||
| assert render_expr(expr_str) == "drop" | ||
|
|
||
|
|
||
| def test_invalid_json_returns_input(): | ||
| assert render_expr("{not json") == "{not json" | ||
|
|
||
|
|
||
| def test_unknown_statement_falls_back_gracefully(): | ||
| expr = [{"weird_stmt": {"a": 1}}] | ||
| out = render_expr(expr) | ||
| assert "weird_stmt" in out # does not raise, keeps information | ||
|
|
||
|
|
||
| # ---------- real UFW / iptables-nft rules from server1 ---------- | ||
| def test_ufw_iifname_lo_accept(): | ||
| expr = [ | ||
| {"match": {"op": "==", "left": {"meta": {"key": "iifname"}}, "right": "lo"}}, | ||
| {"counter": {"packets": 92, "bytes": 7604}}, | ||
| {"accept": None}, | ||
| ] | ||
| assert render_expr(expr) == "iifname lo counter accept" | ||
|
|
||
|
|
||
| def test_ufw_xt_conntrack_names_the_match(): | ||
| expr = [ | ||
| {"xt": {"type": "match", "name": "conntrack"}}, | ||
| {"counter": {"packets": 7860, "bytes": 994422}}, | ||
| {"drop": None}, | ||
| ] | ||
| assert render_expr(expr) == "conntrack counter drop" | ||
|
|
||
|
|
||
| def test_ufw_l4proto_icmp_xt_accept(): | ||
| expr = [ | ||
| {"match": {"op": "==", "left": {"meta": {"key": "l4proto"}}, "right": "icmp"}}, | ||
| {"xt": {"type": "match", "name": "icmp"}}, | ||
| {"counter": {"packets": 362, "bytes": 33264}}, | ||
| {"accept": None}, | ||
| ] | ||
| assert render_expr(expr) == "meta l4proto icmp icmp counter accept" | ||
|
|
||
|
|
||
| def test_docker_jump_chain(): | ||
| expr = [{"counter": {"packets": 0, "bytes": 0}}, {"jump": {"target": "DOCKER-USER"}}] | ||
| assert render_expr(expr) == "counter jump DOCKER-USER" |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.