Build/playbook re-run safety: corp access, netplan rollback, first-boot retry
H1 corp_admin_cidrs (UFW 22/9009 on the corp NIC from these only; refuses
when it would lock out the current SSH client) and corp_ssh_password_auth
(sshd drop-in, passwords only from the imaging subnet). Defaults keep the
current behaviour and print a loud warning.
H2 pxe-netplan-apply.sh: pre-flight refusal (pxe-MAC NIC holds the default
route, a DHCP lease or the route to an SSH client), run from the USB
before anything changes; post-apply checks (names, imaging IPv4, corp
IPv4 back within 30 s, SSH route) and a 120 s systemd-run auto-revert.
H3 restore() puts the previous NIC names back (two-phase rename via
temporary names, so swaps work) before generate + apply.
H4 a failed rename is fatal (restore + exit 1); the play drops the old
any-interface UFW rules only after verifying both NICs and their
per-interface SSH rules.
H5 first-boot checks the playbook exit code, retries at next boot (max 5),
writes /var/lib/pxe-firstboot.failed and /etc/issue + /etc/motd banners.
M1 build-usb.sh verifies iPXE files against the playbook pins, every wheel
against its own package's hashes, drops the unpinned pip fallback,
requires --site-vars or --no-site-vars.
M2 webapp restarts when the code, wheels or unit changed.
M3 ignore-carrier on the imaging NIC; ip_nonlocal_bind kept for the
apache/networkd boot race.
M4 menu suffix must match blancco_kver; missing kernel with no earlier
build is fatal; a live menu with the staged Blancco test entry is not
overwritten.
M5 build-proxmox-iso.sh is unsupported (hard exit); use build-usb --image.
M6 deploy-unattend.sh: password via SSHPASS env and sudo stdin only,
%q-quoted remote args, sha256 check of the live file; lint-unattend.py
reads PXE_SERVER/PXE_USER/PXE_PASS and uses sshpass -e.
M7 the playbook lints the unattend files before the force: yes copies
(lint-unattend.py ships on CIDATA as scripts/); pre-commit ACMR.
M8 Blancco initramfs build: no ignore_errors, pipefail, trap cleanup.
Also: Send-ImagingStage.ps1 deploy, carrier hook install tasks
(pxe-nic-carrier, networkd-dispatcher), L1 imaging-active hold
(allow_disruptive) and no iPXE downgrade below an Updates-page install,
L2 iso-volid, L3 kver pin read from the playbook, L4 cp -rL, L5 refuse
>= 4 GiB WinPE files, L6 PIP_DIR wipe guard, L7 per-package wheel hashes,
no online fallback, L8 check-updates.py (apt_pkg/cached compare, all
components, exit 2 on no debs / bad answers).
This commit is contained in:
@@ -2,18 +2,21 @@
|
||||
"""check-updates.py - is the offline bundle current? Run before building a USB.
|
||||
|
||||
* offline-packages/*.deb against Ubuntu 24.04 (noble, noble-updates,
|
||||
noble-security; main + universe): which have a newer version, and which
|
||||
noble-security; main, restricted, universe, multiverse): which have a
|
||||
newer version, and which
|
||||
of those are security updates.
|
||||
* webapp/requirements.txt pins against PyPI (newer release?) and osv.dev
|
||||
(known vulnerabilities in the pinned version).
|
||||
|
||||
Exit 0 = nothing outstanding; 1 = security updates or vulnerable pins found
|
||||
(the details are printed); 2 = could not check (no network, bad input).
|
||||
(the details are printed); 2 = could not check (no network, bad or missing
|
||||
input such as a folder with no .debs, an unreadable index or API answer).
|
||||
Needs internet (Ubuntu archive, pypi.org, api.osv.dev); read-only.
|
||||
|
||||
Usage: scripts/check-updates.py [--debs DIR] [--reqs FILE] [--all]
|
||||
"""
|
||||
import argparse
|
||||
import functools
|
||||
import glob
|
||||
import gzip
|
||||
import json
|
||||
@@ -26,7 +29,11 @@ import urllib.request
|
||||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
ARCHIVE = "http://archive.ubuntu.com/ubuntu/dists"
|
||||
SUITES = ("noble", "noble-updates", "noble-security")
|
||||
COMPONENTS = ("main", "universe")
|
||||
COMPONENTS = ("main", "restricted", "universe", "multiverse")
|
||||
|
||||
|
||||
class CheckError(Exception):
|
||||
"""Could not check (exit 2)."""
|
||||
|
||||
|
||||
def fetch(url, timeout=60):
|
||||
@@ -34,21 +41,39 @@ def fetch(url, timeout=60):
|
||||
return r.read()
|
||||
|
||||
|
||||
try:
|
||||
import apt_pkg # python3-apt: in-process comparison, no fork per pair
|
||||
apt_pkg.init_system()
|
||||
except Exception: # not installed: fall back to dpkg, cached
|
||||
apt_pkg = None
|
||||
|
||||
|
||||
@functools.lru_cache(maxsize=None)
|
||||
def newer(a, b):
|
||||
"""True if Debian version a > b."""
|
||||
if a == b:
|
||||
return False
|
||||
if apt_pkg is not None:
|
||||
return apt_pkg.version_compare(a, b) > 0
|
||||
return subprocess.run(["dpkg", "--compare-versions", a, "gt", b]).returncode == 0
|
||||
|
||||
|
||||
def ubuntu_index():
|
||||
"""({package: newest version}, {package: newest security version})."""
|
||||
def ubuntu_index(wanted):
|
||||
"""({package: newest version}, {package: newest security version}),
|
||||
for the package names in wanted only."""
|
||||
best, sec = {}, {}
|
||||
for suite in SUITES:
|
||||
for comp in COMPONENTS:
|
||||
text = gzip.decompress(fetch(f"{ARCHIVE}/{suite}/{comp}/binary-amd64/Packages.gz")).decode("utf-8", "replace")
|
||||
try:
|
||||
text = gzip.decompress(fetch(f"{ARCHIVE}/{suite}/{comp}/binary-amd64/Packages.gz")).decode("utf-8", "replace")
|
||||
except (OSError, EOFError) as ex:
|
||||
raise CheckError(f"{suite}/{comp} index: {ex}")
|
||||
pkg = None
|
||||
for line in text.splitlines():
|
||||
if line.startswith("Package: "):
|
||||
pkg = line[9:].strip()
|
||||
if pkg not in wanted:
|
||||
pkg = None
|
||||
elif line.startswith("Version: ") and pkg:
|
||||
v = line[9:].strip()
|
||||
if pkg not in best or newer(v, best[pkg]):
|
||||
@@ -61,9 +86,9 @@ def ubuntu_index():
|
||||
def check_debs(deb_dir, show_all):
|
||||
debs = glob.glob(os.path.join(deb_dir, "*.deb"))
|
||||
if not debs:
|
||||
print(f"no .deb files in {deb_dir}: nothing to check (a failed download?)")
|
||||
return 1
|
||||
best, sec = ubuntu_index()
|
||||
raise CheckError(f"no .deb files in {deb_dir}: nothing to check (a failed download?)")
|
||||
names = {os.path.basename(d).split("_")[0] for d in debs}
|
||||
best, sec = ubuntu_index(names)
|
||||
stale, security = [], []
|
||||
for d in debs:
|
||||
parts = os.path.basename(d)[:-4].split("_")
|
||||
@@ -82,14 +107,23 @@ def check_debs(deb_dir, show_all):
|
||||
|
||||
|
||||
def check_pins(reqs):
|
||||
pins = re.findall(r"^([A-Za-z0-9_.-]+)==([^\s\\]+)", open(reqs).read(), re.M)
|
||||
try:
|
||||
text = open(reqs).read()
|
||||
except OSError as ex:
|
||||
raise CheckError(f"{reqs}: {ex}")
|
||||
pins = re.findall(r"^([A-Za-z0-9_.-]+)==([^\s\\]+)", text, re.M)
|
||||
if not pins:
|
||||
print(f"no pinned packages in {reqs}")
|
||||
return 0
|
||||
q = {"queries": [{"package": {"name": n, "ecosystem": "PyPI"}, "version": v} for n, v in pins]}
|
||||
req = urllib.request.Request("https://api.osv.dev/v1/querybatch", data=json.dumps(q).encode(),
|
||||
headers={"Content-Type": "application/json"})
|
||||
results = json.load(urllib.request.urlopen(req, timeout=60))["results"]
|
||||
try:
|
||||
results = json.load(urllib.request.urlopen(req, timeout=60))["results"]
|
||||
if not isinstance(results, list) or len(results) != len(pins):
|
||||
raise ValueError("osv.dev returned %r results for %d pins" % (type(results).__name__, len(pins)))
|
||||
except (ValueError, KeyError, TypeError) as ex:
|
||||
raise CheckError(f"osv.dev answer: {ex}")
|
||||
bad = 0
|
||||
print(f"Python pins in {os.path.relpath(reqs, ROOT)}:")
|
||||
for (n, v), r in zip(pins, results):
|
||||
@@ -97,7 +131,7 @@ def check_pins(reqs):
|
||||
latest = json.loads(fetch(f"https://pypi.org/pypi/{n}/json", 30))["info"]["version"]
|
||||
except Exception: # report, keep going
|
||||
latest = "?"
|
||||
vulns = [x["id"] for x in r.get("vulns", [])]
|
||||
vulns = [x.get("id", "?") for x in (r or {}).get("vulns", [])]
|
||||
bad += bool(vulns)
|
||||
flag = "VULNERABLE" if vulns else ("newer " if latest not in (v, "?") else "ok ")
|
||||
print(f" {flag} {n:14} {v:10} latest {latest:10} {', '.join(vulns)}")
|
||||
@@ -112,7 +146,7 @@ def main():
|
||||
a = ap.parse_args()
|
||||
try:
|
||||
problems = check_debs(a.debs, a.all) + check_pins(a.reqs)
|
||||
except OSError as ex:
|
||||
except (OSError, CheckError) as ex:
|
||||
print(f"could not check: {ex}", file=sys.stderr)
|
||||
return 2
|
||||
print("PASSED: nothing outstanding" if not problems else f"ATTENTION: {problems} item(s) need updating")
|
||||
|
||||
Reference in New Issue
Block a user