From 9b1084a8eb8b1b716d83041cee8c53fd6736bddd Mon Sep 17 00:00:00 2001 From: Giacomo Sanchietti Date: Tue, 10 Nov 2026 14:11:43 +0100 Subject: [PATCH 1/3] feat(python3-nethsec): filter conntrack listings Live trace lists only the connections of one client, so list_connections() now accepts extra conntrack arguments and returns an empty list when conntrack prints nothing. Assisted-by: Claude Code:claude-opus-5-5 --- packages/python3-nethsec/src/nethsec/conntrack/__init__.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/packages/python3-nethsec/src/nethsec/conntrack/__init__.py b/packages/python3-nethsec/src/nethsec/conntrack/__init__.py index 8712432fb..d768a15d9 100644 --- a/packages/python3-nethsec/src/nethsec/conntrack/__init__.py +++ b/packages/python3-nethsec/src/nethsec/conntrack/__init__.py @@ -65,7 +65,7 @@ def __parse_connection_info(flow: Element) -> dict: return result -def list_connections(labels: list = None): +def list_connections(labels: list = None, extra_args: list = None): """ List all network connections. @@ -73,6 +73,7 @@ def list_connections(labels: list = None): - labels: optional list of label strings to filter by. Only connections that have ALL the specified labels are returned. If ``None`` or empty, all connections are returned. + - extra_args: optional list of extra conntrack arguments, e.g. ``["--orig-src", "10.0.0.1"]``. Returns: list of connections. @@ -80,7 +81,11 @@ def list_connections(labels: list = None): cmd = ["conntrack", "-L", "-o", "labels,xml"] if labels: cmd.extend(["-l", ",".join(labels)]) + if extra_args: + cmd.extend(extra_args) result = subprocess.run(cmd, capture_output=True, text=True) + if not result.stdout.strip(): + return [] root = ElementTree.fromstring(result.stdout) connections = [] for flow in root.findall('flow'): From d7445e76102a07add5c51b65d8b1cd598550ae77 Mon Sep 17 00:00:00 2001 From: Giacomo Sanchietti Date: Tue, 10 Nov 2026 14:33:51 +0100 Subject: [PATCH 2/3] feat(ns-api): add live trace troubleshooting API Administrators cannot tell which firewall layer blocks a client. ns.livetrace traces one client for a limited time and reports the blocking layer and rule, DNS queries, live connections and routing hints, using nftables tracing and tcpdump. Assisted-by: Claude Code:claude-sonnet-5-5 --- packages/ns-api/Makefile | 6 + packages/ns-api/README.md | 213 +++++++++ packages/ns-api/files/ns-livetrace.py | 571 ++++++++++++++++++++++++ packages/ns-api/files/ns.livetrace | 287 ++++++++++++ packages/ns-api/files/ns.livetrace.json | 13 + 5 files changed, 1090 insertions(+) create mode 100755 packages/ns-api/files/ns-livetrace.py create mode 100755 packages/ns-api/files/ns.livetrace create mode 100644 packages/ns-api/files/ns.livetrace.json diff --git a/packages/ns-api/Makefile b/packages/ns-api/Makefile index 930202e46..4dba8f17e 100644 --- a/packages/ns-api/Makefile +++ b/packages/ns-api/Makefile @@ -23,14 +23,17 @@ define Package/ns-api URL:=https://github.com/NethServer/nethsecurity-controller/ DEPENDS:= \ +adblock \ + +conntrack \ +coreutils-date \ +coreutils-stty \ + +ip-full \ +python3-idna \ +python3-nethsec \ +python3-openssl \ +python3-requests \ +python3-semver \ +python3-urllib \ + +tcpdump \ +whiptail \ +sshpass \ +wireguard-tools @@ -179,6 +182,8 @@ define Package/ns-api/install $(INSTALL_DATA) ./files/ns.ha.json $(1)/usr/share/rpcd/acl.d/ $(INSTALL_BIN) ./files/ns.flows $(1)/usr/libexec/rpcd/ $(INSTALL_DATA) ./files/ns.flows.json $(1)/usr/share/rpcd/acl.d/ + $(INSTALL_BIN) ./files/ns.livetrace $(1)/usr/libexec/rpcd/ + $(INSTALL_DATA) ./files/ns.livetrace.json $(1)/usr/share/rpcd/acl.d/ $(INSTALL_DIR) $(1)/lib/upgrade/keep.d $(INSTALL_CONF) files/ns-api.keep $(1)/lib/upgrade/keep.d/ns-api $(INSTALL_CONF) files/msmtp.keep $(1)/lib/upgrade/keep.d/msmtp @@ -187,6 +192,7 @@ define Package/ns-api/install $(INSTALL_BIN) ./files/load-kernel-modules $(1)/usr/sbin/load-kernel-modules $(INSTALL_BIN) ./files/setup $(1)/usr/sbin/setup $(INSTALL_BIN) ./files/firewall-apply-default-logging $(1)/usr/sbin/firewall-apply-default-logging + $(INSTALL_BIN) ./files/ns-livetrace.py $(1)/usr/libexec/ns-livetrace $(INSTALL_DIR) $(1)/etc/config $(INSTALL_CONF) ./files/config/ns-api $(1)/etc/config/ns-api $(INSTALL_CONF) ./files/config/ns-wizard $(1)/etc/config/ns-wizard diff --git a/packages/ns-api/README.md b/packages/ns-api/README.md index b356623ba..acd6907a7 100644 --- a/packages/ns-api/README.md +++ b/packages/ns-api/README.md @@ -7991,6 +7991,219 @@ Example response: } ``` +## ns.livetrace + +Trace the live traffic of one client to find which layer blocks it. +One session at a time. Results stay readable after the session ends. + +### start + +Start a session. A running session is stopped first. + +Parameters: + +- `ip`: client IP address +- `duration`: seconds, from 10 to 600, default 120 +- `port`: optional, destination port of new connections, from 1 to 65535 +- `protocol`: optional, `tcp` or `udp` + +Missing, `null`, `0` and `""` mean not set. `port` and `protocol` do not filter DNS queries. + +```bash +api-cli ns.livetrace start --data '{"ip": "10.10.10.50", "duration": 120, "port": 443, "protocol": "tcp"}' +``` + +Example response: + +```json +{ + "message": "success" +} +``` + +Errors: + +- validation: `invalid_ip_address` (also for scoped, IPv4-mapped, multicast, loopback and unspecified addresses), + `invalid_duration`, `invalid_port`, `invalid_protocol` +- `collector_not_started`: the collector did not start, the reason is in `failure` of `status` +- `session_busy`: the running session did not stop within 5 seconds + +### stop + +Stop the running session: + +```bash +api-cli ns.livetrace stop +``` + +Example response: + +```json +{ + "message": "success" +} +``` + +### status + +Return the session state: + +```bash +api-cli ns.livetrace status +``` + +Example response: + +```json +{ + "running": true, + "ip": "10.10.10.50", + "duration": 120, + "port": 0, + "protocol": "", + "remaining": 87, + "blocked": [ + { + "protocol": "tcp", + "source": "10.10.10.50", + "destination": "1.0.0.1", + "destination_port": "80", + "layer": "threat_shield_ip", + "table": "banIP", + "chain": "_outbound", + "rule": "blocklist.v4", + "in_interface": "eth2", + "out_interface": "eth1", + "count": 5, + "first_seen": 1793706147, + "last_seen": 1793706152 + } + ], + "dns": [ + { + "name": "doubleclick.com", + "type": "A", + "result": "NXDOMAIN", + "filtered": true, + "count": 1, + "last_seen": 1793706213 + } + ], + "connections": [ + { + "source": "10.10.10.50", + "destination": "1.1.1.1", + "protocol": "tcp", + "source_port": "39494", + "destination_port": "80", + "source_stats": { + "packets": 6, + "bytes": 391 + }, + "destination_stats": { + "packets": 4, + "bytes": 602 + }, + "id": "4272941068", + "timeout": "115", + "state": "TIME_WAIT", + "labels": [ + "netify-analyzed" + ], + "in_interface": "eth2", + "out_interface": "eth1", + "rule": "Accept lan to wan forwarding", + "application": "wikimedia", + "dpi_protocol": "DNS", + "host": "auth.wikimedia.org" + } + ], + "allowed": [], + "failure": "", + "truncated": false +} +``` + +Example `allowed` when the session is not running: + +```json +[ + { + "protocol": "tcp", + "source": "10.10.10.50", + "destination": "1.1.1.1", + "destination_port": "443", + "in_interface": "eth2", + "out_interface": "eth1", + "rule": "Accept lan to wan forwarding", + "count": 3, + "source_ports": ["51234", "51236", "51240"] + } +] +``` + +Fields: + +- `port`, `protocol`: session filters, `0` and `""` when not set +- `failure`: error message of the collector, empty if none +- `truncated`: true if something was skipped: new connections over the rate limit, more than 200 `blocked` or + `dns` entries, more than 1000 flows (oldest dropped from `allowed`) + +`blocked`: new connections dropped or rejected. + +- `layer`: `firewall`, `dpi`, `threat_shield_ip`, `ips`, `dns_filter`, `mac_binding`, `routing` or the nftables table name +- `rule`: fw4 rule name, matching set name (e.g. banIP list) or nftables rule text +- `count`: number of distinct connections +- `application`, `host`: only with layer `dpi`, read from ns-flows, empty if the flow is unknown +- layer `routing` has empty `table` and `chain`, `rule` is one of: + - `no_route`: no route to the destination, or packet refused by reverse path filtering; + multicast and TTL 1 packets are not reported + - `private_to_wan`: a private destination (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, 100.64.0.0/10, fc00::/7) + left through the device of the default route: missing route or tunnel down; the connection is also in `connections` + +`dns`: queries sent by the client and their replies. + +- `result`: response code (`NOERROR`, `NXDOMAIN`, ...), empty if no reply was seen +- `filtered`: true if adblock is enabled and the name, or a parent domain, is in its list; + the most specific allowlist or blocklist entry wins + +`connections`: conntrack entries of the client, same format as `ns.conntrack list`. Filled only while the session is running. + +- `rule`: fw4 rule that accepted the connection, only if the connection was traced +- `in_interface`, `out_interface`: from the trace, or from a route lookup if the connection was not traced +- `application`, `dpi_protocol`, `host`: read from ns-flows, if the flow is classified + +`allowed`: traced allowed connections of the last session, without traffic counters. Filled only when the session is not running. + +- grouped by `protocol`, `source`, `destination`, `destination_port`, `in_interface`, `out_interface` and `rule` +- `count`: number of traced connections, `source_ports`: their source ports +- sorted by `count`, descending + +Limitations: + +- only the first packet of each new connection and DPI-blocked packets are traced +- rate limit: 50 packets per second per direction, burst 100; needs `net.netfilter.nf_conntrack_acct=1` (OpenWrt default) +- Snort (IPS) drops are not shown: a connection dropped by Snort is listed as `allowed` +- a DPI block shows up after the flow is classified: the first packets pass +- chains with priority lower than -199 (e.g. the raw table) are not visible +- DNS filter detection covers adblock with dnsmasq only, not FlashStart +- `filtered` is also set when the query goes to an external resolver, and for adblock SafeSearch domains + (rewritten, not blocked) +- DNS over TCP and DNS over HTTPS are not captured +- with `port` or `protocol` set, ICMP is not traced +- connections opened before the session have no `rule`; the route lookup of their interfaces ignores + policy routing (multi-WAN) and port forwards +- a wrong route is detected only when a private destination leaves through the device of the default route +- port forwards: trace the public source IP with the public port; tracing the internal server shows no `rule` + and no `out_interface`; connections show the public destination address and port +- after a ruleset reload (firewall, banIP, DPI) `nft monitor trace` is restarted: packets traced before the restart + have an empty `rule` and may show the inner chain (e.g. `reject_to_wan`), connections started during the + restart are not listed +- if the traced IP is the remote side of a masqueraded connection, a DPI-blocked reply shows the WAN address and + the client port as destination and is also listed in `allowed`: trace the client instead +- if the collector is killed with SIGKILL, `nft monitor`, `tcpdump` and the `inet ns_livetrace` table stay + until `stop` or the next `start` + ## ns.objects Manage domain sets and host sets. diff --git a/packages/ns-api/files/ns-livetrace.py b/packages/ns-api/files/ns-livetrace.py new file mode 100755 index 000000000..840fdf66e --- /dev/null +++ b/packages/ns-api/files/ns-livetrace.py @@ -0,0 +1,571 @@ +#!/usr/bin/python3 +# +# Copyright (C) 2026 Nethesis S.r.l. +# SPDX-License-Identifier: GPL-2.0-only +# + +# Trace new connections and DNS queries of one client, write results to STATE. +# Usage: ns-livetrace [port] [protocol] + +import glob +import ipaddress +import json +import os +import queue +import re +import signal +import subprocess +import sys +import threading +import time + +import requests +from euci import EUci + +STATE = "/var/run/ns-livetrace.json" +PID = "/var/run/ns-livetrace.pid" +TABLE = "ns_livetrace" +MAX_ITEMS = 200 +MAX_FLOWS = 1000 +LAYERS = { + "fw4": "firewall", + "banIP": "threat_shield_ip", + "netifyd": "dpi", + "snort": "ips", + "adblock": "dns_filter", + "ns-binding": "mac_binding", +} +# chains that only execute the verdict decided by the rule jumping into them +REJECT_CHAINS = ("handle_reject", "_reject") +PRIVATE_NETS = [ + ipaddress.ip_network(n) + for n in ( + "10.0.0.0/8", + "172.16.0.0/12", + "192.168.0.0/16", + "100.64.0.0/10", + "fc00::/7", + ) +] +# fw4 zone chains reached by zone policies, forwardings and user rules +ZONE_VERDICT_RE = re.compile(r"(accept|drop|reject)_(to|from)_") + +TRACE_RE = re.compile(r"^trace id (\S+) \S+ (\S+) (\S+) (.*)$") +VERDICT_RE = re.compile(r"\(verdict (\w+)[^)]*\)$") +COMMENT_RE = re.compile(r'comment "!fw4: (.*)"') +QUERY_RE = re.compile(r": (\d+)\S* (?:\[\S+\] )*(\w+)\? (\S+) ") +REPLY_RE = re.compile(r": (\d+)\S* (?:([A-Z][A-Za-z]+)\S* )?(?:\[\S+\] )*\d+/\d+/\d+") + + +def nft(script, check=False): + subprocess.run( + ["nft", "-f", "-"], input=script, text=True, capture_output=True, check=check + ) + + +def tables(): + # a recreated table gets a new handle + return subprocess.run( + ["nft", "-j", "list", "tables"], capture_output=True, text=True + ).stdout + + +def field(regex, text): + m = re.search(regex, text) + return m.group(1) if m else "" + + +def rule_name(text): + comment = COMMENT_RE.search(text) + if comment: + return comment.group(1) + # banIP rules: the set name tells which list matched + ip_set = re.search(r"@(\S+)", text) + if ip_set: + return ip_set.group(1) + # rule resolved against a stale ruleset cache: no text to show + if text.startswith("unknown "): + return "" + text = re.sub(r"counter packets \d+ bytes \d+ ", "", text[len("rule ") :]) + return VERDICT_RE.sub("", text).strip() + + +def user_jump(t, chain): + # the stored jump is a user rule targeting this zone verdict chain + jump = t.get("jump") + return bool( + ZONE_VERDICT_RE.match(chain) + and jump + and COMMENT_RE.search(jump[2]) + and field(r"\(verdict (?:jump|goto) ([^)\s]+)", jump[2]) == chain + ) + + +def flow_key(protocol, source, source_port, destination, destination_port): + return f"{protocol}|{source}|{source_port}|{destination}|{destination_port}" + + +def is_private(address): + try: + ip = ipaddress.ip_address(address) + return any(ip in n for n in PRIVATE_NETS) + except ValueError: + return False + + +def is_multicast(address): + try: + return ipaddress.ip_address(address).is_multicast + except ValueError: + return False + + +# routes matching each destination, looked up again after ROUTE_TTL seconds +ROUTE_TTL = 10 +ROUTES = {} + + +def default_route_only(address, out): + # true when only default routes match and out is one of their devices + try: + cached = ROUTES.get(address) + if not cached or time.time() - cached[0] > ROUTE_TTL: + if len(ROUTES) >= MAX_FLOWS: + ROUTES.clear() + family = "-6" if ":" in address else "-4" + p = subprocess.run( + ["ip", family, "-j", "route", "list", "match", address], + capture_output=True, + text=True, + ) + ROUTES[address] = (time.time(), json.loads(p.stdout)) + routes = ROUTES[address][1] + return all(r.get("dst") == "default" for r in routes) and out in { + r.get("dev") for r in routes + } + except Exception: + return False + + +def dpi_flows(): + # ns-flows daemon is optional + try: + resp = requests.get("http://127.0.0.1:8080/flows", timeout=2) + return [item.get("flow") or {} for item in resp.json().get("flows", [])] + except Exception: + return [] + + +def dpi_app(flows, source, source_port, destination, destination_port): + ends = {(source, source_port), (destination, destination_port)} + for flow in flows: + local = (str(flow.get("local_ip")), str(flow.get("local_port"))) + other = (str(flow.get("other_ip")), str(flow.get("other_port"))) + if {local, other} == ends: + application = flow.get("detected_application_name") or "" + host = flow.get("host_server_name") or "" + return application.removeprefix("netify."), host + return "", "" + + +class Collector: + def __init__(self, ip, duration, port, protocol): + self.ip = ip + self.duration = duration + self.port = port + self.protocol = protocol + self.start = int(time.time()) + self.traces = {} + self.blocked = {} + self.blocked_tuples = {} + self.flows = {} + self.dns = {} + self.pending_dns = {} + self.filtered = {} + self.failure = "" + self.adblock_list = None + # nft monitor printed a rule missing from its ruleset cache + self.stale_rules = False + # something was skipped by the rate limit or a list cap + self.truncated = False + + def setup(self): + uci = EUci() + if uci.get("adblock", "global", "adb_enabled", default="0") == "1": + self.adblock_list = next( + iter(glob.glob("/tmp/dnsmasq.*.d/adb_list.overall")), None + ) + + def trace_line(self, line): + m = TRACE_RE.match(line) + if not m: + return + tid, table, chain, rest = m.groups() + if table == TABLE: + # each traced packet starts with a packet line of our own table + if rest.startswith("packet: "): + # trace id reused: keep the previous group to be closed on its own + if tid in self.traces: + self.traces[(tid, time.time())] = self.traces.pop(tid) + self.traces[tid] = { + "time": time.time(), + "last": time.time(), + "in": "", + "out": "", + "tuple": None, + "routed": False, + } + return + t = self.traces.get(tid) + if t is None: + return + t["last"] = time.time() + # any input or forward stage chain means the packet passed routing + if "input" in chain or "forward" in chain: + t["routed"] = True + if rest.startswith("packet: "): + t["in"] = t["in"] or field(r'\biif "([^"]+)"', rest) + t["out"] = t["out"] or field(r'\boif "([^"]+)"', rest) + if t["tuple"] is None: + t["ttl"] = field(r"\b(?:ip ttl|ip6 hoplimit) (\d+)", rest) + t["tuple"] = ( + # conntrack names ipv6-icmp as icmpv6 + field(r"\b(?:ip protocol|ip6 nexthdr) (\S+)", rest).replace( + "ipv6-icmp", "icmpv6" + ), + field(r"\bip6? saddr (\S+)", rest), + field(r"\b(?:tcp|udp) sport (\d+)", rest), + field(r"\bip6? daddr (\S+)", rest), + field(r"\b(?:tcp|udp) dport (\d+)", rest), + ) + return + if rest.startswith("unknown rule "): + self.stale_rules = True + is_rule = rest.startswith(("rule ", "unknown rule ")) + verdict = ( + field(r"\(verdict (\w+)", rest) + if is_rule + else field(r"^policy (\w+)", rest) + ) + if verdict in ("jump", "goto"): + if not user_jump(t, chain): + t["jump"] = (table, chain, rest) + elif verdict in ("drop", "reject") and "drop" not in t: + if "jump" in t and (chain in REJECT_CHAINS or user_jump(t, chain)): + t["drop"] = t["jump"] + else: + t["drop"] = ( + table, + chain, + rest if is_rule else "rule policy drop", + ) + elif ( + verdict == "accept" + and table == "fw4" + and not chain.startswith(("srcnat", "dstnat")) + ): + comment = COMMENT_RE.search(t["jump"][2] if user_jump(t, chain) else rest) + if comment and "accept" not in t: + t["accept"] = comment.group(1) + + def finalize(self, idle): + now = time.time() + flows = None + # unrouted groups wait longer: their input/forward lines may still be queued + for tid in [ + tid + for tid, t in self.traces.items() + if now - t["last"] >= (idle if t["routed"] or "drop" in t else idle * 10) + ]: + t = self.traces.pop(tid) + if t["tuple"] is None: + continue + protocol, source, source_port, destination, destination_port = t["tuple"] + if "drop" in t: + table, chain, text = t["drop"] + rule = rule_name(text) + layer = "dpi" if chain == "dpi_actions" else LAYERS.get(table, table) + # a flow accepted first and dropped later by DPI is blocked only + if layer == "dpi": + reply = (destination, destination_port, source, source_port) + for key in (t["tuple"], (protocol, *reply)): + self.flows.pop(flow_key(*key), None) + elif not t["routed"]: + # unjoined multicast and expiring TTL are dropped by routing on purpose + if is_multicast(destination) or t["ttl"] in ("0", "1"): + continue + # left prerouting without reaching input or forward: no route + table, chain, rule, layer = "", "", "no_route", "routing" + else: + key = flow_key(*t["tuple"]) + self.flows.pop(key, None) + self.flows[key] = { + "in_interface": t["in"], + "out_interface": t["out"], + "rule": t.get("accept", ""), + } + if len(self.flows) > MAX_FLOWS: + del self.flows[next(iter(self.flows))] + self.truncated = True + if ( + not t["out"] + or not is_private(destination) + or not default_route_only(destination, t["out"]) + ): + continue + # private destination on the default route: missing route or tunnel down + table, chain, rule, layer = "", "", "private_to_wan", "routing" + key = (protocol, source, destination, destination_port, table, chain, rule) + entry = self.blocked.get(key) + if entry is None: + if len(self.blocked) >= MAX_ITEMS: + self.truncated = True + continue + entry = self.blocked[key] = { + "protocol": protocol, + "source": source, + "destination": destination, + "destination_port": destination_port, + "layer": layer, + "table": table, + "chain": chain, + "rule": rule, + "in_interface": t["in"], + "out_interface": t["out"], + "count": 0, + "first_seen": int(t["time"]), + } + # retry while empty: ns-flows may publish the flow after the first drop + if layer == "dpi" and not entry.get("application"): + if flows is None: + flows = dpi_flows() + entry["application"], entry["host"] = dpi_app( + flows, source, source_port, destination, destination_port + ) + # count distinct connections, not packets + tuples = self.blocked_tuples.setdefault(key, set()) + if t["tuple"] not in tuples: + tuples.add(t["tuple"]) + entry["count"] += 1 + entry["last_seen"] = int(t["time"]) + + def is_filtered(self, name): + if self.adblock_list is None: + return False + if name not in self.filtered: + labels = name.split(".") + suffixes = [".".join(labels[i:]) for i in range(len(labels))] + args = [] + for s in suffixes: + args += ["-e", f"local=/{s}/", "-e", f"local=/{s}/#"] + found = set( + subprocess.run( + ["grep", "-xF", *args, self.adblock_list], + capture_output=True, + text=True, + ).stdout.splitlines() + ) + # most specific match wins, allowlist lines end with "#" + self.filtered[name] = next( + ( + f"local=/{s}/" in found + for s in suffixes + if f"local=/{s}/" in found or f"local=/{s}/#" in found + ), + False, + ) + return self.filtered[name] + + def dns_line(self, line): + m = QUERY_RE.search(line) + if m: + qid, qtype, name = m.group(1), m.group(2), m.group(3).rstrip(".") + if self.pending_dns.get(qid) == (name, qtype): + return + self.pending_dns[qid] = (name, qtype) + entry = self.dns.get((name, qtype)) + if entry is None: + if len(self.dns) >= MAX_ITEMS: + self.truncated = True + return + entry = self.dns[(name, qtype)] = { + "name": name, + "type": qtype, + "result": "", + "filtered": self.is_filtered(name), + "count": 0, + } + entry["count"] += 1 + entry["last_seen"] = int(time.time()) + return + m = REPLY_RE.search(line) + if m and m.group(1) in self.pending_dns: + entry = self.dns.get(self.pending_dns.pop(m.group(1))) + if entry: + entry["result"] = (m.group(2) or "NOERROR").upper() + + def check_skipped(self): + # false when the trace table is gone + p = subprocess.run( + ["nft", "list", "counter", "inet", TABLE, "skipped"], + capture_output=True, + text=True, + ) + if not self.truncated: + self.truncated = field(r"packets (\d+)", p.stdout) not in ("", "0") + return p.returncode == 0 + + def line(self, kind, line): + (self.trace_line if kind == "trace" else self.dns_line)(line) + + def write(self, running): + state = { + "running": running, + "ip": self.ip, + "duration": self.duration, + "port": self.port, + "protocol": self.protocol, + "start": self.start, + "blocked": list(self.blocked.values()), + "dns": list(self.dns.values()), + "flows": self.flows, + "failure": self.failure, + "truncated": self.truncated, + } + with open(STATE + ".tmp", "w") as f: + json.dump(state, f) + os.replace(STATE + ".tmp", STATE) + + +def reader(proc, kind, lines): + for line in proc.stdout: + lines.put((kind, line.rstrip("\n"))) + + +def main(): + ip = str(ipaddress.ip_address(sys.argv[1])) + duration = int(sys.argv[2]) + port = int(sys.argv[3]) if len(sys.argv) > 3 else 0 + protocol = sys.argv[4] if len(sys.argv) > 4 else "" + if protocol not in ("", "tcp", "udp"): + raise ValueError("invalid protocol") + family = "ip6" if ":" in ip else "ip" + # th dport is only meaningful on tcp/udp headers + l4proto = protocol or ("{ tcp, udp }" if port else "") + match = (f"meta l4proto {l4proto} " if l4proto else "") + ( + f"th dport {port} " if port else "" + ) + collector = Collector(ip, duration, port, protocol) + stop = threading.Event() + signal.signal(signal.SIGTERM, lambda *_: stop.set()) + collector.write(True) + procs = [] + threads = [] + lines = queue.Queue() + try: + collector.setup() + nft(f"table inet {TABLE}\ndelete table inet {TABLE}\n") + new = ( + f"{family} saddr {ip} {match}ct state new ct original packets 1", + f"{family} daddr {ip} {match}ct state new ct original packets 1", + ) + # DPI blocks a flow after its first packets + blocked = f"{family} saddr {ip} {match}ct label netify-blocked" + # accept only ends this chain; new connections reaching the counter were not traced + rules = "".join( + f"{r} limit rate 50/second burst 100 packets meta nftrace set 1 accept\n" + for r in (*new, blocked) + ) + "".join(f"{r} counter name skipped\n" for r in new) + nft( + f"table inet {TABLE} {{\ncounter skipped {{}}\nchain pre {{\n" + f"type filter hook prerouting priority -199;\n{rules}}}\n}}\n", + check=True, + ) + # only queries sent by the client and their replies + dns_filter = f"udp and ((src host {ip} and dst port 53) or (dst host {ip} and src port 53))" + + def spawn(kind, cmd): + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + errors="replace", + ) + threads.append( + threading.Thread(target=reader, args=(proc, kind, lines), daemon=True) + ) + threads[-1].start() + return proc + + known = tables() + for kind, cmd in ( + ("trace", ["nft", "monitor", "trace"]), + ("dns", ["tcpdump", "-l", "-n", "-i", "any", dns_filter]), + ): + procs.append(spawn(kind, cmd)) + monitor = threads[0] + with open(PID, "w") as f: + f.write(str(os.getpid())) + end = time.time() + duration + last_write = 0 + while time.time() < end and not stop.is_set(): + try: + collector.line(*lines.get(timeout=0.5)) + except queue.Empty: + pass + if time.time() - last_write >= 1: + # a recreated table reuses rule handles: no unknown rule is printed + current = tables() + if current != known: + known = current + collector.stale_rules = True + # nft monitor caches the ruleset at start: restart it after a reload + if collector.stale_rules: + procs[0].kill() + procs[0].wait() + # lines of the old monitor must not ask for another restart + monitor.join(1) + while not lines.empty(): + collector.line(*lines.get()) + # groups cut by the restart would be reported as no_route + collector.traces = { + k: t + for k, t in collector.traces.items() + if t["routed"] or "drop" in t + } + procs[0] = spawn("trace", procs[0].args) + monitor = threads[-1] + collector.stale_rules = False + for proc in procs: + if proc.poll() is not None: + raise RuntimeError(f"{proc.args[0]} exited") + collector.finalize(0.5) + if not collector.check_skipped(): + raise RuntimeError("trace table removed") + collector.write(True) + last_write = time.time() + except Exception as e: + collector.failure = (getattr(e, "stderr", None) or str(e)).strip() + finally: + collector.check_skipped() + nft(f"delete table inet {TABLE}\n") + for proc in procs: + proc.kill() + proc.wait() + for thread in threads: + thread.join(1) + # process lines still queued before closing the open groups + try: + while not lines.empty(): + collector.line(*lines.get()) + except Exception as e: + collector.failure = collector.failure or str(e) + collector.finalize(0) + collector.write(False) + if os.path.exists(PID): + os.remove(PID) + + +if __name__ == "__main__": + main() diff --git a/packages/ns-api/files/ns.livetrace b/packages/ns-api/files/ns.livetrace new file mode 100755 index 000000000..c19254725 --- /dev/null +++ b/packages/ns-api/files/ns.livetrace @@ -0,0 +1,287 @@ +#!/usr/bin/python3 +# +# Copyright (C) 2026 Nethesis S.r.l. +# SPDX-License-Identifier: GPL-2.0-only +# + +import fcntl +import ipaddress +import json +import os +import signal +import subprocess +import sys +import time +from collections import defaultdict + +import requests +from nethsec import conntrack +from nethsec.utils import validation_error + +STATE = "/var/run/ns-livetrace.json" +PID = "/var/run/ns-livetrace.pid" +LOCK = "/var/run/ns-livetrace.lock" +FLOW_FIELDS = ("protocol", "source", "source_port", "destination", "destination_port") +ALLOWED_FIELDS = ( + "protocol", + "source", + "destination", + "destination_port", + "in_interface", + "out_interface", + "rule", +) + + +def running_pid(): + try: + with open(PID) as f: + pid = int(f.read()) + with open(f"/proc/{pid}/cmdline", "rb") as f: + return pid if b"ns-livetrace" in f.read() else None + except (OSError, ValueError): + return None + + +def delete_table(): + subprocess.run( + ["nft", "delete", "table", "inet", "ns_livetrace"], capture_output=True + ) + + +def stop(): + pid = running_pid() + if pid is not None: + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + pass + # a SIGKILLed collector leaves its table behind + for _ in range(50): + if running_pid() is None: + delete_table() + return + time.sleep(0.1) + + +def start(data): + ip = data.get("ip", "") + duration = data.get("duration", 120) + try: + addr = ipaddress.ip_address(ip) + # nft cannot trace these addresses + if ( + not isinstance(ip, str) + or "%" in str(addr) + or getattr(addr, "ipv4_mapped", None) + or addr.is_multicast + or addr.is_loopback + or addr.is_unspecified + ): + raise ValueError + except ValueError: + return validation_error("ip", "invalid_ip_address", ip) + ip = str(addr) + # bool is an int subclass + if type(duration) is not int or not 10 <= duration <= 600: + return validation_error("duration", "invalid_duration", duration) + port = data.get("port") or 0 + protocol = data.get("protocol") or "" + if type(port) is not int or not 0 <= port <= 65535: + return validation_error("port", "invalid_port", port) + if protocol not in ("", "tcp", "udp"): + return validation_error("protocol", "invalid_protocol", protocol) + # serialize concurrent starts; the fd is not inherited by the collector + with open(LOCK, "w") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + stop() + if running_pid() is not None: + return {"error": "session_busy"} + # a collector dying before its first write must not show the old session + if os.path.exists(STATE): + os.remove(STATE) + proc = subprocess.Popen( + ["/usr/libexec/ns-livetrace", ip, str(duration), str(port), protocol], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + # wait for the collector so the next status reports running + for _ in range(50): + if running_pid() is not None or proc.poll() is not None: + break + time.sleep(0.1) + if running_pid() is None: + # kill the late collector and its helpers so no orphan session survives + try: + os.killpg(proc.pid, signal.SIGKILL) + except ProcessLookupError: + pass + proc.wait() + delete_table() + return {"error": "collector_not_started"} + return {"message": "success"} + + +def dpi_flows(): + # ns-flows daemon is optional + try: + resp = requests.get("http://127.0.0.1:8080/flows", timeout=2) + flows = {} + for item in resp.json().get("flows", []): + flow = item.get("flow") or {} + ct_id = (flow.get("conntrack") or {}).get("id") + if ct_id is None: + continue + application = flow.get("detected_application_name") or "" + flows[str(ct_id)] = { + "application": application.removeprefix("netify."), + "dpi_protocol": flow.get("detected_protocol_name") or "", + "host": flow.get("host_server_name") or "", + } + return flows + except Exception: + return {} + + +def route_devs(addresses): + devs = dict.fromkeys(addresses, "") + names = {} + for a in addresses: + try: + names[str(ipaddress.ip_address(a))] = a + except ValueError: + pass + if not names: + return devs + # one ip process for all addresses, results matched by normalized dst + try: + p = subprocess.run( + ["ip", "-j", "-force", "-batch", "-"], + input="".join(f"route get {a}\n" for a in names), + capture_output=True, + text=True, + ) + except OSError: + return devs + for line in p.stdout.splitlines(): + try: + route = json.loads(line)[0] + name = names[str(ipaddress.ip_address(route["dst"]))] + except (ValueError, IndexError, KeyError, TypeError): + continue + if route.get("type") != "local": + devs[name] = route.get("dev", "") + return devs + + +def status(): + try: + with open(STATE) as f: + state = json.load(f) + except (OSError, ValueError): + state = { + "ip": "", + "duration": 0, + "start": 0, + "blocked": [], + "dns": [], + "flows": {}, + "failure": "", + } + running = running_pid() is not None + connections = [] + if running: + seen = set() + untraced = [] + dpi = dpi_flows() + ip = state["ip"] + port = state.get("port", 0) + # conntrack needs a protocol to filter by port + protocols = [state.get("protocol", "")] + if port and not protocols[0]: + protocols = ["tcp", "udp"] + queries = [] + for p in protocols: + proto = ["-p", p] if p else [] + dport = ["--dport", str(port)] if port else [] + sport = ["--reply-port-src", str(port)] if port else [] + queries += [ + ["--orig-src", ip, *proto, *dport], + ["--reply-src", ip, *proto, *sport], + ] + for args in queries: + for c in conntrack.list_connections(extra_args=args): + if c["id"] in seen: + continue + seen.add(c["id"]) + key = "|".join(c.get(k, "") for k in FLOW_FIELDS) + conn = {**c, **state["flows"].get(key, {}), **dpi.get(c["id"], {})} + # not traced, derive interfaces from the routing table + if "in_interface" not in conn: + untraced.append(conn) + connections.append(conn) + devs = route_devs( + {a for c in untraced for a in (c["source"], c["destination"])} + ) + for conn in untraced: + conn["in_interface"] = devs[conn["source"]] + conn["out_interface"] = devs[conn["destination"]] + allowed = defaultdict(list) + if not running: + # traced allowed flows of the last session, source ports merged + for key, flow in state.get("flows", {}).items(): + protocol, source, sport, destination, port = key.split("|") + extra = (flow.get(k, "") for k in ALLOWED_FIELDS[4:]) + allowed[(protocol, source, destination, port, *extra)].append(sport) + return { + "running": running, + "ip": state["ip"], + "duration": state["duration"], + "port": state.get("port", 0), + "protocol": state.get("protocol", ""), + "remaining": max(0, int(state["start"] + state["duration"] - time.time())) + if running + else 0, + "blocked": state["blocked"], + "dns": state["dns"], + "connections": connections, + "allowed": [ + {**dict(zip(ALLOWED_FIELDS, k)), "count": len(p), "source_ports": p} + for k, p in sorted(allowed.items(), key=lambda i: len(i[1]), reverse=True) + ], + "failure": state.get("failure", ""), + "truncated": state.get("truncated", False), + } + + +if __name__ == "__main__": + cmd = sys.argv[1] + if cmd == "list": + print( + json.dumps( + { + "start": { + "ip": "10.10.10.50", + "duration": 120, + "port": 443, + "protocol": "tcp", + }, + "stop": {}, + "status": {}, + } + ) + ) + elif cmd == "call": + action = sys.argv[2] + if action == "start": + print(json.dumps(start(json.load(sys.stdin)))) + elif action == "stop": + # a starting collector has its table but no PID file yet + with open(LOCK, "w") as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + stop() + print(json.dumps({"message": "success"})) + elif action == "status": + print(json.dumps(status())) diff --git a/packages/ns-api/files/ns.livetrace.json b/packages/ns-api/files/ns.livetrace.json new file mode 100644 index 000000000..cc3f8bbcc --- /dev/null +++ b/packages/ns-api/files/ns.livetrace.json @@ -0,0 +1,13 @@ +{ + "livetrace-manager": { + "description": "Trace live traffic of a client", + "write": {}, + "read": { + "ubus": { + "ns.livetrace": [ + "*" + ] + } + } + } +} From 5f7dfc1b0016db21407a5b924489b7d5db6b7d85 Mon Sep 17 00:00:00 2001 From: Giacomo Sanchietti Date: Tue, 10 Nov 2026 14:37:49 +0100 Subject: [PATCH 3/3] chore(ns-ui): TEMP build UI from live trace branch Build the UI from the nethsecurity-ui firewall_debug branch to test the live trace page. Drop this commit before merging. Assisted-by: Claude Code:claude-fable-5-2 --- packages/ns-ui/Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/ns-ui/Makefile b/packages/ns-ui/Makefile index b266e63a4..da3cbd036 100644 --- a/packages/ns-ui/Makefile +++ b/packages/ns-ui/Makefile @@ -12,7 +12,7 @@ PKG_RELEASE:=1 PKG_SOURCE_PROTO:=git PKG_SOURCE_URL:=https://github.com/NethServer/nethsecurity-ui.git -PKG_SOURCE_VERSION:=455432490c9e32175fa4c25d6441022dcf3be51f +PKG_SOURCE_VERSION:=b848c032ae605b4453e25330a8a9efd7b6dfcb0a PKG_SOURCE_SUBDIR:=nethsecurity-ui-$(PKG_SOURCE_VERSION) PKG_BUILD_DIR:=$(BUILD_DIR)/$(PKG_SOURCE_SUBDIR) PKG_MIRROR_HASH:=skip