From bea41b32a1c76de4d1f8b8a33473036a3506336e Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Tue, 16 Jun 2026 16:44:55 +0200 Subject: [PATCH] fix: fix_snmp firewall rule uses ss + correct zone name on OpenWrt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two bugs prevented the firewall step from working: 1. `netstat` was used to detect the SSH peer IP — not installed on OpenWrt by default, so raw_conn was empty and the entire firewall step was silently skipped. 2. Even if detection had worked, `src='*'` is wrong when zones have `input='REJECT'`. The rule only takes effect before the zone policy if `src` is the exact zone name. Fix: switch to `ss` (always present), strip any IPv6-mapped prefix, then walk `uci show firewall` to find the zone whose network interface shares the same /24 as the peer IP. Use that zone name as `src`. Co-Authored-By: Claude Sonnet 4.6 --- napalm_openwrt/system_mixin.py | 732 +++++++++++++++++++++++++++++++++ 1 file changed, 732 insertions(+) create mode 100644 napalm_openwrt/system_mixin.py diff --git a/napalm_openwrt/system_mixin.py b/napalm_openwrt/system_mixin.py new file mode 100644 index 0000000..be1a69f --- /dev/null +++ b/napalm_openwrt/system_mixin.py @@ -0,0 +1,732 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import re +import time as _time +from typing import Any + + +class OpenWrtSystemMixin: + """Mixin providing system-level NAPALM getters (environment, NTP, SNMP, users, services, updates, device actions).""" + + def get_environment(self) -> dict[str, Any]: + """Return device environment data (CPU, memory). + + CPU usage from ``/proc/stat`` (two samples 1 second apart via ``awk``). + Memory from ``/proc/meminfo``. + """ + cpu_out = self._send_command( + "awk '/^cpu /{idle1=$5; total1=$2+$3+$4+$5+$6+$7+$8} END{print (1-(idle1/total1))*100}' /proc/stat" + ) + mem_out = self._send_command("cat /proc/meminfo") + + cpu_pct = 0.0 + try: + cpu_pct = float(cpu_out.strip()) + except (ValueError, AttributeError): + pass + + mem_total = 0 + mem_available = 0 + for line in mem_out.splitlines(): + if line.startswith("MemTotal:"): + try: + mem_total = int(line.split()[1]) + except (IndexError, ValueError): + pass + elif line.startswith("MemAvailable:"): + try: + mem_available = int(line.split()[1]) + except (IndexError, ValueError): + pass + + return { + "fans": {}, + "temperature": {}, + "power": {}, + "cpu": {0: {"%usage": round(cpu_pct, 1)}}, + "memory": { + "available_ram": mem_available * 1024, + "used_ram": (mem_total - mem_available) * 1024, + }, + } + + def get_system_config(self) -> dict[str, Any]: + """Return system-level configuration from UCI. + + Reads ``uci show system`` and ``uci show dropbear`` to collect: + + * hostname (str) + * timezone (str) — POSIX TZ string, e.g. ``"CET-1CEST,M3.5.0,M10.5.0/3"`` + * zonename (str) — human-readable name, e.g. ``"Europe/Berlin"`` + * ntp_servers (list[str]) + * dropbear_port (int) — SSH port + * dropbear_password_auth (bool) — whether password login is allowed + * dropbear_root_password_auth (bool) + """ + sys_out = self._send_command("uci show system 2>/dev/null || true") + db_out = self._send_command("uci show dropbear 2>/dev/null || true") + + sys_cfg: dict[str, str] = {} + for line in sys_out.splitlines(): + m = re.match(r"system\.@system\[0\]\.(\w+)='([^']*)'", line.strip()) + if m: + sys_cfg[m.group(1)] = m.group(2) + + # NTP server list: all servers on a single line, space-separated quoted values + # e.g. system.ntp.server='0.openwrt.pool.ntp.org' '1.openwrt.pool.ntp.org' ... + ntp_servers: list[str] = [] + for line in sys_out.splitlines(): + if re.match(r"system\.ntp\.server=", line.strip()): + ntp_servers = re.findall(r"'([^']+)'", line) + break + + # Dropbear settings + db_cfg: dict[str, str] = {} + for line in db_out.splitlines(): + # May be @dropbear[0] or named section + m = re.match(r"dropbear\.[@\w]+\.(\w+)='([^']*)'", line.strip()) + if m: + db_cfg.setdefault(m.group(1), m.group(2)) + + try: + ssh_port = int(db_cfg.get("Port", "22")) + except (ValueError, TypeError): + ssh_port = 22 + + def _bool_uci(val: str, default: bool = True) -> bool: + return val.lower() not in ("0", "off", "false", "no") if val else default + + return { + "hostname": sys_cfg.get("hostname", ""), + "timezone": sys_cfg.get("timezone", ""), + "zonename": sys_cfg.get("zonename", ""), + "ntp_servers": ntp_servers, + "dropbear_port": ssh_port, + "dropbear_password_auth": _bool_uci(db_cfg.get("PasswordAuth", "on")), + "dropbear_root_password_auth": _bool_uci(db_cfg.get("RootPasswordAuth", "on")), + } + + def get_snmp_information(self) -> dict[str, Any]: + """Return SNMP configuration from ``uci show snmpd``. + + UCI example:: + + snmpd.@com2sec[0].community='public' + snmpd.@com2sec[0].secname='public' + snmpd.@system[0].sysContact='root@localhost' + snmpd.@system[0].sysLocation='Unknown' + """ + uci_out = self._send_command("uci show snmpd") + + contact = "" + location = "" + chassis_id = "" + community: dict[str, Any] = {} + + # Track com2sec entries by index + com2sec: dict[str, dict[str, str]] = {} + + for line in uci_out.splitlines(): + line_s = line.strip() + m = re.match(r"snmpd\.@com2sec\[(\d+)\]\.(\w+)='([^']*)'", line_s) + if m: + idx, key, val = m.group(1), m.group(2), m.group(3) + com2sec.setdefault(idx, {})[key] = val + continue + m = re.match(r"snmpd\.@system\[0\]\.sys(\w+)='([^']*)'", line_s) + if m: + key, val = m.group(1).lower(), m.group(2) + if key == "contact": + contact = val + elif key == "location": + location = val + elif key == "name": + chassis_id = val + + for entry in com2sec.values(): + name = entry.get("community", entry.get("secname", "")) + if not name: + continue + # OpenWrt snmpd doesn't distinguish rw/ro per community via UCI by default + mode = "ro" + if entry.get("secname", "").lower() in ("private", "readwrite", "rw"): + mode = "rw" + community[name] = { + "acl": entry.get("source", "N/A"), + "mode": mode, + } + + return { + "chassis_id": chassis_id, + "community": community, + "contact": contact, + "location": location, + } + + def get_snmp_config(self): + """Return SNMP agent config if snmpd is installed and running on OpenWrt.""" + try: + from napalm_device_types.models import SNMPConfigDict + except ImportError: + return None + + running = ( + self._send_command( + "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive" + ).strip() == "active" + ) + if not running: + return None + + community = "public" + try: + # UCI config (set by luci-app-snmpd) + uci_comm = self._send_command( + "uci -q get snmpd.public.community 2>/dev/null || " + "uci -q get snmpd.@com2sec[0].community 2>/dev/null || echo ''" + ).strip() + if uci_comm: + community = uci_comm + except Exception: + pass + + return SNMPConfigDict(running=True, community=community, port=161, version="2c") + + def _action_fix_snmp(self) -> dict[str, Any]: + """Install and configure snmpd on OpenWrt. + + Installs snmpd-nossl (the daemon) and luci-app-snmpd (UCI schema + + proper procd init script). Configures community 'public' via UCI. + """ + lines: list = [] + + # 1. Install packages — snmpd-nossl (daemon) + luci-app-snmpd (UCI init) + pm = self._pm_type() + if pm == "apk": + raw = self._send_command("apk add snmpd-nossl luci-app-snmpd 2>&1") + else: + self._send_command("opkg update 2>/dev/null || true") + raw = self._send_command("opkg install snmpd-nossl luci-app-snmpd 2>&1") + out = self._clean_pkg_output(raw) + low = out.lower() + installed = not any(kw in low for kw in ("error:", "failed")) + lines.append(f"[install] {out[-300:]}") + + if not installed and "already installed" not in low: + return {"success": False, "output": "\n".join(lines)} + + # 2. Configure via UCI — modify the existing default sections only. + # Do NOT create new named sections (causes duplicate directives in + # the generated /var/run/snmpd.conf which crashes snmpd). + # Also remove any stale named sections from previous fix attempts. + # The init script reads these UCI field names to generate /var/run/snmpd.conf: + # agent: agentaddress + # com2sec: secname, source, community + # group: group (name!), version, secname + # view: viewname (not name!), type, oid + # access: group, version, level, prefix, read, write, notify + # Default luci-app-snmpd schema uses different field names for group/view/access, + # so we patch all required fields explicitly. + uci_cmds = [ + # Remove any stale named sections from previous runs + "uci -q delete snmpd.agent", + "uci -q delete snmpd.public", + # agent + "uci set snmpd.@agent[0].agentaddress='161'", + # com2sec + "uci set snmpd.@com2sec[0].secname='ro'", + "uci set snmpd.@com2sec[0].source='0.0.0.0/0'", + "uci set snmpd.@com2sec[0].community='public'", + # group — init script reads field 'group' (not 'name') + "uci set snmpd.@group[0].group='rogroup'", + "uci set snmpd.@group[0].version='v2c'", + "uci set snmpd.@group[0].secname='ro'", + # view — init script reads field 'viewname' (not 'name') + "uci set snmpd.@view[0].viewname='all'", + "uci set snmpd.@view[0].type='included'", + "uci set snmpd.@view[0].oid='.1'", + # access — init script needs write + notify or it returns early + "uci set snmpd.@access[0].group='rogroup'", + "uci set snmpd.@access[0].context='none'", + "uci set snmpd.@access[0].version='v2c'", + "uci set snmpd.@access[0].level='noAuthNoPriv'", + "uci set snmpd.@access[0].prefix='exact'", + "uci set snmpd.@access[0].read='all'", + "uci set snmpd.@access[0].write='none'", + "uci set snmpd.@access[0].notify='none'", + "uci commit snmpd", + ] + for cmd in uci_cmds: + self._send_command(f"{cmd} 2>/dev/null || true") + lines.append("[config] Configured snmpd via UCI (all required fields set).") + + # 3. Firewall: allow UDP 161 from the management zone + try: + # Use `ss` (always available on OpenWrt) to find the IP of the current SSH client + raw_conn = self._send_command( + "ss -tnp 2>/dev/null | awk '/ESTAB.*:22/{print $5}' | head -1 | sed 's/:[0-9]*$//'" + ).strip() + # Strip IPv6-mapped prefix if present + if raw_conn.startswith("::ffff:"): + raw_conn = raw_conn[7:] + peer_ip = raw_conn.splitlines()[-1].strip() if raw_conn else "" + + if peer_ip and peer_ip not in ("", "0.0.0.0", "::"): + # Determine which firewall zone owns the interface that routes to peer_ip. + # Walk uci firewall zones: find the zone whose associated network interface + # has an address in the same /24 as peer_ip. + peer_prefix = peer_ip.rsplit(".", 1)[0] if "." in peer_ip else "" + fw_out = self._send_command("uci show firewall 2>/dev/null || true") + src_zone = "*" + # Build map zone_name → zone_uci_key + zone_map: dict[str, str] = {} + for fw_line in fw_out.splitlines(): + m = re.match(r"firewall\.(\w+)\.name='([^']+)'", fw_line.strip()) + if m: + zone_map[m.group(2)] = m.group(1) + # For each zone find its network, then the interface IP + for zone_name, uci_key in zone_map.items(): + net_line = self._send_command( + f"uci -q get firewall.{uci_key}.network 2>/dev/null || true" + ).strip() + if not net_line: + continue + for net_name in net_line.split(): + iface_ip = self._send_command( + f"uci -q get network.{net_name}.ipaddr 2>/dev/null || true" + ).strip() + if iface_ip and "." in iface_ip: + iface_prefix = iface_ip.split("/")[0].rsplit(".", 1)[0] + if peer_prefix and iface_prefix == peer_prefix: + src_zone = zone_name + break + if src_zone != "*": + break + + self._send_command( + f"uci -q delete firewall.snmp_netork 2>/dev/null; " + f"uci set firewall.snmp_netork=rule; " + f"uci set firewall.snmp_netork.name='Allow-SNMP-from-mgmt'; " + f"uci set firewall.snmp_netork.src='{src_zone}'; " + f"uci set firewall.snmp_netork.dest_port='161'; " + f"uci set firewall.snmp_netork.proto='udp'; " + f"uci set firewall.snmp_netork.target='ACCEPT'; " + f"uci commit firewall; " + f"/etc/init.d/firewall reload 2>/dev/null || true" + ) + lines.append(f"[firewall] Added UDP:161 allow rule (src zone: {src_zone}).") + else: + lines.append("[firewall] Could not detect peer IP via ss — skipping firewall step.") + except Exception as exc: + lines.append(f"[firewall] skipped — {exc}") + + # 4. Break any crash-loop, then start cleanly + self._send_command("/etc/init.d/snmpd stop 2>/dev/null; true") + _time.sleep(2) + self._send_command("pkill -9 snmpd 2>/dev/null; true") # kill crash-loop zombie + _time.sleep(3) + self._send_command("/etc/init.d/snmpd enable 2>/dev/null; true") + self._send_command("/etc/init.d/snmpd start 2>/dev/null; true") + _time.sleep(4) + lines.append("[service] snmpd started via procd.") + + # 5. Check if snmpd is now active (no local snmpget on OpenWrt by default) + status = self._send_command( + "/etc/init.d/snmpd running 2>/dev/null && echo active || echo inactive" + ).strip() + success = status == "active" + if success: + lines.append("[ok] snmpd is active.") + else: + lines.append(f"[warn] snmpd status: {status}") + + return {"success": success, "output": "\n".join(lines)} + + def get_users(self) -> dict[str, Any]: + """Return users configured on the device. + + Parses ``/etc/passwd`` for accounts with a valid login shell. + SSH public keys are read from ``~/.ssh/authorized_keys`` + (Dropbear also stores root keys at ``/etc/dropbear/authorized_keys``). + + Level mapping: + - UID 0 (root) → 15 (full access) + - all other users → 1 + """ + passwd_out = self._send_command("cat /etc/passwd") + # Root authorized_keys locations on OpenWrt + root_keys_out = self._send_command( + ["cat /root/.ssh/authorized_keys", "cat /etc/dropbear/authorized_keys"] + ) + + users: dict[str, Any] = {} + valid_shells = {"/bin/sh", "/bin/ash", "/bin/bash", "/usr/bin/fish"} + + for line in passwd_out.splitlines(): + parts = line.strip().split(":") + if len(parts) < 7: + continue + username, password_hash, uid_str, _, _, home, shell = ( + parts[0], parts[1], parts[2], parts[3], parts[4], parts[5], parts[6], + ) + if shell not in valid_shells: + continue + try: + uid = int(uid_str) + except ValueError: + continue + + level = 15 if uid == 0 else 1 + + # Collect SSH keys for this user + sshkeys: list[str] = [] + if uid == 0: + for line_k in root_keys_out.splitlines(): + line_k = line_k.strip() + if line_k and not line_k.startswith("#"): + sshkeys.append(line_k) + else: + # Try reading per-user authorized_keys + keys_out = self._send_command(f"cat {home}/.ssh/authorized_keys 2>/dev/null") + for line_k in keys_out.splitlines(): + line_k = line_k.strip() + if line_k and not line_k.startswith("#"): + sshkeys.append(line_k) + + users[username] = { + "level": level, + "password": password_hash, + "sshkeys": sshkeys, + } + + return users + + def get_services(self) -> list[dict[str, Any]]: + """Return all system services with their running and enabled state. + + Uses ``ubus call service list`` for running/PID info and + ``/etc/rc.d/S*`` symlinks for enabled-at-boot state. + """ + import json as _json + + # -- enabled set: names from /etc/rc.d/S symlinks ---- + rc_out = self._send_command( + "ls /etc/rc.d/ 2>/dev/null | grep '^S' | sed 's/^S[0-9]*//'" + ) + enabled: set[str] = {s.strip() for s in rc_out.splitlines() if s.strip()} + + # -- running info from procd via ubus --------------------------------- + ubus_raw = self._send_command("ubus call service list 2>/dev/null") + ubus_data: dict = {} + try: + ubus_data = _json.loads(ubus_raw) + except (ValueError, TypeError): + pass + + # Build index from ubus data + service_map: dict[str, dict] = {} + for svc_name, svc_info in ubus_data.items(): + if not isinstance(svc_info, dict): + continue + instances = svc_info.get("instances", {}) + running = any( + inst.get("running", False) + for inst in instances.values() + if isinstance(inst, dict) + ) + pid = next( + ( + inst.get("pid", 0) + for inst in instances.values() + if isinstance(inst, dict) and inst.get("running") + ), + 0, + ) + service_map[svc_name] = {"running": running, "pid": pid} + + # -- all init scripts ------------------------------------------------- + init_raw = self._send_command("ls -1 /etc/init.d/ 2>/dev/null") + init_scripts: set[str] = {s.strip() for s in init_raw.splitlines() if s.strip()} + + # Merge: all known services (from init.d + ubus) + all_names = init_scripts | set(service_map.keys()) + # Exclude procd internal pseudo-service + all_names.discard("") + + result: list[dict[str, Any]] = [] + for name in sorted(all_names): + info = service_map.get(name, {}) + result.append({ + "name": name, + "running": info.get("running", False), + "enabled": name in enabled, + "pid": info.get("pid", 0), + }) + + return result + + def manage_service(self, name: str, action: str) -> dict[str, Any]: + """Execute a lifecycle action (start/stop/restart/enable/disable) on a service.""" + import re as _re + if not _re.match(r'^[a-zA-Z0-9_\-]+$', name): + raise ValueError(f"Invalid service name: {name!r}") + if action not in ('start', 'stop', 'restart', 'enable', 'disable'): + raise ValueError(f"Invalid action: {action!r}") + output = self._send_command(f"/etc/init.d/{name} {action} 2>&1") + return {"success": True, "output": output} + + def get_device_warnings(self) -> list[dict[str, Any]]: + """Return a list of warning dicts for issues detected on this device. + + Currently detects: + - lldpd not installed (LLDP neighbor discovery unavailable) + - package updates available (uses local package cache, no network call) + - update notifications not configured (auc not installed, opkg only) + """ + warnings: list[dict[str, Any]] = [] + + # 1. LLDP daemon + lldpd_path = self._send_command("which lldpd 2>/dev/null").strip() + if not lldpd_path: + warnings.append({ + "code": "lldpd_not_installed", + "severity": "warning", + "action": "install_lldpd", + }) + + pm = self._pm_type() + + # 2. Package updates available (local cache only – no opkg update) + def _pkg_name(line: str, pm_type: str) -> str: + """Extract just the package name from an upgradable line. + + apk: 'luci-app-firewall-26.143~abc < 26.151~def' → 'luci-app-firewall' + opkg: 'luci-app-firewall - 1.0 - 1.1' → 'luci-app-firewall' + """ + import re as _re + if pm_type == "apk": + # Strip trailing ' < ...' then remove the version suffix (-\d...) + name_ver = line.split(" ")[0] + m = _re.match(r'^(.*?)-\d', name_ver) + return m.group(1) if m else name_ver + else: + return line.split(" - ")[0].strip() + + try: + if pm == "apk": + raw_upg = self._send_command("apk version 2>/dev/null | grep '<'") + else: + raw_upg = self._send_command("opkg list-upgradable 2>/dev/null") + upgradable = [ln.strip() for ln in raw_upg.splitlines() if ln.strip()] + except Exception: + upgradable = [] + + if upgradable: + pkg_names = [_pkg_name(ln, pm) for ln in upgradable] + warnings.append({ + "code": "updates_available", + "severity": "info", + "action": None, + "meta": { + "count": len(upgradable), + "packages": pkg_names[:10], + }, + }) + + # 3. Attended sysupgrade client not installed (opkg systems only) + if pm == "opkg": + auc_path = self._send_command("which auc 2>/dev/null").strip() + if not auc_path: + warnings.append({ + "code": "update_notifications_disabled", + "severity": "warning", + "action": "install_auc", + }) + + # 4. base64 not available — needed for efficient config apply + b64_path = self._send_command("command -v base64 2>/dev/null").strip() + if not b64_path: + warnings.append({ + "code": "no_base64", + "severity": "warning", + "action": "install_coreutils_base64", + }) + + return warnings + + def get_available_updates(self) -> list[dict[str, Any]]: + """Return list of upgradable packages from the local package manager cache.""" + import re as _re + pm = self._pm_type() + updates: list[dict[str, Any]] = [] + + if pm == "apk": + # Output format: "pkgname-current_ver < new_ver" + raw = self._send_command("apk version 2>/dev/null | grep '<'") + for line in raw.splitlines(): + line = line.strip() + m = _re.match(r'^(.+)-(\d\S*)\s+<\s+(\S+)', line) + if m: + updates.append({ + "name": m.group(1), + "current_version": m.group(2), + "new_version": m.group(3), + }) + else: + # opkg output: "pkgname - current_ver - new_ver" + raw = self._send_command("opkg list-upgradable 2>/dev/null") + for line in raw.splitlines(): + parts = [p.strip() for p in line.split(" - ")] + if len(parts) == 3: + updates.append({ + "name": parts[0], + "current_version": parts[1], + "new_version": parts[2], + }) + + return sorted(updates, key=lambda u: u["name"]) + + def apply_updates(self, packages: list[str]) -> dict[str, Any]: + """Upgrade the given packages using the device's package manager.""" + import re as _re + for pkg in packages: + if not _re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg): + raise ValueError(f"Invalid package name: {pkg!r}") + pm = self._pm_type() + pkg_args = " ".join(packages) + if pm == "apk": + cmd = f"apk upgrade {pkg_args} 2>&1" + else: + cmd = f"opkg upgrade {pkg_args} 2>&1" + output = self._send_command(cmd) + return {"success": True, "output": output} + + # ------------------------------------------------------------------ + # NTP + # ------------------------------------------------------------------ + + def get_ntp_servers(self) -> dict[str, Any]: + """Return configured NTP servers from ``uci show system``. + + UCI example:: + + system.ntp.server='0.openwrt.pool.ntp.org 1.openwrt.pool.ntp.org' + """ + uci_out = self._send_command("uci show system") + servers: dict[str, Any] = {} + + for line in uci_out.splitlines(): + # Handles both list and single-value UCI representations + m = re.match(r"system\.ntp\.server(?:\[\d+\])?='([^']*)'", line.strip()) + if m: + for srv in m.group(1).split(): + srv = srv.strip() + if srv: + servers[srv] = {} + + return servers + + def get_ntp_peers(self) -> dict[str, Any]: + """Return NTP peers from ``uci show system``. + + OpenWrt's busybox ntpd does not differentiate peers from servers; + the same UCI ``ntp.server`` list is returned. + """ + return self.get_ntp_servers() + + def get_ntp_stats(self) -> list[dict[str, Any]]: + """Return NTP synchronisation statistics. + + Tries ``ntpq -pn`` first (ntpd), then ``chronyc sources -v`` (chrony). + Returns an empty list when neither tool is available. + + ``ntpq -pn`` example line:: + + *188.114.101.4 188.114.100.1 4 u 107 256 377 164.228 -13.866 2.695 + + ``chronyc sources -v`` example line:: + + ^* 192.168.1.1 2 6 17 8 +2345us[ 0ns] +/- 15ms + """ + ntpq_out = self._send_command("ntpq -pn") + if ntpq_out and not ntpq_out.startswith(("ntpq: ", "sh: ", "ash: ", "command not found")): + return self._parse_ntpq(ntpq_out) + + chrony_out = self._send_command("chronyc sources -v") + if chrony_out and not chrony_out.startswith(("sh: ", "ash: ", "command not found")): + return self._parse_chronyc(chrony_out) + + return [] + + @staticmethod + def _parse_ntpq(output: str) -> list[dict[str, Any]]: + """Parse ``ntpq -pn`` tabular output.""" + stats = [] + for line in output.splitlines(): + line_s = line.strip() + if not line_s or line_s.startswith(("remote", "=")): + continue + # First char is the tally code (* = synchronized, + = candidate, etc.) + tally = line_s[0] if line_s[0] in "* +-x.o#" else " " + parts = line_s[1:].split() + if len(parts) < 10: + continue + try: + stats.append({ + "remote": parts[0], + "referenceid": parts[1], + "synchronized": tally == "*", + "stratum": int(parts[2]), + "type": parts[3], + "when": parts[4], + "hostpoll": int(parts[5]), + "reachability": int(parts[6], 8), # octal + "delay": float(parts[7]), + "offset": float(parts[8]), + "jitter": float(parts[9]), + }) + except (ValueError, IndexError): + continue + return stats + + @staticmethod + def _parse_chronyc(output: str) -> list[dict[str, Any]]: + """Parse ``chronyc sources -v`` tabular output.""" + stats = [] + for line in output.splitlines(): + line_s = line.strip() + # Data lines start with ^* ^+ ^- ^? + m = re.match(r"^(\^[*+\-?])\s+(\S+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.*)", line_s) + if not m: + continue + tally = m.group(1) + try: + stats.append({ + "remote": m.group(2), + "referenceid": "", + "synchronized": tally == "^*", + "stratum": int(m.group(3)), + "type": "u", + "when": m.group(6), + "hostpoll": int(m.group(4)), + "reachability": int(m.group(5), 8) if re.match(r"^[0-7]+$", m.group(5)) else 0, + "delay": 0.0, + "offset": 0.0, + "jitter": 0.0, + }) + except (ValueError, IndexError): + continue + return stats