feat: report a dpkg left halfway and when the host's own automatic updates last succeeded

An unattended upgrade cut off while it built an initrd left a kernel package
half-configured on an Ubuntu host. For three months apt refused every upgrade,
while apt.systemd.daily exited quietly and the timers reported success
(NetOrk/netork#738).

HOST_STATUS_COMMAND now also reads, without root:

- dpkg's state the way apt checks it: numbered files left in
  /var/lib/dpkg/updates, or a package half-installed, unpacked,
  half-configured, waiting for triggers or flagged for reinstallation.
  While dpkg, apt, aptitude or unattended-upgrade runs, packages are
  halfway on purpose, so the answer is then None.
- /var/lib/apt/periodic/upgrade-stamp, which apt.systemd.daily touches only
  after unattended-upgrade succeeded, and the interval of
  APT::Periodic::Unattended-Upgrade in apt's units. update-success-stamp is
  not read: every apt-get update touches it, whoever runs it.

HostStatusDict gains package_manager_interrupted, interrupted_packages,
auto_updates_last_success and auto_updates_interval, all NotRequired.

napalm_device_types/host_status.py

@@ -24,7 +24,22 @@ without a ``/lib/modules`` of its own.
 
 **Auto updates** is apt's ``APT::Periodic::Unattended-Upgrade`` (set, not "0",
 and ``apt-daily-upgrade.timer`` not disabled) or an enabled dnf-automatic timer.
