diff --git a/AppImage/scripts/build_appimage.sh b/AppImage/scripts/build_appimage.sh index c92424c5..b9034e2b 100755 --- a/AppImage/scripts/build_appimage.sh +++ b/AppImage/scripts/build_appimage.sh @@ -144,6 +144,8 @@ cp "$SCRIPT_DIR/smartctl_resolver.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo " cp "$SCRIPT_DIR/disk_identity.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ disk_identity.py not found" cp "$SCRIPT_DIR/health_thresholds.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ health_thresholds.py not found" cp "$SCRIPT_DIR/managed_installs.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ managed_installs.py not found" +cp "$SCRIPT_DIR/lxc_update_observation.py" "$APP_DIR/usr/bin/" +cp "$SCRIPT_DIR/lxc_update_finalization.py" "$APP_DIR/usr/bin/" cp "$SCRIPT_DIR/lxc_apps.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ lxc_apps.py not found" cp "$SCRIPT_DIR/custom_links.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ custom_links.py not found" cp "$SCRIPT_DIR/recreate_docker_container.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ recreate_docker_container.py not found" diff --git a/AppImage/scripts/flask_server.py b/AppImage/scripts/flask_server.py index 6c0d1e86..6f19195c 100644 --- a/AppImage/scripts/flask_server.py +++ b/AppImage/scripts/flask_server.py @@ -14253,8 +14253,7 @@ def api_vm_apps_dismiss(vmid): _LXC_APPLY_UPDATES_SCRIPT = "/usr/local/share/proxmenux/scripts/lxc/apply_updates.sh" _LXC_UPDATE_RUN_ID_RE = re.compile(r'^[A-Za-z0-9._:-]{1,128}$') _LXC_UPDATE_TARGET_RE = re.compile(r'^[A-Za-z0-9._:-]{1,180}$') -_lxc_update_finalization_lock = threading.RLock() -_lxc_update_finalizations: dict[tuple[int, str], dict] = {} +from lxc_update_finalization import lock as _lxc_update_finalization_lock, runs as _lxc_update_finalizations _LXC_UPDATE_FINALIZATION_TTL = 6 * 3600 @@ -14547,6 +14546,37 @@ def _finalize_lxc_update( verification_pending = _fast_guest_status(vmid, 'lxc') != 'running' docker_inventory = None + if verification_pending and status in {'success', 'failure', 'partial'}: + try: + import managed_installs + import lxc_update_observation + observation = lxc_update_observation.read(vmid, safe_run_id) + verified = managed_installs.store_lxc_update_observation(vmid, observation) + print(f'[ProxMenux] stopped LXC verification: vmid={vmid} ' + f'run_id={safe_run_id} handoff={observation is not None} ' + f'verified={verified}', flush=True) + if not verified: + verification_errors.append('post-update OS verification unavailable') + apps_persisted = True + if observation is not None and 'apps' in observation: + import lxc_apps + try: + app_result = lxc_apps.store_installed_observations(vmid, observation['apps']) + verification_errors.extend(app_result['errors']) + if status == 'success': + lxc_apps.mark_manual_versions_unverified( + vmid, [item[4:] for item in requested if item.startswith('app:')]) + _publish_guest_modal_cache_revision(vmid) + except Exception: + apps_persisted = False + verification_errors.append('post-update application persistence failed') + # Retain a rejected handoff for diagnosis rather than silently + # deleting the only evidence. Capture prunes old private files. + if observation is not None and verified and apps_persisted: + lxc_update_observation.discard(vmid, safe_run_id) + except Exception as exc: + verification_errors.append(f'post-update OS verification failed: {exc}') + if not verification_pending: try: import managed_installs diff --git a/AppImage/scripts/lxc_apps.py b/AppImage/scripts/lxc_apps.py index f67a3460..10eb5736 100644 --- a/AppImage/scripts/lxc_apps.py +++ b/AppImage/scripts/lxc_apps.py @@ -4503,6 +4503,112 @@ def emit_all_pending_updates() -> int: return count +def _installed_observation_fingerprint(app: dict) -> str: + # Only version-probe configuration, never updater commands or credentials. + fields = ('installed_via', 'installed_regex', 'tag_regex') + _DETECTOR_FIELDS + return hashlib.sha256(json.dumps({k: app.get(k) for k in fields}, + sort_keys=True).encode()).hexdigest() + + +def capture_installed_observations(vmid) -> list[dict]: + """Probe registered automatic versions while a temporary CT still runs. + + No shared writes, upstream requests, discovery or update commands. The + backend later merges this private handoff against the current sidecar. + Manual values and OCI/image registrations cannot be proven this way. + """ + with _cache_lock: + registrations = copy.deepcopy((_read_sidecar(vmid) or {}).get('apps') or []) + observations = [] + for app in registrations: + if app.get('installed_via') not in set(_VALID_METHODS) - {'manual', 'oci_image'}: + continue + try: + version, error = detect_installed_version(vmid, app) + except Exception: + version, error = None, True + verified = (isinstance(version, str) and 0 < len(version) <= 256 + and not re.search(r'[\x00-\x1f\x7f]', version) and not error) + observations.append({'id': app.get('id'), + 'fingerprint': _installed_observation_fingerprint(app), + 'checked_at': _now_iso(), 'verified': bool(verified), + 'installed_version': version if verified else None}) + return observations + + +def store_installed_observations(vmid, observations) -> dict: + """Merge trusted captured versions without executing anything in the CT.""" + errors = [] + if not isinstance(observations, list): + return {'applied': 0, 'errors': ['post-update application observation unavailable']} + seen = set() + parsed = [] + for item in observations: + try: + app_id = item['id'] + fingerprint = item['fingerprint'] + version = item['installed_version'] + checked = datetime.datetime.fromisoformat(item['checked_at'].replace('Z', '+00:00')) + age = time.time() - checked.timestamp() + valid = (isinstance(app_id, str) and 0 < len(app_id) <= 128 + and app_id not in seen and re.fullmatch(r'[a-zA-Z0-9._:-]+', app_id) + and isinstance(fingerprint, str) and re.fullmatch(r'[a-f0-9]{64}', fingerprint) + and checked.tzinfo is not None and 0 <= age <= 3600 + and type(item['verified']) is bool + and ((item['verified'] and isinstance(version, str) and 0 < len(version) <= 256 + and not re.search(r'[\x00-\x1f\x7f]', version)) + or (not item['verified'] and version is None))) + if not valid: + raise ValueError('invalid observation') + seen.add(app_id) + parsed.append((item, checked)) + except (KeyError, TypeError, ValueError, AttributeError, OverflowError): + # Validate the whole payload before its first shared write. + return {'applied': 0, 'errors': ['invalid post-update application observation']} + if not parsed: + return {'applied': 0, 'errors': []} + with _cache_lock: + sidecar = _read_sidecar(vmid) + applied = 0 + for item, checked in parsed: + app = _find_app(sidecar or {}, item['id']) + if not app or _installed_observation_fingerprint(app) != item['fingerprint']: + errors.append('application configuration changed after capture') + continue + state = dict(app.get('state') or {}) + try: + current = datetime.datetime.fromisoformat((state.get('checked_at') or '').replace('Z', '+00:00')) + if current >= checked: + errors.append('newer application observation preserved') + continue + except (TypeError, ValueError): + pass + if item['verified']: + state['installed_version'] = item['installed_version'] + state['update_available'] = (compare(item['installed_version'], state['latest_version']) + if state.get('latest_version') else None) + # An installed-version probe does not recheck the upstream; + # preserve an existing upstream error rather than hiding it. + if state.get('error') == 'post-update installed version could not be verified': + state['error'] = None + state.pop('last_known_installed_version', None) + else: + state['last_known_installed_version'] = (state.get('installed_version') + or state.get('last_known_installed_version')) + state['installed_version'] = None + state['update_available'] = None + state['error'] = 'post-update installed version could not be verified' + errors.append(state['error']) + state['checked_at'] = item['checked_at'] + app['state'] = state + applied += 1 + if applied: + sidecar['updated_at'] = _now_iso() + if not _write_sidecar(vmid, sidecar): + raise OSError('post-update application persistence failed') + return {'applied': applied, 'errors': errors} + + def check_all( vmid, force: bool = False, notify: bool = True, ) -> Optional[dict]: diff --git a/AppImage/scripts/lxc_update_finalization.py b/AppImage/scripts/lxc_update_finalization.py new file mode 100644 index 00000000..57cadc8e --- /dev/null +++ b/AppImage/scripts/lxc_update_finalization.py @@ -0,0 +1,11 @@ +"""One process-wide LXC completion state, independent of Flask import name. + +AppRun executes flask_server as __main__, while collectors import it as +flask_server. Both completion paths must share this canonical module so +the terminal hook and compatibility HTTP callback cannot consume a run twice. +No guest actions, registry writes or background tasks occur on import. +""" +import threading + +lock = threading.RLock() +runs: dict[tuple[int, str], dict] = {} diff --git a/AppImage/scripts/lxc_update_observation.py b/AppImage/scripts/lxc_update_observation.py new file mode 100644 index 00000000..ded9ffd3 --- /dev/null +++ b/AppImage/scripts/lxc_update_observation.py @@ -0,0 +1,180 @@ +"""Trusted, run-scoped package observation captured before a temporary CT stops. + +The runner only writes this private handoff, not the shared registry. The +Monitor consumes it under its existing registry lock after the process exits. +Never accepted as an HTTP payload and never refreshes package indexes. +""" +from __future__ import annotations + +import argparse +import datetime +import json +import os +from pathlib import Path +import re +import stat +import subprocess +import tempfile + +import managed_installs + +OBSERVATION_DIR = Path("/run/proxmenux/lxc-update-observations") +MAX_AGE = 3600 +MAX_BYTES = 2 * 1024 * 1024 + + +def _path(vmid: int, run_id: str) -> Path: + if int(vmid) <= 0 or not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", run_id): + raise ValueError("invalid LXC update identity") + return OBSERVATION_DIR / f"{int(vmid)}-{run_id}.json" + + +def _private_directory(*, create=True) -> None: + if create: + OBSERVATION_DIR.mkdir(mode=0o700, parents=True, exist_ok=True) + info = OBSERVATION_DIR.lstat() + if (not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() + or info.st_mode & 0o077): + raise OSError("unsafe LXC observation directory") + + +def capture(vmid: int, run_id: str, *, refresh_indexes: bool = False, + include_apps: bool = False) -> dict: + path = _path(vmid, run_id) + _private_directory() + # Remove only obsolete private handoffs left by terminated backend runs. + now = datetime.datetime.now(datetime.timezone.utc) + for old in OBSERVATION_DIR.iterdir(): + info = old.lstat() + if (re.fullmatch(r"[1-9][0-9]*-[A-Za-z0-9._:-]{1,128}\.json", old.name) + and stat.S_ISREG(info.st_mode) and info.st_uid == os.geteuid() + and now.timestamp() - info.st_mtime > MAX_AGE): + old.unlink() + entry = {"id": f"lxc:{int(vmid)}", "type": "lxc", "_vmid": str(int(vmid))} + native_detected = False + try: + detected = managed_installs._detect_lxc_containers(only_vmid=int(vmid)) + if not detected: + raise RuntimeError("running LXC detection unavailable or disabled") + entry = detected[0] + native_detected = not (entry.get('_is_oci') or entry.get('_managed_oci_app')) + if refresh_indexes: + # This is a separate, explicitly requested maintenance action, + # never the read-only button or the normal post-update capture. + if entry.get('_is_oci') or entry.get('_managed_oci_app'): + raise RuntimeError('OCI containers require their own updater') + family = managed_installs._probe_lxc_os(str(vmid)) + commands = {'debian': ['apt-get', 'update', '--error-on=any'], + 'ubuntu': ['apt-get', 'update', '--error-on=any'], + 'alpine': ['apk', 'update']} + if family not in commands: + raise RuntimeError('unsupported or unknown current OS') + refreshed = subprocess.run([managed_installs._PCT_BIN, 'exec', str(vmid), + '--', *commands[family]], capture_output=True, text=True, + timeout=managed_installs._LXC_CACHE_REFRESH_TIMEOUT_SEC) + if refreshed.returncode != 0: + raise RuntimeError('package index refresh failed; no installation was attempted') + result = managed_installs._check_lxc_updates(entry, refresh_package_cache=False) + except Exception as exc: + entry["_os_family"] = None + result = {"available": False, "last_check": managed_installs._now_iso(), + "error": f"post-update package check failed: {exc}"} + observation = {"vmid": int(vmid), "run_id": run_id, + "captured_at": now.isoformat(), "entry": entry, "result": result} + # Only the post-install runner requests app probes. Check-only and the + # maintenance preflight keep their original OS-only contract; OCI never + # enters the generic app probe path, even when explicitly requested. + if include_apps and not refresh_indexes and native_detected: + try: + import lxc_apps + observation['apps'] = lxc_apps.capture_installed_observations(vmid) + except Exception: + observation['apps'] = None + fd, temporary = tempfile.mkstemp(prefix=".observation-", dir=OBSERVATION_DIR) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(observation, handle) + os.replace(temporary, path) + print(f"[ProxMenux] LXC observation captured: vmid={int(vmid)} " + f"run_id={run_id} os={entry.get('_os_family')} " + f"count={result.get('_count')} error={bool(result.get('error'))}", flush=True) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + return observation + + +def read(vmid: int, run_id: str) -> dict | None: + path = _path(vmid, run_id) + def reject(reason: str) -> None: + # Only identity and a fixed reason, never the guest's payload or secrets. + print(f"[ProxMenux] LXC observation rejected: vmid={int(vmid)} " + f"run_id={run_id} reason={reason}", flush=True) + return None + try: + _private_directory(create=False) + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, "r", encoding="utf-8") as handle: + info = os.fstat(handle.fileno()) + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() + or info.st_mode & 0o077 or info.st_size > MAX_BYTES): + return reject('unsafe-file') + observation = json.load(handle) + if not isinstance(observation, dict): + return reject('invalid-document') + captured = datetime.datetime.fromisoformat(observation["captured_at"]) + age = (datetime.datetime.now(datetime.timezone.utc) - captured).total_seconds() + entry, result = observation["entry"], observation["result"] + if (observation["vmid"] != int(vmid) or observation["run_id"] != run_id + or not 0 <= age <= MAX_AGE or not isinstance(entry, dict) + or entry.get("id") != f"lxc:{int(vmid)}" or entry.get("type") != "lxc" + or str(entry.get("_vmid")) != str(int(vmid)) or not isinstance(result, dict)): + return reject('identity-age-or-shape') + if not result.get("error"): + # An incomplete result is unknown, not an implicitly successful zero. + is_oci = bool(entry.get('_is_oci') or entry.get('_managed_oci_app')) + if ((not is_oci and entry.get("_os_family") not in managed_installs._SUPPORTED_OS_FAMILIES) + or not isinstance(result.get("available"), bool) + or not result.get("last_check") + or any(type(result.get(key)) is not int or result[key] < 0 + for key in ("_count", "_security_count")) + or not isinstance(result.get("_packages"), list)): + return reject('incomplete-success') + # OCI skips the generic OS probe, so no family is required. Its + # trusted result must nevertheless be the complete skipped zero. + if is_oci and (result['available'] or result['_count'] != 0 + or result['_security_count'] != 0 or result['_packages']): + return reject('invalid-oci-skip') + print(f"[ProxMenux] LXC observation accepted: vmid={int(vmid)} " + f"run_id={run_id} count={result.get('_count')} " + f"error={bool(result.get('error'))}", flush=True) + return observation + except (OSError, ValueError, TypeError, KeyError, OverflowError) as exc: + return reject(type(exc).__name__) + + +def discard(vmid: int, run_id: str) -> None: + try: + _private_directory(create=False) + _path(vmid, run_id).unlink() + except FileNotFoundError: + pass + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--vmid", required=True, type=int) + parser.add_argument("--run-id", required=True) + parser.add_argument('--include-apps', action='store_true', + help='Capture registered installed versions after an update') + args = parser.parse_args(argv) + try: + capture(args.vmid, args.run_id, include_apps=args.include_apps) + return 0 + except Exception as exc: + print(f"Post-update package observation unavailable: {exc}") + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/AppImage/scripts/managed_installs.py b/AppImage/scripts/managed_installs.py index a54707ec..430c652a 100644 --- a/AppImage/scripts/managed_installs.py +++ b/AppImage/scripts/managed_installs.py @@ -520,10 +520,11 @@ def _lxc_updates_notification_enabled() -> bool: return False -def _list_pve_lxcs() -> list[dict]: +def _list_pve_lxcs(*, strict: bool = False) -> list[dict]: """Return basic info per LXC on this node via ``pct list``. Each item is ``{vmid, status, name}``. Empty list on any failure — never - raises so the detector caller can continue. + raises by default. Strict callers distinguish a failed read from a + verified empty inventory before removing persisted observations. """ try: r = subprocess.run( @@ -531,16 +532,27 @@ def _list_pve_lxcs() -> list[dict]: capture_output=True, text=True, timeout=5, ) except (subprocess.TimeoutExpired, FileNotFoundError, OSError): + if strict: + raise RuntimeError('LXC inventory could not be read') from None return [] if r.returncode != 0: + if strict: + raise RuntimeError('LXC inventory could not be read') return [] + lines = r.stdout.splitlines() + if strict and (not lines or not lines[0].strip().startswith('VMID') or 'Status' not in lines[0]): + raise RuntimeError('LXC inventory is incomplete') out: list[dict] = [] - for line in r.stdout.splitlines()[1:]: # skip header row + for line in lines[1:]: # skip header row + if not line.strip(): + continue # `pct list` columns: VMID Status Lock Name # `Lock` is empty most of the time, so split max 4 ways parts = line.split(None, 3) if len(parts) < 2: + if strict: + raise RuntimeError('LXC inventory is incomplete') continue vmid = parts[0] status = parts[1] @@ -548,6 +560,8 @@ def _list_pve_lxcs() -> list[dict]: # be the name itself if Lock was omitted by the formatter. name = parts[-1] if len(parts) >= 3 else "" if not vmid.isdigit(): + if strict: + raise RuntimeError('LXC inventory is incomplete') continue out.append({"vmid": vmid, "status": status, "name": name}) return out @@ -888,7 +902,7 @@ def _probe_lxc_os(vmid: str, *, raise_on_read_error: bool = False) -> Optional[s def _detect_lxc_containers(only_vmid: Optional[int] = None) -> list[dict]: - """Enumerate running Debian/Ubuntu CTs as registry entries. + """Detect running supported CTs and retain known stopped CTs by presence. OS detection is cached in the registry entry (`_os_family`), so the expensive ``pct exec cat /etc/os-release`` only runs during detection @@ -920,7 +934,7 @@ def _detect_lxc_containers(only_vmid: Optional[int] = None) -> list[dict]: if isinstance(it, dict) and it.get("type") == "lxc" } - cts = _list_pve_lxcs() + cts = _list_pve_lxcs(strict=True) # Set of CTs currently managed by oci_manager (Secure Gateway etc). # Their update path is the OCI app's own updater — we mark them so # the LXC row in the UI redirects the user there instead of running @@ -931,11 +945,17 @@ def _detect_lxc_containers(only_vmid: Optional[int] = None) -> list[dict]: for ct in cts: if only_vmid is not None and str(ct.get("vmid")) != str(int(only_vmid)): continue - if ct["status"] != "running": - continue vmid = ct["vmid"] cid = f"lxc:{vmid}" prior = existing_by_id.get(cid) or {} + if ct["status"] != "running": + # A stopped CT still exists. Only the full detection sweep keeps + # its registry identity; targeted refresh requires a running CT. + # Do not copy cached probe metadata back over a newer observation. + if only_vmid is None and prior: + out.append({'id': cid, 'type': 'lxc', '_vmid': vmid, + 'name': ct.get('name') or prior.get('name') or f'CT-{vmid}'}) + continue # OCI-image marker is cached — the CT origin doesn't change # over its lifetime, and reading the pct config file is cheap @@ -1085,11 +1105,14 @@ def detect_and_register() -> dict: Returns the new registry. """ discovered: dict[str, dict] = {} + failed_types: set[str] = set() for detector in _DETECTORS: try: result = detector() except Exception as e: print(f"[ProxMenux] managed_installs detector {detector.__name__} failed: {e}") + if detector is _detect_lxc_containers: + failed_types.add('lxc') continue for entry in _normalise_detector_result(result): if not entry.get("id"): @@ -1115,7 +1138,7 @@ def detect_and_register() -> dict: for it in items: if not it.get("id") or it.get("removed_at"): continue - if it["id"] not in discovered: + if it["id"] not in discovered and it.get('type') not in failed_types: it["removed_at"] = now reg["items"] = items @@ -1810,9 +1833,20 @@ def check_for_updates(force: bool = False) -> list[dict]: with _lock: reg = _read_registry() items = reg.get("items", []) + lxc_states: dict[str, str] = {} + if any(it.get('type') == 'lxc' and not it.get('removed_at') for it in items): + try: + lxc_states = {str(ct['vmid']): ct['status'] for ct in _list_pve_lxcs(strict=True)} + except RuntimeError: + # Do not overwrite an observation or attempt guest probes + # when presence could not be verified by the host. + print('[ProxMenux] LXC inventory unavailable; stored checks retained') for it in items: if it.get("removed_at"): continue + if (it.get('type') == 'lxc' + and lxc_states.get(str(it.get('_vmid') or it.get('id', '').removeprefix('lxc:'))) != 'running'): + continue checker = _CHECKERS.get(it.get("type")) if not checker: continue @@ -1836,6 +1870,48 @@ def check_for_updates(force: bool = False) -> list[dict]: return updates_available +def store_lxc_update_observation(vmid: int, observation: Optional[dict]) -> bool: + """Store a trusted host-runner observation in the backend's registry lock. + + Missing verification clears the old counts, never fabricates a fresh zero. + This does not start a guest or run a package-manager command. + """ + vmid = int(vmid) + now = _now_iso() + with _lock: + # A trusted targeted observation must never replace unrelated rows + # with an empty registry merely because the old file is unreadable. + try: + with open(_REGISTRY_PATH, encoding='utf-8') as handle: + existing = json.load(handle) + if (not isinstance(existing, dict) or not isinstance(existing.get('items'), list) + or any(not isinstance(item, dict) for item in existing['items'])): + raise OSError('could not verify existing LXC update registry') + except FileNotFoundError: + pass + reg = _read_registry() + items = reg.setdefault("items", []) + target = next((it for it in items if it.get("id") == f"lxc:{vmid}" + and it.get("type") == "lxc" and not it.get("removed_at")), None) + if target is None: + if observation is None: + return False + target = _new_detected_entry(observation["entry"], now) + items.append(target) + if observation is not None: + _merge_detected_entry(target, observation["entry"], now) + result = observation["result"] + else: + target["_os_family"] = None + result = {"available": False, "last_check": now, + "error": "post-update package verification unavailable while LXC is stopped"} + _store_update_result(target, result) + reg["version"] = _SCHEMA_VERSION + if not _write_registry(reg): + raise OSError("could not persist post-update LXC observation") + return not bool(result.get("error")) + + def refresh_lxc(vmid: int, *, refresh_package_cache: bool = True) -> Optional[dict]: """Detect and refresh exactly one running LXC. diff --git a/scripts/lxc/apply_updates.sh b/scripts/lxc/apply_updates.sh index 3393d7c8..39b46b6a 100755 --- a/scripts/lxc/apply_updates.sh +++ b/scripts/lxc/apply_updates.sh @@ -142,10 +142,26 @@ fi # restore their original stopped state on every exit path. STATE=$(pct status "$VMID" | awk '{print $2}') STARTED_BY_PROXMENUX=0 +OBSERVATION_ATTEMPTED=0 +capture_before_stop() { + if [[ "$STARTED_BY_PROXMENUX" == "1" && "$OBSERVATION_ATTEMPTED" == "0" && -z "${RUN_ID:-}" && "${EXECUTION_MODE:-}" == "web" ]]; then + OBSERVATION_ATTEMPTED=1 + echo "WARNING: post-update verification has no RUN_ID; the result cannot be verified." >&2 + fi + if [[ "$STARTED_BY_PROXMENUX" == "1" && "$OBSERVATION_ATTEMPTED" == "0" && -n "${RUN_ID:-}" ]]; then + OBSERVATION_ATTEMPTED=1 + # Capture while pct exec works. Only the backend writes the registry. + if ! python3 /usr/local/share/proxmenux/monitor-app/usr/bin/lxc_update_observation.py \ + --vmid "$VMID" --run-id "$RUN_ID" --include-apps; then + echo "WARNING: post-update package verification unavailable; the result will remain unverified." >&2 + fi + fi +} restore_original_state() { local rc=$? trap - EXIT INT TERM if [[ "$STARTED_BY_PROXMENUX" == "1" ]]; then + capture_before_stop echo echo "Restoring original state: stopping CT $VMID…" if ! pct shutdown "$VMID" --timeout 60; then @@ -397,6 +413,9 @@ fi # OCI-style CTs whose PID 1 is a user entrypoint (a plain `apt # upgrade` doesn't restart the app; a CT reboot does). if [[ "$RESTART" == "1" ]]; then + # A reboot may return before pct exec is usable again. Keep the verified + # package snapshot from before reboot, then restore the original state. + capture_before_stop echo "--- Rebooting CT $VMID ---" if ! pct reboot "$VMID"; then echo "ERROR: pct reboot failed." >&2 diff --git a/tests/lxc_updates/test_app_observation.py b/tests/lxc_updates/test_app_observation.py new file mode 100644 index 00000000..a5917dd1 --- /dev/null +++ b/tests/lxc_updates/test_app_observation.py @@ -0,0 +1,145 @@ +"""Installed versions must survive a temporary CT's post-update shutdown.""" +import copy +from datetime import datetime, timezone, timedelta +import tempfile +from pathlib import Path +import sys +import unittest +from unittest.mock import Mock, patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'AppImage/scripts')) +import lxc_apps as apps +import lxc_update_observation as handoff + + +class InstalledObservationTests(unittest.TestCase): + def setUp(self): + self.folder = tempfile.TemporaryDirectory() + self.addCleanup(self.folder.cleanup) + self.data = {'apps': [dict(id='docker', name='Docker', installed_via='binary', + binary_path='/usr/bin/docker', state=dict(installed_version='29.8.1', + latest_version='29.9.0', update_available=True, checked_at='2026-01-01T00:00:00Z')), + dict(id='manual', installed_via='manual', installed_version='1.0', + state=dict(installed_version='1.0', latest_version='2.0', update_available=True))], + 'bulk_update': {'targets': ['os', 'app:manual']}, 'schedule': {'enabled': False}} + self.write = Mock(return_value=True) + self.probe = Mock(return_value=('29.9.0', None)) + for name, value in [('_read_sidecar', lambda _: copy.deepcopy(self.data)), + ('_write_sidecar', self.write), + ('detect_installed_version', self.probe)]: + p = patch.object(apps, name, value) + p.start() + self.addCleanup(p.stop) + + def capture(self): + return apps.capture_installed_observations(210) + + def test_new_docker_version_clears_old_pending_without_upstream_or_updater(self): + with patch.object(apps, 'fetch_latest_upstream_details') as upstream: + observations = self.capture() + result = apps.store_installed_observations(210, observations) + self.assertEqual(result['errors'], []) + state = self.write.call_args.args[1]['apps'][0]['state'] + self.assertEqual(state['installed_version'], '29.9.0') + self.assertFalse(state['update_available']) + self.assertEqual(state['latest_version'], '29.9.0') + self.assertEqual(self.probe.call_count, 1, 'manual versions cannot be auto-proven') + upstream.assert_not_called() + self.assertEqual(self.write.call_args.args[1]['bulk_update'], self.data['bulk_update']) + self.assertEqual(self.write.call_args.args[1]['schedule'], self.data['schedule']) + + def test_capture_never_writes_shared_sidecar_and_skips_manual_docker_oci(self): + self.data['apps'].extend([dict(id='oci', installed_via='oci_image'), + dict(id='image', installed_via='docker')]) + self.assertEqual(len(self.capture()), 1) + self.write.assert_not_called() + self.assertEqual(self.probe.call_count, 1) + + def test_failed_probe_preserves_last_known_separately_and_does_not_claim_current(self): + self.probe.return_value = (None, 'secret must never be copied to handoff') + observations = self.capture() + self.assertNotIn('secret', str(observations)) + result = apps.store_installed_observations(210, observations) + self.assertTrue(result['errors']) + state = self.write.call_args.args[1]['apps'][0]['state'] + self.assertEqual(state['last_known_installed_version'], '29.8.1') + self.assertIsNone(state['installed_version']) + self.assertIsNone(state['update_available']) + self.assertTrue(state['error']) + + def test_config_edit_deletion_and_newer_check_cannot_be_overwritten(self): + for kind in ('config', 'deleted', 'newer'): + with self.subTest(kind=kind): + observations = self.capture() + original = copy.deepcopy(self.data) + if kind == 'config': self.data['apps'][0]['binary_path'] = '/other/docker' + if kind == 'deleted': self.data['apps'] = [] + if kind == 'newer': self.data['apps'][0]['state']['checked_at'] = '9999-01-01T00:00:00Z' + result = apps.store_installed_observations(210, observations) + self.assertTrue(result['errors']) + self.write.assert_not_called() + self.data = original + + def test_malformed_duplicate_future_and_expired_observations_never_write_or_probe(self): + observation = self.capture()[0] + for value in (None, {}, [{}], [observation, observation], + [{**observation, 'fingerprint': '../unsafe'}], + [{**observation, 'checked_at': '9999-01-01T00:00:00Z'}], + [{**observation, 'checked_at': '2000-01-01T00:00:00Z'}], + [{**observation, 'installed_version': ['29.9.0']}], + [{**observation, 'verified': 'yes'}]): + with self.subTest(value=value): + self.assertTrue(apps.store_installed_observations(210, value)['errors']) + self.write.assert_not_called() + self.assertEqual(self.probe.call_count, 1) + + def test_stale_boundary_has_margin_and_rejects_only_after_hour(self): + observation = self.capture()[0] + fixed = datetime(2026, 1, 2, tzinfo=timezone.utc) + with patch.object(apps.time, 'time', return_value=fixed.timestamp()): + for age in (3599, 3600, 3601): + self.write.reset_mock() + value = {**observation, 'checked_at': (fixed - timedelta(seconds=age)).isoformat()} + result = apps.store_installed_observations(210, [value]) + self.assertEqual(bool(result['errors']), age > 3600) + self.assertEqual(self.write.called, age <= 3600) + + def test_failed_persistence_is_an_error(self): + self.write.return_value = False + with self.assertRaises(OSError): + apps.store_installed_observations(210, self.capture()) + + def test_installed_probe_does_not_hide_upstream_error(self): + self.data['apps'][0]['state']['error'] = 'upstream unavailable' + apps.store_installed_observations(210, self.capture()) + self.assertEqual(self.write.call_args.args[1]['apps'][0]['state']['error'], 'upstream unavailable') + + +class CaptureScopeTests(unittest.TestCase): + def test_check_only_and_oci_never_capture_app_probes(self): + with tempfile.TemporaryDirectory() as directory, \ + patch.object(handoff, 'OBSERVATION_DIR', Path(directory) / 'observations'), \ + patch.object(handoff.managed_installs, '_detect_lxc_containers', + return_value=[{'_vmid': '210', '_is_oci': True}]), \ + patch.object(handoff.managed_installs, '_check_lxc_updates', return_value={}), \ + patch.object(apps, 'capture_installed_observations') as capture: + handoff.capture(210, 'check-only') + handoff.capture(210, 'oci', include_apps=True) + capture.assert_not_called() + + def test_missing_guest_or_maintenance_preflight_never_capture_apps(self): + with tempfile.TemporaryDirectory() as directory, \ + patch.object(handoff, 'OBSERVATION_DIR', Path(directory) / 'observations'), \ + patch.object(handoff.managed_installs, '_detect_lxc_containers', return_value=[]) as detect, \ + patch.object(handoff.managed_installs, '_probe_lxc_os', return_value='debian'), \ + patch.object(handoff.managed_installs, '_check_lxc_updates', return_value={}), \ + patch.object(handoff.subprocess, 'run', return_value=Mock(returncode=0)), \ + patch.object(apps, 'capture_installed_observations') as capture: + handoff.capture(210, 'missing', include_apps=True) + detect.return_value = [{'_vmid': '210'}] + handoff.capture(210, 'maintenance', refresh_indexes=True, include_apps=True) + capture.assert_not_called() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/lxc_updates/test_stopped_registry_retention.py b/tests/lxc_updates/test_stopped_registry_retention.py new file mode 100644 index 00000000..ecc41ed7 --- /dev/null +++ b/tests/lxc_updates/test_stopped_registry_retention.py @@ -0,0 +1,170 @@ +"""Actual registry/detection/checker consumers; PVE transports are isolated.""" +import ast +import copy +import datetime +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import Mock, patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / 'AppImage/scripts')) +import managed_installs as managed + + +class StoppedRegistryTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + path = Path(self.temp.name) / 'registry.json' + self.now = datetime.datetime.utcnow().isoformat() + 'Z' + self.checked = {'last_check': self.now, 'available': True, 'error': None, + '_count': 1, '_packages': [{'name': 'fixture-package', 'current': '1', 'latest': '2'}]} + self.row = {'id': 'lxc:210', 'type': 'lxc', '_vmid': '210', 'name': 'fixture', + '_os_family': 'debian', 'update_check': copy.deepcopy(self.checked)} + path.write_text(json.dumps({'items': [self.row]})) + self.inventory = [{'vmid': '210', 'status': 'stopped', 'name': 'fixture'}] + patches = [patch.object(managed, '_REGISTRY_PATH', str(path)), + patch.object(managed, '_DB_DIR', self.temp.name), + patch.object(managed, '_DETECTORS', [managed._detect_lxc_containers]), + patch.object(managed, '_lxc_updates_detection_enabled', return_value=True), + patch.object(managed, '_list_pve_lxcs', side_effect=lambda **_: self.inventory), + patch.object(managed, '_get_oci_managed_vmids', return_value={})] + self.forbidden = {} + for name in ['_probe_lxc_os', '_probe_lxc_is_oci', '_identify_helper_slug', '_run_pct_pkg_listing']: + mock = Mock(side_effect=AssertionError('stopped guest probe forbidden')) + self.forbidden[name] = mock + patches.append(patch.object(managed, name, mock)) + for p in patches: + p.start() + self.addCleanup(p.stop) + + def assert_no_guest_probe(self): + for mock in self.forbidden.values(): + mock.assert_not_called() + + def test_startup_detection_preserves_stopped_result_then_notification_check_skips_guest(self): + managed.detect_and_register() + self.assertEqual(managed.get_active_items()[0]['update_check'], self.checked) + checker = Mock(side_effect=AssertionError('checker must not run on stopped guest')) + with patch.object(managed, '_CHECKERS', {'lxc': checker}): + self.assertEqual(managed.check_for_updates(force=True), []) + checker.assert_not_called() + self.assertEqual(managed.get_active_items()[0]['update_check'], self.checked) + self.assert_no_guest_probe() + + def test_monitor_status_projection_survives_process_reinitialization(self): + managed.detect_and_register() + source = ast.parse((ROOT / 'AppImage/scripts/flask_server.py').read_text()) + function = next(n for n in source.body if isinstance(n, ast.FunctionDef) + and n.name == '_get_lxc_update_status_map') + scope = {'os': os} + exec(compile(ast.Module(body=[function], type_ignores=[]), 'actual-status-map', 'exec'), scope) + status = scope['_get_lxc_update_status_map']()['210'] + self.assertEqual(status['last_check'], self.now) + self.assertEqual(status['count'], 1) + self.assertTrue(status['available']) + self.assert_no_guest_probe() + + def test_legacy_stopped_row_removed_by_startup_is_reactivated_without_new_observation(self): + reg = managed._read_registry() + reg['items'][0]['removed_at'] = self.now + managed._write_registry(reg) + managed.detect_and_register() + active = managed.get_active_items() + self.assertEqual(len(active), 1) + self.assertEqual(active[0]['update_check'], self.checked) + self.assert_no_guest_probe() + + def test_genuinely_deleted_guest_and_disabled_detection_remain_removed(self): + for disabled in [False, True]: + managed._write_registry({'items': [copy.deepcopy(self.row)]}) + self.inventory = [] if not disabled else [{'vmid': '210', 'status': 'stopped', 'name': 'fixture'}] + with patch.object(managed, '_lxc_updates_detection_enabled', return_value=not disabled): + managed.detect_and_register() + self.assertEqual(managed.get_active_items(), []) + self.assert_no_guest_probe() + + def test_unknown_stopped_guest_does_not_get_a_fabricated_package_result(self): + self.inventory.append({'vmid': '300', 'status': 'stopped', 'name': 'new'}) + managed.detect_and_register() + self.assertEqual([x['id'] for x in managed.get_active_items()], ['lxc:210']) + self.assert_no_guest_probe() + + def test_targeted_refresh_does_not_start_or_probe_stopped_guest(self): + self.assertIsNone(managed.refresh_lxc(210, refresh_package_cache=False)) + self.assertEqual(managed.get_active_items()[0]['update_check'], self.checked) + self.assert_no_guest_probe() + + def test_inventory_failure_preserves_known_row_without_checker_or_fresh_timestamp(self): + with patch.object(managed, '_list_pve_lxcs', side_effect=RuntimeError('inventory unavailable')): + managed.detect_and_register() + managed.check_for_updates() + self.assertEqual(managed.get_active_items()[0]['update_check'], self.checked) + self.assert_no_guest_probe() + + def test_running_guest_uses_real_checker_and_refreshes_result(self): + self.inventory[0]['status'] = 'running' + checker = Mock(return_value={'available': False, 'last_check': self.now, '_count': 0}) + with patch.object(managed, '_CHECKERS', {'lxc': checker}): + managed.check_for_updates() + checker.assert_called_once() + self.assertFalse(managed.get_active_items()[0]['update_check']['available']) + + def test_presence_merge_does_not_overwrite_a_newer_check_during_detection(self): + fresh = {'entry': {**self.row, '_os_family': 'alpine'}, + 'result': {'available': True, 'last_check': self.now, '_count': 2}} + def concurrent_check(): + self.assertTrue(managed.store_lxc_update_observation(210, fresh)) + return {} + with patch.object(managed, '_get_oci_managed_vmids', side_effect=concurrent_check): + managed.detect_and_register() + row = managed.get_active_items()[0] + self.assertEqual(row['update_check']['_count'], 2) + self.assertEqual(row['_os_family'], 'alpine') + self.assert_no_guest_probe() + + def test_retained_result_still_expires_at_existing_48_hour_boundary(self): + managed.detect_and_register() + current = managed.get_active_items()[0]['update_check'] + now = datetime.datetime.fromisoformat(self.now.rstrip('Z')) + for seconds, expected in [(172799, False), (172800, True), (172801, True)]: + instant = (now + datetime.timedelta(seconds=seconds)).replace(tzinfo=datetime.timezone.utc) + class Clock(datetime.datetime): + @classmethod + def now(cls, tz=None): + return instant + with patch.object(managed.datetime, 'datetime', Clock): + self.assertEqual(managed.lxc_update_freshness(current)['stale'], expected) + self.assert_no_guest_probe() + + +class InventoryContractTests(unittest.TestCase): + def test_default_failure_contract_and_strict_detector_contract(self): + for result in [subprocess.CompletedProcess([], 1, '', ''), subprocess.CompletedProcess([], 0, '', '')]: + with patch.object(managed.subprocess, 'run', return_value=result): + self.assertEqual(managed._list_pve_lxcs(), []) + with self.assertRaises(RuntimeError): + managed._list_pve_lxcs(strict=True) + + def test_timeout_and_malformed_rows_are_not_confirmed_removal(self): + for failure in [subprocess.TimeoutExpired('pct', 5), FileNotFoundError(), OSError()]: + with patch.object(managed.subprocess, 'run', side_effect=failure): + self.assertEqual(managed._list_pve_lxcs(), []) + with self.assertRaises(RuntimeError): + managed._list_pve_lxcs(strict=True) + with patch.object(managed.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, 'VMID Status Lock Name\ninvalid-row\n', '')): + with self.assertRaises(RuntimeError): + managed._list_pve_lxcs(strict=True) + + def test_confirmed_empty_inventory_is_not_a_failed_read(self): + with patch.object(managed.subprocess, 'run', return_value=subprocess.CompletedProcess([], 0, 'VMID Status Lock Name\n', '')): + self.assertEqual(managed._list_pve_lxcs(strict=True), []) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/lxc_updates/test_stopped_update_observation.py b/tests/lxc_updates/test_stopped_update_observation.py new file mode 100644 index 00000000..645b7ef7 --- /dev/null +++ b/tests/lxc_updates/test_stopped_update_observation.py @@ -0,0 +1,476 @@ +"""Real update runner -> trusted handoff -> real completion/API, fake CT only.""" +import ast +from contextlib import redirect_stdout +from datetime import datetime +import io +import json +import os +from pathlib import Path +import re +import shlex +import subprocess +import sys +import threading +import time +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch +import uuid + +import test_observed_state as observed +import managed_installs as managed +import lxc_update_observation as handoff +from flask import request + +ROOT = observed.ROOT +RUNNER = ROOT / "scripts/lxc/apply_updates.sh" +REAL_RUN = subprocess.run + + +class StoppedUpdateObservationTests(unittest.TestCase): + def setUp(self): + self.real_run = REAL_RUN + self.fixture = observed.ObservedStateRoutesTests() + self.fixture.setUp() + self.addCleanup(self.fixture.doCleanups) + self.folder = Path(self.fixture.temp.name) + self.state = self.folder / "state" + self.state.write_text("stopped") + self.log = self.folder / "commands" + self.log.write_text("") + self.bin = self.folder / "bin" + self.bin.mkdir() + self.observations = self.folder / "observations" + ctx = patch.object(handoff, "OBSERVATION_DIR", self.observations) + ctx.start() + self.addCleanup(ctx.stop) + registry = managed.get_registry() + registry["items"][0]["update_check"]["_count"] = 20 + managed._write_registry(registry) + self.env = {**os.environ, "PATH": f"{self.bin}:{os.environ['PATH']}", + "PYTHONPATH": str(ROOT / "AppImage/scripts"), + "FIXTURE_DIR": str(self.folder), "VMID": "100", "TARGET": "os", + "BACKUP": "0", "RESTART": "0", "RUN_HELPER": "0", + "UPDATE_COMMAND": "", "RUN_ID": "manual-fixture", + "PROXMENUX_LOCK_DIR": str(self.folder), "REMAINING": "0"} + self.write("pct", f"#!{sys.executable}\n" + ''' +import json, os, pathlib, sys +root = pathlib.Path(os.environ['FIXTURE_DIR']) +args = sys.argv[1:] +with (root/'commands').open('a') as log: log.write(json.dumps(args)+'\\n') +state = root/'state' +if args[0] == 'list': print('VMID Status Name\\n100 '+state.read_text()+' fixture') +elif args[0] == 'status': print('status: '+state.read_text()) +elif args[0] == 'start': state.write_text('running') +elif args[0] == 'shutdown': + if os.environ.get('SHUTDOWN_FAIL') == '1': sys.exit(1) + state.write_text('stopped') +elif args[0] == 'reboot': + if os.environ.get('REBOOT_FAIL') == '1': sys.exit(1) + if os.environ.get('REBOOT_TRANSITION') == '1': state.write_text('restarting') +elif args[0] == 'exec': + if state.read_text() != 'running': sys.exit(1) + command = ' '.join(args[3:]) + if '/etc/os-release' in command: print('ID=debian') + elif 'dist-upgrade' in command: + if os.environ.get('UPDATE_FAIL') == '1': sys.exit(4) + if os.environ.get('CLEAR_AFTER_UPGRADE') == '1': (root/'remaining').write_text('0') + elif 'apt-get update' in command: + if os.environ.get('INDEX_FAIL') == '1': sys.exit(3) + elif 'apt list --upgradable' in command: + if os.environ.get('PROBE_FAIL') == '1': sys.exit(1) + print('Listing...') + pending = (root/'remaining').read_text() if (root/'remaining').exists() else os.environ['REMAINING'] + for n in range(int(pending)): + print(f'pkg{n}/stable 2.0 amd64 [upgradable from: 1.0]') + elif command == '/usr/bin/docker --version': + if os.environ.get('DOCKER_PROBE_FAIL') == '1': sys.exit(1) + print('Docker version 29.9.0') + else: sys.exit(91) +else: sys.exit(92) +''') + driver = self.folder / "capture.py" + driver.write_text(''' +import os, pathlib +import managed_installs as m +import lxc_update_observation as h +import lxc_apps as apps +root = pathlib.Path(os.environ['FIXTURE_DIR']) +m._DB_DIR = str(root) +m._REGISTRY_PATH = str(root/'registry.json') +m._PCT_BIN = str(root/'bin/pct') +h.OBSERVATION_DIR = root/'observations' +apps._APPS_DIR = str(root/'apps') +apps._PCT_BIN = str(root/'bin/pct') +with (root/'commands').open('a') as log: log.write('CAPTURE\\n') +if os.environ.get('CAPTURE_FAIL') == '1': raise SystemExit(1) +m._detect_lxc_containers = lambda **_: ([{'id':'lxc:100','type':'lxc','_vmid':'100', + '_is_oci':os.environ.get('OCI') == '1'}] if (root/'state').read_text() == 'running' else []) +raise SystemExit(h.main()) +''') + self.write("python3", "#!/bin/bash\n" + f"exec {shlex.quote(sys.executable)} {shlex.quote(str(driver))} \"${{@:2}}\"\n") + self.write("sleep", "#!/bin/sh\nexit 0\n") + self.write("flock", "#!/bin/sh\nexit 0\n") + names = {"_finalize_lxc_update", "_terminal_lxc_update_completed", + "api_lxc_updates_applied", "_run_scheduled_update"} + nodes = [n for n in ast.parse(observed.SERVER.read_text()).body + if isinstance(n, ast.FunctionDef) and n.name in names] + self.scope = self.fixture.scope + self.scope.update({ + "os": os, "re": re, "time": time, "uuid": uuid, "datetime": datetime, "json": json, + "request": request, "_LXC_APPLY_UPDATES_SCRIPT": str(RUNNER), + "_LXC_INSTALL_CHECKED_OS_SCRIPT": str(ROOT / 'scripts/lxc/install_checked_os_updates.sh'), + "_APPLY_UPDATES_SCRIPT": str(RUNNER), + "_normalise_lxc_update_run_id": lambda value, **_: value or "manual-fixture", + "_normalise_lxc_update_targets": lambda values, target: values or [target], + "_normalise_lxc_update_labels": lambda values: values or [], + "_json_list": lambda value: json.loads(value or '[]'), + "_lxc_update_finalizations": {}, "_lxc_update_finalization_lock": threading.RLock(), + "_LXC_UPDATE_FINALIZATION_TTL": 3600, + "_fast_guest_status": lambda *_: self.state.read_text(), + "_lxc_update_snapshot": lambda *_: {"ct_name": "fixture"}, + "_lxc_update_target_labels": lambda *_: ["OS"], + "_lxc_update_details": lambda **kwargs: str(kwargs), + "_publish_guest_modal_cache_revision": Mock(), + "notification_manager": Mock(), + "_create_lxc_update_log": lambda *_: (None, None), + "_append_lxc_update_log": lambda *_: None, + "_prune_lxc_update_logs": lambda *_: None, + "_normalise_schedule_targets": lambda sched: sched.get('targets') or ['os'], + "_scheduled_helper_enabled": lambda *_: False, + "_compose_scheduled_update_command": lambda *_, **__: '', + "_inspect_lxc_reboot_requirement": lambda *_, **__: (False, [], None), + }) + exec(compile(ast.Module(body=nodes, type_ignores=[]), 'update-completion', 'exec'), self.scope) + + def write(self, name, content): + path = self.bin / name + path.write_text(content) + path.chmod(0o755) + + def run_manual(self, **extra): + env = {**self.env, **extra} + result = self.real_run(['/bin/bash', str(RUNNER)], env=env, text=True, + capture_output=True, timeout=20) + # The runner's separate process has not modified the shared registry. + self.assertEqual(managed.get_registry()['items'][0]['update_check']['_count'], 20) + self.scope['_terminal_lxc_update_completed'](script_path=str(RUNNER), + params={'RUN_ID': env.get('RUN_ID'), 'VMID': '100', 'TARGET': 'os'}, + exit_code=result.returncode, duration_seconds=1) + return result + + def test_manual_hook_stores_verified_zero_after_shutdown_without_restart(self): + result = self.run_manual() + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertEqual(self.state.read_text(), 'stopped') + row = self.fixture.row()['update_check'] + self.assertEqual(row['count'], 0) + self.assertIsNone(row['error']) + self.assertEqual(row['os_family'], 'debian') + log = self.log.read_text() + self.assertLess(log.index('CAPTURE'), log.index('shutdown')) + self.assertLess(log.index('apt list --upgradable'), log.index('shutdown')) + self.assertNotIn('apt-get update', log) + self.assertEqual(log.count('dist-upgrade'), 1) + self.assertFalse(list(self.observations.glob('*.json'))) + self.assertEqual(managed.get_registry()['items'][1]['update_check']['_count'], 7) + self.assertEqual(self.fixture.commands, [], 'GET and stopped finalization run no guest probes') + + def test_partial_update_keeps_verified_remaining_packages(self): + result = self.run_manual(REMAINING='3', UPDATE_FAIL='1') + self.assertEqual(result.returncode, 4) + self.assertEqual(self.state.read_text(), 'stopped') + self.assertEqual(self.fixture.row()['update_check']['count'], 3) + + def assert_unverified(self, flag): + result = self.run_manual(**{flag: '1'}) + self.assertEqual(result.returncode, 0, 'verification failure is not updater failure') + self.assertEqual(self.state.read_text(), 'stopped') + stored = managed.get_registry()['items'][0]['update_check'] + self.assertTrue(stored['error']) + self.assertNotIn('_count', stored) + self.assertTrue(self.fixture.row()['update_check']['error']) + + def test_failed_probe_does_not_present_old_count_or_fresh_zero(self): + self.assert_unverified('PROBE_FAIL') + + def test_missing_capture_does_not_present_old_count_or_fresh_zero(self): + self.assert_unverified('CAPTURE_FAIL') + + def test_web_runner_without_identity_reports_unverified_instead_of_silent_skip(self): + env = {**self.env, 'EXECUTION_MODE': 'web'} + env.pop('RUN_ID') + result = self.real_run(['/bin/bash', str(RUNNER)], env=env, + capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0) + self.assertEqual(self.state.read_text(), 'stopped') + self.assertIn('verification has no RUN_ID', result.stderr) + self.assertNotIn('CAPTURE', self.log.read_text()) + + def test_rejected_handoff_is_retained_with_safe_diagnostic_and_no_false_zero(self): + self.observations.mkdir(mode=0o700) + path = self.observations / '100-manual-fixture.json' + path.write_text(json.dumps({'private_payload': 'must-not-be-logged'})) + path.chmod(0o600) + output = io.StringIO() + with redirect_stdout(output): + self.scope['_finalize_lxc_update'](100, run_id='manual-fixture', + status='success', source='manual', target='os') + self.assertTrue(path.exists()) + self.assertIn('reason=KeyError', output.getvalue()) + self.assertIn('handoff=False verified=False', output.getvalue()) + self.assertNotIn('must-not-be-logged', output.getvalue()) + self.assertTrue(self.fixture.row()['update_check']['error']) + self.assertNotIn('_count', managed.get_registry()['items'][0]['update_check']) + + def test_frontend_completion_route_deduplicates_authoritative_hook(self): + self.run_manual() + class InlineThread: + def __init__(self, target, args, kwargs, **_): + self.callback = lambda: target(*args, **kwargs) + def start(self): + self.callback() + self.scope['threading'] = SimpleNamespace(Thread=InlineThread) + response = self.fixture.client.post('/api/lxc-updates/100/applied', json={ + 'success': True, 'run_id': 'manual-fixture', 'target': 'os'}) + self.assertEqual(response.status_code, 202) + self.assertEqual(self.fixture.row()['update_check']['count'], 0) + self.scope['notification_manager'].emit_event.assert_called_once() + + def test_entrypoint_and_imported_server_share_finalization_for_same_run(self): + # AppRun executes flask_server as __main__; health collectors later + # import flask_server by name. Those namespaces must not own separate + # finalization maps when the terminal hook and HTTP callback collide. + tree = ast.parse(observed.SERVER.read_text()) + state_names = {'_lxc_update_finalization_lock', '_lxc_update_finalizations'} + initializers = [] + for node in tree.body: + if isinstance(node, ast.ImportFrom) and node.module == 'lxc_update_finalization': + initializers.append(node) + elif isinstance(node, ast.Assign) and any( + isinstance(target, ast.Name) and target.id in state_names for target in node.targets): + initializers.append(node) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name) and node.target.id in state_names: + initializers.append(node) + hook_scope = dict(self.scope) + exec(compile(ast.Module(body=initializers, type_ignores=[]), 'entrypoint-state', 'exec'), hook_scope) + exec(compile(ast.Module(body=initializers, type_ignores=[]), 'imported-state', 'exec'), self.scope) + for scope in (hook_scope, self.scope): + scope['_lxc_update_finalizations'].clear() + self.addCleanup(scope['_lxc_update_finalizations'].clear) + functions = [node for node in tree.body if isinstance(node, ast.FunctionDef) + and node.name in {'_finalize_lxc_update', '_terminal_lxc_update_completed'}] + exec(compile(ast.Module(body=functions, type_ignores=[]), 'entrypoint-hook', 'exec'), hook_scope) + self.scope['_terminal_lxc_update_completed'] = hook_scope['_terminal_lxc_update_completed'] + self.run_manual() + self.assertEqual(self.fixture.row()['update_check']['count'], 0) + class InlineThread: + def __init__(self, target, args, kwargs, **_): + self.callback = lambda: target(*args, **kwargs) + def start(self): + self.callback() + self.scope['threading'] = SimpleNamespace(Thread=InlineThread) + response = self.fixture.client.post('/api/lxc-updates/100/applied', json={ + 'success': True, 'run_id': 'manual-fixture', 'target': 'os'}) + self.assertEqual(response.status_code, 202) + result = self.fixture.row()['update_check'] + self.assertEqual(result['count'], 0) + self.assertIsNone(result['error']) + self.scope['notification_manager'].emit_event.assert_called_once() + + def test_scheduled_path_executes_real_runner_and_real_finalization(self): + def run(command, **kwargs): + kwargs['env'].update({key: value for key, value in self.env.items() + if key != 'RUN_ID'}) + return self.real_run(command, **kwargs) + self.scope['subprocess'] = SimpleNamespace(run=run, TimeoutExpired=subprocess.TimeoutExpired) + result = self.scope['_run_scheduled_update'](100, {'targets': ['os'], 'target': 'os'}) + self.assertEqual(result['status'], 'success', result) + self.assertEqual(self.state.read_text(), 'stopped') + self.assertEqual(self.fixture.row()['update_check']['count'], 0) + + def test_legacy_cli_without_run_id_preserves_update_and_original_state(self): + env = dict(self.env) + env.pop('RUN_ID') + result = self.real_run(['/bin/bash', str(RUNNER)], env=env, + capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0) + self.assertEqual(self.state.read_text(), 'stopped') + self.assertNotIn('CAPTURE', self.log.read_text()) + + def test_appimage_packaging_copies_the_actual_observation_helper(self): + builder = (ROOT / 'AppImage/scripts/build_appimage.sh').read_text() + copies = [line for line in builder.splitlines() + if line.startswith('cp ') and '$SCRIPT_DIR/lxc_update_observation.py' in line] + self.assertEqual(len(copies), 1) + appdir = self.folder / 'AppDir' + (appdir / 'usr/bin').mkdir(parents=True) + result = self.real_run(['/bin/bash', '-ec', copies[0]], env={**os.environ, + 'SCRIPT_DIR': str(ROOT / 'AppImage/scripts'), 'APP_DIR': str(appdir)}, + capture_output=True, text=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual((appdir / 'usr/bin/lxc_update_observation.py').read_bytes(), + (ROOT / 'AppImage/scripts/lxc_update_observation.py').read_bytes()) + shared = [line for line in builder.splitlines() + if line.startswith('cp ') and '$SCRIPT_DIR/lxc_update_finalization.py' in line] + self.assertEqual(len(shared), 1) + result = self.real_run(['/bin/bash', '-ec', shared[0]], env={**os.environ, + 'SCRIPT_DIR': str(ROOT / 'AppImage/scripts'), 'APP_DIR': str(appdir)}, + capture_output=True, text=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual((appdir / 'usr/bin/lxc_update_finalization.py').read_bytes(), + (ROOT / 'AppImage/scripts/lxc_update_finalization.py').read_bytes()) + + def test_bad_or_untrusted_handoffs_are_not_accepted(self): + self.run_manual() + self.observations.mkdir(exist_ok=True) + path = self.observations / '100-manual-fixture.json' + for body in ('{}', '[]', '{bad'): + path.write_text(body) + path.chmod(0o600) + self.assertIsNone(handoff.read(100, 'manual-fixture')) + path.unlink() + path.symlink_to(self.folder / 'registry.json') + self.assertIsNone(handoff.read(100, 'manual-fixture')) + with self.assertRaises(ValueError): + handoff.read(100, '../../escape') + + def test_wrong_run_expired_incomplete_and_public_observations_are_rejected(self): + handoff.capture(100, 'manual-fixture') + original = handoff.read(100, 'manual-fixture') + self.assertIsNotNone(original) + path = self.observations / '100-manual-fixture.json' + cases = [] + for key, value in (('vmid', 200), ('run_id', 'wrong-run'), + ('captured_at', '2020-01-01T00:00:00+00:00'), + ('captured_at', '2999-01-01T00:00:00+00:00')): + cases.append({**original, key: value}) + cases.append({**original, 'result': {'available': False, 'last_check': 'now'}}) + for data in cases: + path.write_text(json.dumps(data)) + self.assertIsNone(handoff.read(100, 'manual-fixture')) + path.write_text(json.dumps(original)) + path.chmod(0o644) + self.assertIsNone(handoff.read(100, 'manual-fixture')) + path.unlink() + os.mkfifo(path, mode=0o600) + self.assertIsNone(handoff.read(100, 'manual-fixture')) + + def test_oci_observation_never_runs_a_generic_package_listing(self): + result = self.run_manual(OCI='1') + self.assertEqual(result.returncode, 0) + self.assertNotIn('apt list --upgradable', self.log.read_text()) + self.assertTrue(self.fixture.row()['update_check']['is_oci_lxc']) + + def test_oci_handoff_without_family_accepts_only_complete_skipped_zero(self): + with patch.object(managed, '_probe_lxc_is_oci', return_value=True): + observation = handoff.capture(100, 'manual-fixture') + observation['entry']['_os_family'] = None + path = self.observations / '100-manual-fixture.json' + path.write_text(json.dumps(observation)) + self.assertIsNotNone(handoff.read(100, 'manual-fixture')) + for patch_result in ({'available': True}, {'_count': 1}, + {'_security_count': 1}, {'_packages': [{'name': 'unexpected'}]}): + path.write_text(json.dumps({**observation, 'result': {**observation['result'], **patch_result}})) + self.assertIsNone(handoff.read(100, 'manual-fixture')) + observation['entry']['_is_oci'] = False + path.write_text(json.dumps(observation)) + self.assertIsNone(handoff.read(100, 'manual-fixture'), 'non-OCI still needs a verified supported family') + + def test_reboot_failure_preserves_exit_code_and_still_verifies_before_stop(self): + result = self.run_manual(RESTART='1', REBOOT_FAIL='1') + self.assertEqual(result.returncode, 6) + self.assertEqual(self.state.read_text(), 'stopped') + self.assertEqual(self.fixture.row()['update_check']['count'], 0) + + def test_capture_precedes_reboot_and_is_not_retried_in_transitional_guest(self): + result = self.run_manual(RESTART='1', REBOOT_TRANSITION='1') + self.assertEqual(result.returncode, 0) + self.assertEqual(self.state.read_text(), 'stopped') + self.assertEqual(self.fixture.row()['update_check']['count'], 0) + log = self.log.read_text() + self.assertLess(log.index('CAPTURE'), log.index('reboot')) + self.assertEqual(log.count('CAPTURE'), 1) + + def test_registry_failure_does_not_discard_the_valid_observation(self): + result = self.real_run(['/bin/bash', str(RUNNER)], env=self.env, + capture_output=True, text=True, timeout=20) + self.assertEqual(result.returncode, 0) + with patch.object(managed, '_write_registry', return_value=False): + outcome = self.scope['_finalize_lxc_update'](100, run_id='manual-fixture', + status='success', source='manual', target='os') + self.assertTrue(outcome['verification_errors']) + self.assertIsNotNone(handoff.read(100, 'manual-fixture')) + self.assertEqual(managed.get_registry()['items'][0]['update_check']['_count'], 20) + + def test_originally_running_guest_keeps_existing_finalization_path(self): + self.state.write_text('running') + import lxc_apps + with patch.object(lxc_apps, 'mark_manual_versions_unverified'), \ + patch.object(lxc_apps, 'check_all'): + result = self.run_manual() + self.assertEqual(result.returncode, 0) + self.assertEqual(self.state.read_text(), 'running') + self.assertNotIn('CAPTURE', self.log.read_text()) + self.assertEqual(self.fixture.row()['update_check']['count'], 0) + + def test_stopped_finalizer_consumes_captured_app_versions_without_guest_probes(self): + import lxc_apps + self.state.write_text('running') + observation = handoff.capture(100, 'manual-fixture') + observation['apps'] = [{'id': 'docker', 'installed_version': '29.9.0'}] + self.state.write_text('stopped') + with patch.object(handoff, 'read', return_value=observation), \ + patch.object(lxc_apps, 'store_installed_observations', create=True, + return_value={'errors': [], 'applied': 1}) as store, \ + patch.object(lxc_apps, 'check_all') as check, \ + patch.object(lxc_apps, 'mark_manual_versions_unverified') as manual: + self.scope['_finalize_lxc_update'](100, run_id='manual-fixture', + status='success', source='manual', target='both', + requested_targets=['os', 'app:ghstats']) + store.assert_called_once_with(100, observation['apps']) + manual.assert_called_once_with(100, ['ghstats']) + check.assert_not_called() + + def test_real_os_runner_updates_registered_docker_before_shutdown_and_backend_merge(self): + import lxc_apps + folder = self.folder / 'apps' + folder.mkdir() + sidecar = folder / '100.json' + original = {'apps': [{'id': 'docker', 'installed_via': 'binary', + 'binary_path': '/usr/bin/docker', 'state': {'installed_version': '29.8.1', + 'latest_version': '29.9.0', 'checked_at': '2026-01-01T00:00:00Z', + 'update_available': True}}], 'schedule': {'enabled': False}} + sidecar.write_text(json.dumps(original)) + with patch.object(lxc_apps, '_APPS_DIR', str(folder)), \ + patch.object(lxc_apps, 'check_all') as check: + result = self.run_manual() + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertEqual(self.state.read_text(), 'stopped') + saved = json.loads(sidecar.read_text()) + self.assertEqual(saved['apps'][0]['state']['installed_version'], '29.9.0') + self.assertFalse(saved['apps'][0]['state']['update_available']) + self.assertEqual(saved['schedule'], original['schedule']) + commands = self.log.read_text() + self.assertLess(commands.index('/usr/bin/docker'), commands.index('shutdown')) + self.assertEqual(commands.count('/usr/bin/docker'), 1) + check.assert_not_called() + + def test_failed_app_write_retains_handoff_and_does_not_claim_verified(self): + import lxc_apps + self.state.write_text('running') + observation = handoff.capture(100, 'manual-fixture') + observation['apps'] = [] + path = handoff._path(100, 'manual-fixture') + path.write_text(json.dumps(observation)) + self.state.write_text('stopped') + with patch.object(lxc_apps, 'store_installed_observations', side_effect=OSError): + outcome = self.scope['_finalize_lxc_update'](100, run_id='manual-fixture', + status='success', source='manual', target='os') + self.assertTrue(path.exists()) + self.assertIn('post-update application persistence failed', outcome['verification_errors']) + + +if __name__ == '__main__': + unittest.main()