feat: packages, services and updates
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 25s
CI / test (3.10) (push) Successful in 26s
CI / test (3.11) (push) Successful in 24s
CI / test (3.12) (push) Successful in 26s
CI / test (3.10) (pull_request) Successful in 25s
CI / test (3.11) (pull_request) Successful in 24s
CI / test (3.12) (pull_request) Successful in 25s
The write side NetOrk/netork#799 needs, on both drivers. - Packages: get_packages (FreeBSD `pkg query` with the repository as source, nothing on a classic system without pkg; OpenBSD `pkg_info`), install_package / uninstall_package as root (`pkg install`/`pkg delete`, `pkg_add -I`/`pkg_delete`), names checked before anything is sent, a failure raised with what the tool printed. - Updates: get_available_updates. FreeBSD `pkg upgrade -n`, with `security` from VuXML (`pkg audit`: True for a listed package, False for any other, None when the audit could not run); OpenBSD `pkg_add -u -n -v` (no security verdict). Base-system patches are one `base-system` entry, as decided in #799: OpenBSD's `syspatch -c`, and on a classic FreeBSD what `freebsd-update` has fetched; on pkgbase the base system is packages from FreeBSD-base. - Services: get_services lists the enabled ones (FreeBSD `service ... status` as root, as root-only pidfiles hide daemons otherwise, without root when sudo refuses; OpenBSD `rcctl check`), manage_service runs start/stop/restart/enable/disable as root and returns success and output. Fixtures recorded on the FreeBSD 15.1 and OpenBSD 7.9 VMs with genuinely outdated packages (FreeBSD pointed at the latest branch, an OpenBSD package taken back to its release build). Checked live on both, including a service restart and installing and removing a package.
This commit is contained in:
@@ -347,3 +347,146 @@ def netstat_listening(text: str) -> list[dict[str, Any]]:
|
||||
continue
|
||||
sockets.append(_socket(proto, local, family, None, None))
|
||||
return _unique(sockets)
|
||||
|
||||
|
||||
# -- packages, updates, services ---------------------------------------------------
|
||||
|
||||
_PACKAGE_NAME = re.compile(r"^(?P<name>\S+?)-(?P<version>\d\S*)$")
|
||||
|
||||
|
||||
def split_package(full: str) -> tuple[str, str]:
|
||||
"""An OpenBSD package name: ``bash-completion-2.17.0`` -> (``bash-completion``, ``2.17.0``).
|
||||
|
||||
The version starts at the first ``-`` followed by a digit.
|
||||
"""
|
||||
match = _PACKAGE_NAME.match(full)
|
||||
return (match["name"], match["version"]) if match else (full, "")
|
||||
|
||||
|
||||
def pkg_query(text: str) -> list[dict[str, Any]]:
|
||||
"""FreeBSD ``pkg query '%n\\t%v\\t%R\\t%sb\\t%c'``; the source is the repository."""
|
||||
packages = []
|
||||
for line in text.splitlines():
|
||||
parts = line.split("\t")
|
||||
if len(parts) < 5:
|
||||
continue
|
||||
name, version, repo, size, comment = parts[:5]
|
||||
packages.append(
|
||||
{
|
||||
"name": name,
|
||||
"version": version,
|
||||
"installed": True,
|
||||
"description": comment,
|
||||
"size": int(size) if size.isdigit() else 0,
|
||||
"source": repo,
|
||||
}
|
||||
)
|
||||
return packages
|
||||
|
||||
|
||||
def pkg_info(text: str) -> list[dict[str, Any]]:
|
||||
"""OpenBSD ``pkg_info``: ``name-version comment`` per installed package."""
|
||||
packages = []
|
||||
for line in text.splitlines():
|
||||
full, _, comment = line.partition(" ")
|
||||
if not full:
|
||||
continue
|
||||
name, version = split_package(full)
|
||||
packages.append(
|
||||
{
|
||||
"name": name,
|
||||
"version": version,
|
||||
"installed": True,
|
||||
"description": comment.strip(),
|
||||
"size": 0,
|
||||
"source": "pkg_add",
|
||||
}
|
||||
)
|
||||
return packages
|
||||
|
||||
|
||||
_PKG_UPGRADE = re.compile(
|
||||
r"^\s+(?P<name>\S+): (?P<old>\S+) -> (?P<new>\S+)(?: \[(?P<repo>[^\]]+)\])?"
|
||||
)
|
||||
|
||||
|
||||
def pkg_upgrades(text: str) -> list[dict[str, Any]]:
|
||||
"""FreeBSD ``pkg upgrade -n``: the packages it would upgrade, with their repository."""
|
||||
updates = []
|
||||
for line in text.splitlines():
|
||||
match = _PKG_UPGRADE.match(line)
|
||||
if match:
|
||||
updates.append(
|
||||
{
|
||||
"name": match["name"],
|
||||
"current_version": match["old"],
|
||||
"new_version": match["new"],
|
||||
"origin": match["repo"],
|
||||
}
|
||||
)
|
||||
return updates
|
||||
|
||||
|
||||
def pkg_audit(text: str) -> set[str]:
|
||||
"""FreeBSD ``pkg audit -q``: the names of the vulnerable packages (VuXML)."""
|
||||
names = set()
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if line:
|
||||
names.add(line.rpartition("-")[0] or line)
|
||||
return names
|
||||
|
||||
|
||||
_CANDIDATE = re.compile(r"^Update candidates: (?P<old>\S+) -> (?P<new>\S+)$")
|
||||
|
||||
|
||||
def pkg_add_candidates(text: str) -> list[dict[str, Any]]:
|
||||
"""OpenBSD ``pkg_add -u -n -v``: the candidates whose version changes."""
|
||||
updates: list[dict[str, Any]] = []
|
||||
seen = set()
|
||||
for line in text.splitlines():
|
||||
match = _CANDIDATE.match(line.strip())
|
||||
if not match or match["old"] == match["new"] or match["old"] in seen:
|
||||
continue
|
||||
seen.add(match["old"])
|
||||
name, old = split_package(match["old"])
|
||||
_, new = split_package(match["new"])
|
||||
updates.append({"name": name, "current_version": old, "new_version": new})
|
||||
return updates
|
||||
|
||||
|
||||
def syspatch(text: str) -> list[str]:
|
||||
"""OpenBSD ``syspatch -c`` / ``-l``: one patch name per line."""
|
||||
return [line.strip() for line in text.splitlines() if re.match(r"^\d{3}_\S+$", line.strip())]
|
||||
|
||||
|
||||
_RUNNING = re.compile(r"is running as pid (\d+)")
|
||||
|
||||
|
||||
def service_status(text: str) -> list[dict[str, Any]]:
|
||||
"""FreeBSD: ``name<TAB>first line of 'service name status'`` per enabled service."""
|
||||
services = []
|
||||
for line in text.splitlines():
|
||||
name, _, status = line.partition("\t")
|
||||
if not name:
|
||||
continue
|
||||
running = _RUNNING.search(status)
|
||||
services.append(
|
||||
{
|
||||
"name": name,
|
||||
"running": bool(running),
|
||||
"enabled": True,
|
||||
"pid": int(running[1]) if running else 0,
|
||||
}
|
||||
)
|
||||
return services
|
||||
|
||||
|
||||
def rcctl_check(text: str) -> list[dict[str, Any]]:
|
||||
"""OpenBSD: ``name<TAB>1|0`` per enabled service, from ``rcctl check``."""
|
||||
services = []
|
||||
for line in text.splitlines():
|
||||
name, _, ok = line.partition("\t")
|
||||
if name:
|
||||
services.append({"name": name, "running": ok.strip() == "1", "enabled": True, "pid": 0})
|
||||
return services
|
||||
|
||||
Reference in New Issue
Block a user