Christian Manivong
ee0200462b
CI / test (3.10) (push) Successful in 28s
CI / test (3.11) (push) Successful in 23s
CI / test (3.12) (push) Successful in 29s
CI / test (3.10) (pull_request) Successful in 26s
CI / test (3.11) (pull_request) Successful in 30s
CI / test (3.12) (pull_request) Successful in 32s
fix_snmp wrote an application's snmpd.conf (community public, Managed by netOrk, netork@localhost) and judged itself by a local snmpget. The drivers implement napalm-device-types 4.13's SnmpAgentMixin steps instead: net-snmp installed, snmpd.conf with one rocommunity line per allowed source (staged without privilege, installed by root), snmpd enabled and restarted. pf and ipfw rules stay the operator's. A community or text that would break the file is refused. The caller verifies with its own SNMP client. Also get_management_peer() from SSH_CONNECTION, ensure_package from the base package, and get_snmp_config no longer assumes "public". BREAKING CHANGE: run_device_action is removed. Requires napalm-device-types 4.13. Part of #12 (the OpenBSD new_version text is still open).
542 lines
23 KiB
Python
542 lines
23 KiB
Python
"""What FreeBSD and OpenBSD share: SSH, and the readers whose tools agree.
|
|
|
|
Every command runs on its own SSH exec channel (napalm-device-types'
|
|
``run_on_transport``): no PTY to parse a prompt from, stdout and stderr apart,
|
|
and a real exit status. Root comes from ``sudo -S`` with the sudo password on
|
|
stdin, as napalm-linux does it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import logging
|
|
import re
|
|
import socket
|
|
from shlex import quote
|
|
from typing import Any, Optional
|
|
|
|
import paramiko
|
|
from napalm.base.exceptions import ConnectionClosedException, ConnectionException
|
|
from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin
|
|
from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict
|
|
from napalm_device_types.host_status import HostStatusMixin
|
|
from napalm_device_types.channel import (
|
|
ByteStream,
|
|
CommandResult,
|
|
open_stream_on_transport,
|
|
run_on_transport,
|
|
)
|
|
|
|
from napalm_bsd import parse
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: The update-list entry for base-system patches (OpenBSD syspatch, classic
|
|
#: FreeBSD freebsd-update): one entry, as they are applied together (#799).
|
|
BASE_SYSTEM = "base-system"
|
|
|
|
_SERVICE_ACTIONS = frozenset({"start", "stop", "restart", "enable", "disable"})
|
|
_SERVICE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
|
|
_PACKAGE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.+-]*$")
|
|
|
|
#: What may stand in an snmpd.conf value: no whitespace, no quote, no comment.
|
|
_SNMP_WORD = re.compile(r'^[^\s"#]+$')
|
|
#: The SSH client side of this connection: the address the host sees it come from.
|
|
_PEER_COMMAND = 'printf "%s\\n" "$SSH_CONNECTION"'
|
|
#: Why ``run_command`` refuses stdin to a privileged command while a sudo
|
|
#: password is set (#8).
|
|
_PRIVILEGED_STDIN = (
|
|
"run_command(privileged=True) takes no stdin while a sudo password is set: "
|
|
"sudo -S reads the password from stdin, and a sudo that does not prompt "
|
|
"(NOPASSWD, a cached timestamp) leaves it for the command, which would read it "
|
|
"as the first line of the data. Stage the data without privilege and have root "
|
|
"install that file."
|
|
)
|
|
|
|
#: Stages content without privilege: a fresh ``mktemp`` file only the login user
|
|
#: may read, never a fixed name another user could create first. Prints its path.
|
|
_STAGE_TEMPLATE = "/tmp/napalm-bsd-stage.XXXXXX"
|
|
_STAGE_PREFIX = _STAGE_TEMPLATE[: -len("XXXXXX")]
|
|
_STAGE = (
|
|
f"umask 077 && t=$(mktemp {_STAGE_TEMPLATE}) && "
|
|
'{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }'
|
|
)
|
|
|
|
def _source(value: str) -> str:
|
|
"""An allowed source as an address or network, or ValueError."""
|
|
return str(ipaddress.ip_network(value.strip(), strict=False))
|
|
|
|
|
|
def render_snmpd_conf(settings: SnmpAgentSettingsDict) -> str:
|
|
"""net-snmp's configuration for *settings*: read-only v2c, per allowed source."""
|
|
community = settings["community"]
|
|
if not _SNMP_WORD.match(community):
|
|
raise ValueError("the community must be one word without quotes or #")
|
|
listen = settings.get("listen")
|
|
lines = [f"agentAddress udp:{listen + ':' if listen else ''}161"]
|
|
sources = [_source(s) for s in settings.get("allowed_sources", [])]
|
|
lines += [f"rocommunity {community} {s}" for s in sources] or [f"rocommunity {community}"]
|
|
for key, directive in (("location", "sysLocation"), ("contact", "sysContact")):
|
|
value = settings.get(key, "")
|
|
if value:
|
|
if "\n" in value:
|
|
raise ValueError(f"{key} must be one line")
|
|
lines.append(f"{directive} {value}")
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver):
|
|
"""Base for the BSD drivers; a concrete one names the commands that differ."""
|
|
|
|
VENDOR = ""
|
|
#: A full init sequence, as on any general-purpose host.
|
|
REBOOT_SETTLE_SECONDS = 90
|
|
|
|
#: Commands whose wording differs between the BSDs.
|
|
HOSTNAME_COMMAND = "hostname"
|
|
OS_VERSION_COMMAND = "uname -sr"
|
|
KERNEL_COMMAND = "uname -r"
|
|
#: Prints ``key=value`` lines; see :meth:`_platform`.
|
|
PLATFORM_COMMAND = ""
|
|
ROUTES_COMMAND = "netstat -rn"
|
|
#: Lists listening sockets with their processes; as root it names them all.
|
|
LISTENING_COMMAND = ""
|
|
#: Lists them without root, if the command above needs it to see them all.
|
|
LISTENING_FALLBACK_COMMAND = ""
|
|
#: lstart is five words; :func:`parse.processes` reads it as one field.
|
|
PS_COMMAND = "ps -axww -o user,pid,ppid,%cpu,%mem,vsz,rss,stat,lstart,command"
|
|
|
|
def __init__(
|
|
self,
|
|
hostname: str,
|
|
username: str,
|
|
password: str,
|
|
timeout: int = 60,
|
|
optional_args: Optional[dict] = None,
|
|
) -> None:
|
|
self.hostname = hostname
|
|
self.username = username
|
|
self.password = password
|
|
self.timeout = timeout
|
|
args = optional_args or {}
|
|
self.port: int = int(args.get("port", 22))
|
|
self._key_file: Optional[str] = args.get("key_file")
|
|
self._sudo_password: Optional[str] = args.get("sudo_password")
|
|
self._allow_agent: bool = bool(args.get("allow_agent", False))
|
|
self._look_for_keys: bool = bool(args.get("look_for_keys", False))
|
|
self._client: Optional[paramiko.SSHClient] = None
|
|
self._root: Optional[bool] = None
|
|
|
|
# -- connection --------------------------------------------------------------
|
|
|
|
def open(self) -> None:
|
|
client = paramiko.SSHClient()
|
|
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
try:
|
|
client.connect(
|
|
self.hostname,
|
|
port=self.port,
|
|
username=self.username,
|
|
password=self.password or None,
|
|
key_filename=self._key_file,
|
|
allow_agent=self._allow_agent,
|
|
look_for_keys=self._look_for_keys,
|
|
timeout=self.timeout,
|
|
)
|
|
except (paramiko.SSHException, socket.error) as exc:
|
|
client.close()
|
|
raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc
|
|
self._client = client
|
|
self._root = None
|
|
|
|
def close(self) -> None:
|
|
if self._client:
|
|
self._client.close()
|
|
self._client = None
|
|
|
|
def is_alive(self) -> dict[str, bool]:
|
|
transport = self._client.get_transport() if self._client else None
|
|
return {"is_alive": bool(transport and transport.is_active())}
|
|
|
|
def _transport(self) -> Any:
|
|
transport = self._client.get_transport() if self._client else None
|
|
if transport is None or not transport.is_active():
|
|
raise ConnectionClosedException("Not connected")
|
|
return transport
|
|
|
|
# -- command channel (napalm-device-types CommandChannelMixin) ---------------
|
|
|
|
def _is_root(self) -> bool:
|
|
if self._root is None:
|
|
self._root = run_on_transport(self._transport(), "id -u").stdout.strip() == "0"
|
|
return self._root
|
|
|
|
def _privileged(self, command: str, privileged: bool) -> tuple[str, Optional[bytes]]:
|
|
"""The command line that runs *command* as asked, and what goes to stdin first.
|
|
|
|
With a sudo password, ``sudo -S`` reads it from stdin, so it never
|
|
appears in a process list; without one, ``sudo -n`` fails at once
|
|
where a prompt would hang.
|
|
"""
|
|
if not privileged or self._is_root():
|
|
return command, None
|
|
if self._sudo_password:
|
|
return f"sudo -S -p '' sh -c {quote(command)}", f"{self._sudo_password}\n".encode()
|
|
return f"sudo -n sh -c {quote(command)}", None
|
|
|
|
def run_command(
|
|
self,
|
|
command: str,
|
|
*,
|
|
privileged: bool = False,
|
|
timeout: float = 60,
|
|
stdin: Optional[bytes] = None,
|
|
) -> CommandResult:
|
|
"""Run *command* on an exec channel: no PTY, stderr apart, a real exit code.
|
|
|
|
:raises ValueError: for *stdin* to a privileged command while a sudo
|
|
password is set (see ``_PRIVILEGED_STDIN``).
|
|
"""
|
|
line, prefix = self._privileged(command, privileged)
|
|
if prefix and stdin is not None:
|
|
raise ValueError(_PRIVILEGED_STDIN)
|
|
return run_on_transport(self._transport(), line, stdin=prefix or stdin, timeout=timeout)
|
|
|
|
def _install_file_as_root(self, path: str, content: str) -> CommandResult:
|
|
"""Replace *path* with *content*, as a new file owned by root, mode 644.
|
|
|
|
The content is staged without privilege (``_STAGE``), so it never follows
|
|
a sudo password on stdin. Root then installs the staged file next to *path*
|
|
and renames it over *path*; nothing root runs follows a symlink at *path*.
|
|
Nothing runs as root when staging failed.
|
|
"""
|
|
staged = self.run_command(_STAGE, timeout=30, stdin=content.encode())
|
|
lines = staged.stdout.strip().splitlines()
|
|
tmp = lines[-1] if lines else ""
|
|
if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]:
|
|
reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}"
|
|
return CommandResult(reason, "", staged.exit_code or 1)
|
|
tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path))
|
|
directory = quote(path.rsplit("/", 1)[0] or "/")
|
|
return self.run_command(
|
|
f"mkdir -p {directory} && install -m 644 -o root -g wheel {tmp_q} {part_q}"
|
|
f" && mv -f {part_q} {path_q}; s=$?; rm -f {tmp_q} {part_q}; exit $s",
|
|
privileged=True,
|
|
timeout=30,
|
|
)
|
|
|
|
def open_stream(self, command: str, *, privileged: bool = False) -> ByteStream:
|
|
line, prefix = self._privileged(command, privileged)
|
|
return open_stream_on_transport(self._transport(), line, stdin_prefix=prefix)
|
|
|
|
def _read(
|
|
self,
|
|
command: str,
|
|
*,
|
|
privileged: bool = False,
|
|
timeout: float = 60,
|
|
ok: tuple[int, ...] = (0,),
|
|
) -> CommandResult:
|
|
"""Run a reader's command; raise when it did not read.
|
|
|
|
A reader that cannot read raises rather than return "nothing": an empty
|
|
update list would close every patch clock netOrk keeps (napalm-bsd#4).
|
|
"""
|
|
result = self.run_command(command, privileged=privileged, timeout=timeout)
|
|
if result.exit_code not in ok:
|
|
reason = (result.stderr or result.stdout).strip() or f"exit {result.exit_code}"
|
|
raise RuntimeError(f"{command}: {reason}")
|
|
return result
|
|
|
|
def _out(self, command: str, *, privileged: bool = False, timeout: float = 60) -> str:
|
|
"""What *command* printed. A failing command prints nothing useful,
|
|
and the readers below treat empty output as "nothing there"."""
|
|
return self.run_command(command, privileged=privileged, timeout=timeout).stdout.strip()
|
|
|
|
#: A file that exists until the host's first boot is over; "" when there is none to ask.
|
|
FIRST_BOOT_MARKER = ""
|
|
|
|
def first_boot_pending(self) -> bool:
|
|
"""Whether the host is still in its first boot.
|
|
|
|
A FreeBSD cloud image upgrades its base system on the first boot, starts
|
|
sshd only after that and restarts right away; ``/etc/rc`` removes
|
|
``/firstboot`` just before the restart. netOrk waits for this before it
|
|
sets up a new VM (NetOrk/netork#795).
|
|
"""
|
|
if not self.FIRST_BOOT_MARKER:
|
|
return False
|
|
return self.run_command(f"test -e {self.FIRST_BOOT_MARKER}").exit_code == 0
|
|
|
|
def _run_host_status_command(self, command: str) -> str:
|
|
"""The transport for ``HostStatusMixin.get_host_status``: read-only, no root.
|
|
|
|
On FreeBSD the report compares ``freebsd-version -k`` with ``-r``
|
|
(napalm-device-types 4.1); elsewhere "reboot required" stays unknown.
|
|
"""
|
|
return self.run_command(command).stdout
|
|
|
|
# -- facts -------------------------------------------------------------------
|
|
|
|
def _platform(self) -> dict[str, str]:
|
|
"""``key=value`` lines from :attr:`PLATFORM_COMMAND`."""
|
|
if not self.PLATFORM_COMMAND:
|
|
return {}
|
|
pairs = (line.partition("=") for line in self._out(self.PLATFORM_COMMAND).splitlines())
|
|
return {key.strip(): value.strip() for key, sep, value in pairs if sep}
|
|
|
|
def _hardware(self) -> tuple[str, str, str]:
|
|
"""``(vendor, model, serial)``; each concrete driver reads its own source."""
|
|
return "", "", ""
|
|
|
|
def _os_version(self) -> str:
|
|
return self._out(self.OS_VERSION_COMMAND)
|
|
|
|
def _uptime(self) -> int:
|
|
booted = parse.boottime(self._out("sysctl -n kern.boottime"))
|
|
now = self._out("date +%s")
|
|
if booted is None or not now.isdigit():
|
|
return -1
|
|
return max(int(now) - booted, 0)
|
|
|
|
def get_facts(self) -> dict[str, Any]:
|
|
fqdn = self._out(self.HOSTNAME_COMMAND)
|
|
vendor, model, serial = self._hardware()
|
|
interfaces = parse.ifconfig(self._out("ifconfig -a"))
|
|
return {
|
|
"hostname": fqdn.split(".")[0],
|
|
"fqdn": fqdn,
|
|
"vendor": vendor or self.VENDOR,
|
|
"model": model,
|
|
"serial_number": serial,
|
|
"os_version": self._os_version(),
|
|
"uptime": self._uptime(),
|
|
"interface_list": [n for n, i in interfaces.items() if not i["loopback"]],
|
|
"running_kernel": self._out(self.KERNEL_COMMAND),
|
|
}
|
|
|
|
# -- interfaces, routes, neighbours ------------------------------------------
|
|
|
|
def get_interfaces(self) -> dict[str, Any]:
|
|
return {
|
|
name: {
|
|
"is_up": iface["up"],
|
|
"is_enabled": iface["enabled"],
|
|
"description": iface["description"],
|
|
"last_flapped": -1.0,
|
|
"speed": iface["speed"],
|
|
"mtu": iface["mtu"],
|
|
"mac_address": iface["mac"],
|
|
}
|
|
for name, iface in parse.ifconfig(self._out("ifconfig -a")).items()
|
|
}
|
|
|
|
def get_interfaces_ip(self) -> dict[str, Any]:
|
|
result: dict[str, Any] = {}
|
|
for name, iface in parse.ifconfig(self._out("ifconfig -a")).items():
|
|
if not (iface["ipv4"] or iface["ipv6"]):
|
|
continue
|
|
result[name] = {
|
|
family: {addr: {"prefix_length": prefix} for addr, prefix in iface[family].items()}
|
|
for family in ("ipv4", "ipv6")
|
|
}
|
|
return result
|
|
|
|
def get_route_to(
|
|
self, destination: str = "", protocol: str = "", longer: bool = False
|
|
) -> dict[str, list[dict[str, Any]]]:
|
|
"""The routing table, keyed by network, in NAPALM's route shape."""
|
|
routes: dict[str, list[dict[str, Any]]] = {}
|
|
for route in parse.routes(self._out(self.ROUTES_COMMAND)):
|
|
if destination and route["network"] != destination:
|
|
continue
|
|
if protocol and route["protocol"] != protocol.lower():
|
|
continue
|
|
routes.setdefault(route["network"], []).append(
|
|
{
|
|
"protocol": route["protocol"],
|
|
"family": route["family"],
|
|
"current_active": True,
|
|
"last_active": False,
|
|
"age": -1,
|
|
"next_hop": route["next_hop"],
|
|
"outgoing_interface": route["interface"],
|
|
"selected_next_hop": True,
|
|
"preference": 0,
|
|
"routing_table": "global",
|
|
"protocol_attributes": {},
|
|
}
|
|
)
|
|
return routes
|
|
|
|
def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
|
|
return parse.arp(self._out("arp -an"))
|
|
|
|
def get_lldp_neighbors(self) -> dict[str, list[dict[str, Any]]]:
|
|
"""Neither BSD ships an LLDP daemon in its base system."""
|
|
return {}
|
|
|
|
# -- listening sockets -------------------------------------------------------
|
|
|
|
def _parse_listening(self, output: str, *, attributed: bool) -> list[dict[str, Any]]:
|
|
"""Parse :attr:`LISTENING_COMMAND` (*attributed*) or its fallback."""
|
|
raise NotImplementedError
|
|
|
|
def get_listening_sockets(self) -> dict[str, Any]:
|
|
"""Every listening TCP and bound UDP socket, with the process holding it.
|
|
|
|
Same shape as napalm-device-types' ``ListeningSocketsMixin`` -- whose
|
|
``ss``/cgroup reading is Linux's -- and the same rule: read as root
|
|
first, and without root when that brings nothing back, which the
|
|
reading then says (``attributed: False``). BSD has no systemd units or
|
|
container IDs to name.
|
|
"""
|
|
privileged = self.run_command(self.LISTENING_COMMAND, privileged=True)
|
|
if privileged.exit_code == 0 and privileged.stdout.strip():
|
|
return {
|
|
"attributed": True,
|
|
"sockets": self._parse_listening(privileged.stdout, attributed=True),
|
|
}
|
|
command = self.LISTENING_FALLBACK_COMMAND or self.LISTENING_COMMAND
|
|
plain = self.run_command(command, privileged=False)
|
|
return {
|
|
"attributed": False,
|
|
"sockets": self._parse_listening(plain.stdout, attributed=False),
|
|
}
|
|
|
|
# -- packages -----------------------------------------------------------------
|
|
|
|
#: ``{name}`` is the package (with its version, when one is asked for).
|
|
INSTALL_COMMAND = ""
|
|
UNINSTALL_COMMAND = ""
|
|
|
|
def _package_command(self, template: str, name: str) -> None:
|
|
"""Run a package command as root; a failure raises with what it printed."""
|
|
result = self.run_command(template.format(name=name), privileged=True, timeout=600)
|
|
if result.exit_code != 0:
|
|
raise RuntimeError(
|
|
(result.stderr or result.stdout).strip() or f"{name}: exit {result.exit_code}"
|
|
)
|
|
|
|
def install_package(self, name: str, version: str = "") -> None:
|
|
if not _PACKAGE_NAME.match(name) or (version and not _PACKAGE_NAME.match(version)):
|
|
raise ValueError(f"Not a package name: {name!r}")
|
|
self._package_command(self.INSTALL_COMMAND, f"{name}-{version}" if version else name)
|
|
|
|
def uninstall_package(self, name: str) -> None:
|
|
if not _PACKAGE_NAME.match(name):
|
|
raise ValueError(f"Not a package name: {name!r}")
|
|
self._package_command(self.UNINSTALL_COMMAND, name)
|
|
|
|
# -- services -----------------------------------------------------------------
|
|
|
|
#: Prints ``name<TAB>status`` per enabled service; parsed by :meth:`_parse_services`.
|
|
SERVICE_STATUS_COMMAND = ""
|
|
#: ``{name}`` and ``{action}`` (start, stop, restart, enable, disable).
|
|
SERVICE_ACTION_COMMAND = ""
|
|
|
|
def _parse_services(self, output: str) -> list[dict[str, Any]]:
|
|
raise NotImplementedError
|
|
|
|
def get_services(self) -> list[dict[str, Any]]:
|
|
"""The enabled services, whether each runs, and its PID where known.
|
|
|
|
Read as root where the status needs it, and without root otherwise.
|
|
"""
|
|
result = self.run_command(self.SERVICE_STATUS_COMMAND, privileged=True, timeout=120)
|
|
if result.exit_code != 0 or not result.stdout.strip(): # no root: sudo refused
|
|
result = self.run_command(self.SERVICE_STATUS_COMMAND, timeout=120)
|
|
return self._parse_services(result.stdout)
|
|
|
|
def manage_service(self, name: str, action: str) -> dict[str, Any]:
|
|
"""Applies *action* to the service *name*, as root.
|
|
|
|
:returns: ``{"success": bool, "output": str}``
|
|
:raises ValueError: for an unknown action or an invalid name, before
|
|
anything is sent.
|
|
"""
|
|
if action not in _SERVICE_ACTIONS or not _SERVICE_NAME.match(name):
|
|
raise ValueError(f"Cannot {action!r} service {name!r}")
|
|
command = self.SERVICE_ACTION_COMMAND.format(name=name, action=action)
|
|
result = self.run_command(command, privileged=True, timeout=120)
|
|
output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
|
|
return {"success": result.exit_code == 0, "output": output}
|
|
|
|
# -- SNMP (net-snmp from packages, NetOrk/netork#800) -------------------------
|
|
|
|
#: Where net-snmp reads its configuration, and how its daemon is started.
|
|
SNMPD_CONF = ""
|
|
SNMPD_START = ""
|
|
#: net-snmp's daemon; OpenBSD's base snmpd is /usr/sbin/snmpd.
|
|
SNMPD_DAEMON = "/usr/local/sbin/snmpd"
|
|
|
|
def get_management_peer(self) -> Optional[str]:
|
|
"""The address this host sees the SSH connection come from (``SSH_CONNECTION``)."""
|
|
out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split()
|
|
try:
|
|
return str(ipaddress.ip_address(out[0])) if out else None
|
|
except ValueError:
|
|
return None
|
|
|
|
# The steps of napalm-device-types' configure_snmp_agent: net-snmp rather than
|
|
# the base daemons (FreeBSD bsnmpd, OpenBSD snmpd), as it answers UCD-SNMP-MIB.
|
|
|
|
def _step(self, command: str, done: str) -> FixStepOutcomeDict:
|
|
result = self.run_command(command, privileged=True, timeout=600)
|
|
output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip())))
|
|
return {"ok": result.exit_code == 0, "output": done if result.exit_code == 0 else output[-300:]}
|
|
|
|
def _install_snmp_agent(self) -> FixStepOutcomeDict:
|
|
return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed")
|
|
|
|
def _write_snmp_agent_config(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
|
|
written = self._install_file_as_root(self.SNMPD_CONF, render_snmpd_conf(settings))
|
|
if written.exit_code:
|
|
return {"ok": False, "output": (written.stdout or written.stderr).strip()[-300:]}
|
|
return {"ok": True, "output": f"wrote {self.SNMPD_CONF}"}
|
|
|
|
def _allow_snmp_sources(self, settings: SnmpAgentSettingsDict) -> FixStepOutcomeDict:
|
|
# pf and ipfw rule sets are the operator's; the agent itself answers only
|
|
# the allowed sources (one rocommunity line each).
|
|
return {"ok": True, "output": "pf/ipfw unchanged; snmpd answers only the allowed sources"}
|
|
|
|
def _start_snmp_agent(self) -> FixStepOutcomeDict:
|
|
return self._step(self.SNMPD_START, "snmpd enabled and running")
|
|
|
|
def get_snmp_config(self) -> Optional[dict[str, Any]]:
|
|
"""net-snmp's community and port, if its daemon runs; None otherwise."""
|
|
if not self._out(f"pgrep -f {self.SNMPD_DAEMON}"):
|
|
return None
|
|
# A community that is not in the file is not known; no default is assumed.
|
|
community, port = "", 161
|
|
for line in self._out(f"cat {self.SNMPD_CONF}").splitlines():
|
|
words = line.split()
|
|
if len(words) >= 2 and words[0].lower() in (
|
|
"rocommunity",
|
|
"rwcommunity",
|
|
"rocommunity6",
|
|
):
|
|
community = words[1]
|
|
elif words and words[0] == "agentAddress":
|
|
found = re.search(r":(\d+)", line)
|
|
port = int(found[1]) if found else port
|
|
return {"running": True, "community": community, "port": port, "version": "2c"}
|
|
|
|
# -- accounts, processes, cron -----------------------------------------------
|
|
|
|
def get_os_users(self) -> list[dict[str, Any]]:
|
|
return parse.users(self._out("cat /etc/passwd"), self._out("cat /etc/group"))
|
|
|
|
def get_processes(self) -> list[dict[str, Any]]:
|
|
return parse.processes(self._out(self.PS_COMMAND))
|
|
|
|
def get_cron_jobs(self) -> list[dict[str, str]]:
|
|
"""The system crontab and the login user's own.
|
|
|
|
Other users' crontabs (``/var/cron/tabs``) are root's to read.
|
|
"""
|
|
return [
|
|
*parse.crontab(self._out("cat /etc/crontab"), system=True),
|
|
*parse.crontab(self._out("crontab -l"), user=self.username),
|
|
]
|