fix: fix_snmp firewall rule uses ss + correct zone name on OpenWrt
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 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
co-authored by
Claude Sonnet 4.6
Claude Sonnet 4.6
parent
85041c13bb
commit
bea41b32a1
@@ -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 ``~<user>/.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<priority><name> 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
|
||||||
Reference in New Issue
Block a user