-It is ``None`` on a host with neither apt nor dnf-automatic.
+It is ``None`` on a host with neither apt nor dnf-automatic. When they last ran
+successfully is ``/var/lib/apt/periodic/upgrade-stamp``, which
+``apt.systemd.daily`` touches only after unattended-upgrade succeeded.
+``update-success-stamp`` would say nothing: every ``apt-get update`` touches it,
+whoever runs it. How often they are meant to run is the same setting, in apt's
+units (a bare number is days).
+
+**An interrupted package manager** is dpkg left halfway, the way apt checks it:
+numbered files left in ``/var/lib/dpkg/updates``, or a package in a state only
+a run cut off leaves behind (half-installed, unpacked, half-configured, waiting
+for triggers, or flagged for reinstallation). apt then refuses to install
+anything until ``dpkg --configure -a`` has run, while ``apt.systemd.daily``
+exits quietly every day (NetOrk/netork#738). While dpkg, apt, aptitude or
+unattended-upgrade is running, packages are halfway on purpose -- also between
+two of apt's dpkg calls -- so the answer is then ``None``, as on a host without
+dpkg.
 """
 
 from __future__ import annotations
@@ -32,7 +47,7 @@ from __future__ import annotations
 import re
 from typing import Dict, List, Optional, Tuple, TYPE_CHECKING
 
-from napalm_device_types.models import HostStatusDict
+from napalm_device_types.models import HostStatusDict, InterruptedPackageDict
 from napalm_device_types.terminal import strip_terminal_codes
 
 _BEGIN = "HSTAT_BEGIN"
@@ -40,6 +55,11 @@ _END = "HSTAT_END"
 _REBOOT_FILE = "/var/run/reboot-required"
 _APT_TIMER = "apt-daily-upgrade.timer"
 _DNF_TIMERS = ("dnf-automatic.timer", "dnf-automatic-install.timer")
+_UPGRADE_STAMP = "/var/lib/apt/periodic/upgrade-stamp"
+_DPKG_JOURNAL = "/var/lib/dpkg/updates"
+#: The processes that leave packages halfway while they work (``comm``, at most
+#: 15 characters: unattended-upgrade shows as ``unattended-upgr``).
+_PACKAGE_MANAGERS = "dpkg|apt|apt-get|aptitude|unattended-upgr"
 
 #: One line, POSIX ``sh``, read-only, no privileges. The frame markers are
 #: printed in two halves so that an echoing transport does not show them early.
@@ -56,7 +76,13 @@ HOST_STATUS_COMMAND = (
     "echo '[freebsd-kernel]'; freebsd-version -k; "
     "echo '[freebsd-running]'; freebsd-version -r; fi; "
     "if command -v apt-config >/dev/null 2>&1; then echo '[apt-config]'; "
-    "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; fi; "
+    "apt-config dump 2>/dev/null | grep '^APT::Periodic::Unattended-Upgrade '; "
+    f"echo '[upgrade-stamp]'; stat -c %Y {_UPGRADE_STAMP} 2>/dev/null; fi; "
+    "if command -v dpkg-query >/dev/null 2>&1; then "
+    f"ps -e -o comm= 2>/dev/null | grep -qxE '{_PACKAGE_MANAGERS}' && echo '[dpkg-busy]'; "
+    f"echo '[dpkg-journal]'; ls -1 {_DPKG_JOURNAL} 2>/dev/null | head -n 20; "
+    "echo '[dpkg-audit]'; dpkg-query -W -f='${db:Status-Abbrev} ${Package}\\n' 2>/dev/null "
+    "| awk 'substr($0, 2, 1) ~ /[HUFWt]/ || substr($0, 3, 1) == \"R\"' | head -n 50; fi; "
     f"echo '[timers]'; for u in {_APT_TIMER} {' '.join(_DNF_TIMERS)}; do "
     'printf \'%s %s\\n\' "$u" "$(systemctl is-enabled "$u" 2>/dev/null)"; done; '
     "printf '%s%s\\n' HSTAT_ END; } 2>/dev/null | cat"
@@ -64,6 +90,17 @@ HOST_STATUS_COMMAND = (
 
 _PERIODIC = re.compile(r'^APT::Periodic::Unattended-Upgrade\s+"([^"]*)"')
 _OFF_STATES = frozenset({"disabled", "masked"})
+_INTERVAL = re.compile(r"(\d+)([smhd]?)")
+_UNIT_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "": 86400}
+_JOURNAL_ENTRY = re.compile(r"\d+")
+#: dpkg's second status letter -> its word, for the states a cut-off run leaves.
+_HALFWAY = {
+    "H": "half-installed",
+    "U": "unpacked",
+    "F": "half-configured",
+    "W": "triggers-awaited",
+    "t": "triggers-pending",
+}
 
 
 def _sections(output: str) -> Dict[str, List[str]]:
@@ -143,17 +180,54 @@ def _timer_states(sections: Dict[str, List[str]]) -> Dict[str, str]:
     return states
 
 
+def _periodic(sections: Dict[str, List[str]]) -> Optional[str]:
+    """The value of ``APT::Periodic::Unattended-Upgrade``, or None when it is not set."""
+    lines = sections.get("apt-config") or []
+    match = next(filter(None, (_PERIODIC.match(line) for line in lines)), None)
+    return match.group(1) if match else None
+
+
 def _auto_updates(sections: Dict[str, List[str]]) -> Optional[bool]:
     timers = _timer_states(sections)
     if any(timers.get(t) == "enabled" for t in _DNF_TIMERS):
         return True
     if "apt-config" not in sections:
         return None
-    match = next(filter(None, (_PERIODIC.match(line) for line in sections["apt-config"])), None)
-    switched_on = match is not None and match.group(1) not in ("", "0")
+    switched_on = _periodic(sections) not in (None, "", "0")
     return switched_on and timers.get(_APT_TIMER) not in _OFF_STATES
 
 
+def _interval(sections: Dict[str, List[str]]) -> Optional[int]:
+    """apt's interval in seconds, the way ``apt.systemd.daily`` reads it; 0 for "always"."""
+    value = _periodic(sections)
+    if value == "always":
+        return 0
+    match = _INTERVAL.fullmatch(value or "")
+    if not match or int(match.group(1)) == 0:
+        return None
+    return int(match.group(1)) * _UNIT_SECONDS[match.group(2)]
+
+
+def _last_success(sections: Dict[str, List[str]]) -> Optional[int]:
+    stamp = (sections.get("upgrade-stamp") or [""])[0]
+    return int(stamp) if stamp.isdigit() else None
+
+
+def _halfway(line: str) -> Optional[InterruptedPackageDict]:
+    status, _, name = line.partition(" ")
+    name = name.strip()
+    state = "reinstall-required" if status[2:3] == "R" else _HALFWAY.get(status[1:2])
+    return {"name": name, "state": state} if name and state else None
+
+
+def _dpkg(sections: Dict[str, List[str]]) -> Tuple[Optional[bool], List[InterruptedPackageDict]]:
+    if "dpkg-audit" not in sections or "dpkg-busy" in sections:
+        return None, []
+    packages = [p for p in map(_halfway, sections["dpkg-audit"]) if p is not None]
+    journal = any(_JOURNAL_ENTRY.fullmatch(f) for f in sections.get("dpkg-journal") or [])
+    return bool(packages) or journal, packages
+
+
 def parse_host_status(output: str) -> HostStatusDict:
     """Parse what :data:`HOST_STATUS_COMMAND` printed.
 
@@ -161,10 +235,15 @@ def parse_host_status(output: str) -> HostStatusDict:
     """
     sections = _sections(output)
     required, reason = _reboot(sections)
+    interrupted, packages = _dpkg(sections)
     return {
         "reboot_required": required,
         "reboot_reason": reason,
         "auto_updates": _auto_updates(sections),
+        "auto_updates_last_success": _last_success(sections),
+        "auto_updates_interval": _interval(sections),
+        "package_manager_interrupted": interrupted,
+        "interrupted_packages": packages,
     }
 
 
@@ -185,11 +264,16 @@ class HostStatusMixin:
 
     def get_host_status(self) -> HostStatusDict:
         """
-        Returns whether the host needs a reboot and whether it patches itself.
+        Returns whether the host needs a reboot, whether it patches itself, and
+        whether its package manager was left halfway.
 
         * reboot_required (bool or None)
         * reboot_reason (string or None)
         * auto_updates (bool or None)
+        * auto_updates_last_success (Unix time or None)
+        * auto_updates_interval (seconds or None)
+        * package_manager_interrupted (bool or None)
+        * interrupted_packages (list of {name, state})
 
         :raises ValueError: if the host's output carried no intact report.
         """

napalm_device_types/models.py

@@ -75,6 +75,15 @@ class UpdateDict(TypedDict):
     security: NotRequired[Optional[bool]]
 
 
+class InterruptedPackageDict(TypedDict):
+    """A package an interrupted package manager run left halfway."""
+
+    name: str
+    #: dpkg's word for it: "half-installed", "unpacked", "half-configured",
+    #: "triggers-awaited", "triggers-pending", or "reinstall-required".
+    state: str
+
+
 class HostStatusDict(TypedDict):
     """What a host says about its own patch state (``HostStatusMixin.get_host_status``)."""
 
@@ -84,6 +93,16 @@ class HostStatusDict(TypedDict):
     reboot_reason: Optional[str]
     #: True when the host installs updates on its own (unattended-upgrades, dnf-automatic).
     auto_updates: Optional[bool]
+    #: True when a package manager run was cut off and has to be finished
+    #: (``dpkg --configure -a``) before anything else installs; None when it
+    #: cannot tell, or a package manager is running right now.
+    package_manager_interrupted: NotRequired[Optional[bool]]
+    #: The packages that run left halfway; empty when only its journal says so.
+    interrupted_packages: NotRequired[List[InterruptedPackageDict]]
+    #: When the host's own automatic updates last ran successfully (Unix time).
+    auto_updates_last_success: NotRequired[Optional[int]]
+    #: How often they are meant to run, in seconds; 0 for every time the timer fires.
+    auto_updates_interval: NotRequired[Optional[int]]
 
 
 # ---------------------------------------------------------------------------