Merge pull request #332 from Vaso73/fix/runtime-verified-app-detectors

fix: improve application version detection
This commit is contained in:
MacRimi
2026-09-03 22:25:41 +02:00
committed by GitHub
7 changed files with 515 additions and 13 deletions
+2
View File
@@ -5887,6 +5887,8 @@ const handleDownloadLogs = async (vmid: number, vmName: string) => {
{
updateCommand: aw.update_command!,
appName: aw.name || "",
targetIds: [`app:${aw.id}`],
targetLabels: [aw.name || t("vmLxc.updates.applicationDefaultName")],
},
)}
className={hasUpdate ? pendingBtnCls : upToDate ? upToDateBtnCls : neutralBtnCls}
+1
View File
@@ -147,6 +147,7 @@ chmod +x "$APP_DIR/usr/bin/recreate_docker_container.py" 2>/dev/null || true
cp "$SCRIPT_DIR/update_docker_engine.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ update_docker_engine.py not found"
chmod +x "$APP_DIR/usr/bin/update_docker_engine.py" 2>/dev/null || true
cp "$APPIMAGE_ROOT/../json/app_tracking_hints.json" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ app_tracking_hints.json not found"
cp "$APPIMAGE_ROOT/../json/runtime_verified_overrides.json" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ runtime_verified_overrides.json not found"
cp "$SCRIPT_DIR/flask_terminal_routes.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ flask_terminal_routes.py not found"
cp "$SCRIPT_DIR/hardware_monitor.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ hardware_monitor.py not found"
cp "$SCRIPT_DIR/temperature_sensor_resolver.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️ temperature_sensor_resolver.py not found"
+10
View File
@@ -14292,6 +14292,16 @@ def _finalize_lxc_update(
verification_errors.append(f'OS refresh failed: {exc}')
try:
import lxc_apps
# Per-app custom commands may update software that uses a
# manually typed version. Their successful exit does not prove
# a new version, so invalidate that old claim before the forced
# refresh. The UI then stays honest instead of showing a stale
# "update available" result.
if status == 'success':
lxc_apps.mark_manual_versions_unverified(
vmid,
[item[4:] for item in requested if item.startswith('app:')],
)
refreshed_sidecar = lxc_apps.check_all(vmid, force=True)
refreshed_sidecar = refreshed_sidecar or {'vmid': vmid, 'apps': []}
_vm_cache_put(_vm_apps_cache, vmid, refreshed_sidecar)
+193 -10
View File
@@ -136,6 +136,14 @@ _TRACKING_HINTS_BUNDLED = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"app_tracking_hints.json",
)
# Runtime-verified detector overrides ship with the AppImage. Unlike the
# regular catalog, they are deliberately not fetched from main: an installed
# Monitor may otherwise download an older catalog entry that resurrects a
# stale helper marker over a detector verified on a real container.
_RUNTIME_VERIFIED_OVERRIDES_BUNDLED = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"runtime_verified_overrides.json",
)
_tracking_hints_lock = threading.RLock()
_tracking_hints_cache: Optional[dict] = None
_tracking_hints_ts: float = 0.0
@@ -189,16 +197,109 @@ def _load_bundled_hints() -> dict:
return {}
def _load_runtime_verified_overrides() -> dict:
"""Load optional live-detector promotions packaged with the Monitor."""
try:
with open(_RUNTIME_VERIFIED_OVERRIDES_BUNDLED) as f:
data = json.load(f)
apps = data.get("apps") if isinstance(data, dict) else None
return apps if isinstance(apps, dict) else {}
except (OSError, json.JSONDecodeError):
return {}
def _is_helper_marker_detector(detector: dict) -> bool:
return (
detector.get("installed_via") == "file"
and bool(re.fullmatch(
r"/root/\.[A-Za-z0-9_.-]+", str(detector.get("file_path") or "")
))
)
def _apply_runtime_verified_overrides(hints: dict, overrides: dict) -> dict:
"""Promote packaged, runtime-proven detectors over a remote catalog.
The remote catalog remains the normal no-rebuild update channel. Entries
marked non-operational are skipped; all other runtime-verified entries
replace detector fields. Presentation metadata and non-conflicting
fallbacks remain intact. A legacy ``/root/.<app>`` primary is retained as
a fallback so older helper layouts do not lose their only version source.
"""
result = {
slug: dict(hint) for slug, hint in (hints or {}).items()
if isinstance(slug, str) and isinstance(hint, dict)
}
for slug, spec in (overrides or {}).items():
if not isinstance(slug, str) or not isinstance(spec, dict):
continue
detector = spec.get("detector")
if not bool(spec.get("operational", True)) or not isinstance(detector, dict):
continue
method = detector.get("installed_via")
if method not in _VALID_METHODS:
continue
current = dict(result.get(slug) or {})
marker_fallbacks = []
if _is_helper_marker_detector(current):
marker_fallbacks.append({
"path": current["file_path"],
"regex": current.get("file_regex") or current.get("installed_regex") or "",
"source": "helper_marker",
})
# Detector and upstream fields must be replaced as one coherent set;
# retaining e.g. an old file_path next to a binary detector is what
# previously kept stale helper marker versions alive.
for key in _DETECTOR_FIELDS + (
"installed_via", "repo", "github_source", "tag_regex",
"installed_regex", "upstream_type", "upstream_url",
"upstream_json_path", "docker_image",
):
current.pop(key, None)
current.update(detector)
fallbacks = []
for candidate in (current.get("file_fallbacks") or []):
if isinstance(candidate, dict) and candidate.get("path"):
fallbacks.append(dict(candidate))
spec_fallbacks = spec.get("file_fallbacks")
if not isinstance(spec_fallbacks, list):
spec_fallbacks = []
for candidate in spec_fallbacks + marker_fallbacks:
if isinstance(candidate, dict) and candidate.get("path"):
fallbacks.append(dict(candidate))
if fallbacks:
unique_fallbacks = []
known_paths = set()
for candidate in fallbacks:
path = candidate.get("path")
if path in known_paths:
continue
known_paths.add(path)
unique_fallbacks.append(candidate)
current["file_fallbacks"] = unique_fallbacks
# These optional fields are explicitly allowed to update runtime
# behavior too, while ordinary catalog presentation remains untouched.
for key in ("alt_detectors", "default_ports", "logo", "website"):
if key in spec:
current[key] = spec[key]
result[slug] = current
return result
def _fetch_tracking_hints() -> dict:
"""Return the curated tracking-hint map (slug → hint dict).
Fetch order: memory cache (fresh) → GitHub raw merged with the bundled
catalog → on-disk cache → bundled JSON. A newly built AppImage can contain
a larger catalog than ``main`` while a phase is being validated; in that
case the smaller remote file must not erase bundled detectors. When the
remote catalog has equal or greater coverage it wins per entry, preserving
the no-rebuild update path. Never raises — a total failure returns an empty
dict so callers can just ``.get(slug)``.
catalog → on-disk cache → bundled JSON. Packaged runtime-verified
overrides are then applied to every source. They prevent a stale remote
catalog from downgrading a detector already proven live, while all normal
catalog updates continue to arrive without an AppImage rebuild. Never
raises — a total failure returns an empty dict so callers can just
``.get(slug)``.
"""
global _tracking_hints_cache, _tracking_hints_ts
with _tracking_hints_lock:
@@ -206,6 +307,7 @@ def _fetch_tracking_hints() -> dict:
if _tracking_hints_cache is not None and (now - _tracking_hints_ts) < _TRACKING_HINTS_TTL:
return _tracking_hints_cache
bundled = _load_bundled_hints()
runtime_overrides = _load_runtime_verified_overrides()
try:
req = urllib.request.Request(
_TRACKING_HINTS_URL,
@@ -220,6 +322,7 @@ def _fetch_tracking_hints() -> dict:
else:
hints = dict(remote)
hints.update(bundled)
hints = _apply_runtime_verified_overrides(hints, runtime_overrides)
_tracking_hints_cache = hints
_tracking_hints_ts = now
try:
@@ -237,11 +340,15 @@ def _fetch_tracking_hints() -> dict:
try:
with open(_TRACKING_HINTS_DISK) as f:
disk = json.load(f)
_tracking_hints_cache = disk.get("hints") or {}
_tracking_hints_cache = _apply_runtime_verified_overrides(
disk.get("hints") or {}, runtime_overrides
)
_tracking_hints_ts = float(disk.get("ts") or 0)
return _tracking_hints_cache
except (OSError, json.JSONDecodeError):
_tracking_hints_cache = bundled
_tracking_hints_cache = _apply_runtime_verified_overrides(
bundled, runtime_overrides
)
_tracking_hints_ts = now # avoid re-hammering
return _tracking_hints_cache
@@ -2580,6 +2687,40 @@ def update_app(vmid, app_id: str, payload: dict) -> tuple[bool, Any]:
return True, _read_sidecar(vmid)
def mark_manual_versions_unverified(vmid, app_ids) -> int:
"""Invalidate manual version claims after their own updater succeeds.
A custom update command is intentionally arbitrary shell owned by the
operator. Its exit code says that the command completed, not which
version is now installed. For manually tracked applications, retaining
the old typed value would create a false update warning (or a false
"current" result). Mark only explicitly targeted manual apps so a
helper-wide update never changes unrelated registrations.
"""
wanted = {
str(app_id).strip()
for app_id in (app_ids or [])
if isinstance(app_id, str) and str(app_id).strip()
}
if not wanted:
return 0
with _cache_lock:
sidecar = _read_sidecar(vmid)
if not sidecar:
return 0
changed = 0
for app in sidecar.get("apps") or []:
if app.get("id") not in wanted or app.get("installed_via") != "manual":
continue
if not app.get("manual_version_needs_confirmation"):
app["manual_version_needs_confirmation"] = True
changed += 1
if changed:
sidecar["updated_at"] = _now_iso()
_write_sidecar(vmid, sidecar)
return changed
def delete_app(vmid, app_id: str) -> bool:
with _cache_lock:
sidecar = _read_sidecar(vmid)
@@ -3239,13 +3380,55 @@ def _detect_with_alt_healing(vmid, app: dict) -> tuple:
``healed_bool`` is True when the working detector was an alt and
the app dict was rewritten.
"""
slug = app.get("helper_slug")
hint = (_fetch_tracking_hints() or {}).get(slug) or {}
# A modern Community Scripts marker (/root/.<app>) is a useful
# fallback, but it is not a live process probe. It can stay behind when
# an operator upgrades an application outside the helper script. When a
# later runtime-verified catalog entry promotes a non-marker primary
# detector, migrate existing marker-backed registrations to that stronger
# detector on their next check. This is deliberately generic: adding a
# verified runtime override for another helper app automatically repairs
# its already-saved sidecars too.
is_helper_marker = (
app.get("installed_via") == "file"
and re.fullmatch(r"/root/\.[A-Za-z0-9_.-]+", str(app.get("file_path") or ""))
)
if is_helper_marker and isinstance(hint, dict):
primary = {"installed_via": hint.get("installed_via")}
for key in _DETECTOR_FIELDS:
if key in hint:
primary[key] = hint[key]
primary_probe = _detector_probe_config(hint, primary)
primary_is_marker = (
primary_probe.get("installed_via") == "file"
and re.fullmatch(r"/root/\.[A-Za-z0-9_.-]+", str(primary_probe.get("file_path") or ""))
)
if primary_probe.get("installed_via") and not primary_is_marker:
primary_installed, _primary_error = detect_installed_version(vmid, primary_probe)
if primary_installed:
# Keep user-owned presentation and updater fields, replacing
# only the detector configuration that was previously a stale
# helper marker. The hint continues to supply its marker as
# a fallback if the live probe disappears on an older layout.
for key in _DETECTOR_FIELDS:
app.pop(key, None)
app.update(primary_probe)
return primary_installed, None, True
if app.get("installed_via") == "manual" and app.get("manual_version_needs_confirmation"):
# A successful arbitrary user command does not prove what version it
# installed. Never present the previously typed manual value as a
# fresh observation; the user can save the app again once they have a
# trustworthy version source or value.
return None, None, False
installed, err = detect_installed_version(vmid, app)
if installed or not err:
return installed, err, False
slug = app.get("helper_slug")
if not slug:
return installed, err, False
hint = (_fetch_tracking_hints() or {}).get(slug) or {}
# Build a unified fallback list from both:
# • alt_detectors — cross-method (file→binary, file→dpkg, …)
# • file_fallbacks — same-method secondary file paths (legacy
@@ -0,0 +1,274 @@
import json
import sys
import tempfile
import unittest
import importlib.util
from pathlib import Path
from unittest import mock
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
if str(SCRIPTS_DIR) not in sys.path:
sys.path.insert(0, str(SCRIPTS_DIR))
import lxc_apps
GENERATOR_PATH = Path(__file__).resolve().parents[3] / ".github" / "scripts" / "generate_app_tracking_catalog.py"
GENERATOR_SPEC = importlib.util.spec_from_file_location("app_tracking_generator_under_test", GENERATOR_PATH)
app_tracking_generator = importlib.util.module_from_spec(GENERATOR_SPEC)
sys.modules[GENERATOR_SPEC.name] = app_tracking_generator
GENERATOR_SPEC.loader.exec_module(app_tracking_generator)
VERSION_RE = r"(?i)(?:v|release[-_/]?)?(\d+(?:\.\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)"
TRAEFIK_LIVE_RE = r"(?m)^Version:\s*(\d+(?:\.\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)"
class LiveVersionDetectionTests(unittest.TestCase):
def test_runtime_override_wins_over_stale_remote_helper_marker(self):
"""A live detector must survive an older remote catalog entry."""
stale_remote = {
"traefik": {
"installed_via": "file",
"file_path": "/root/.traefik",
"file_regex": VERSION_RE,
"repo": "traefik/traefik",
"tag_regex": VERSION_RE,
}
}
overrides = {
"traefik": {
"operational": True,
"detector": {
"installed_via": "binary",
"binary_path": "/usr/bin/traefik",
"binary_args": ["version"],
"installed_regex": TRAEFIK_LIVE_RE,
"repo": "traefik/traefik",
"tag_regex": VERSION_RE,
},
}
}
result = lxc_apps._apply_runtime_verified_overrides(stale_remote, overrides)
self.assertEqual(result["traefik"]["installed_via"], "binary")
self.assertEqual(result["traefik"]["binary_path"], "/usr/bin/traefik")
self.assertNotIn("file_path", result["traefik"])
self.assertEqual(
result["traefik"]["file_fallbacks"],
[{"path": "/root/.traefik", "regex": VERSION_RE, "source": "helper_marker"}],
)
def test_fetch_applies_runtime_override_after_remote_catalog_merge(self):
"""The actual fetch path must not let a remote stale entry win."""
remote = {
"traefik": {
"installed_via": "file",
"file_path": "/root/.traefik",
"file_regex": VERSION_RE,
},
"another-app": {"installed_via": "manual"},
}
bundled = {
"traefik": {
"installed_via": "binary",
"binary_path": "/usr/bin/traefik",
"binary_args": ["version"],
"installed_regex": TRAEFIK_LIVE_RE,
}
}
overrides = {
"traefik": {
"operational": True,
"detector": dict(bundled["traefik"]),
}
}
response = mock.MagicMock()
response.read.return_value = json.dumps(remote).encode("utf-8")
response.__enter__.return_value = response
previous_cache = lxc_apps._tracking_hints_cache
previous_ts = lxc_apps._tracking_hints_ts
try:
with tempfile.TemporaryDirectory() as temp_dir:
with (
mock.patch.object(lxc_apps, "_TRACKING_HINTS_DISK", str(Path(temp_dir) / "hints.json")),
mock.patch.object(lxc_apps, "_load_bundled_hints", return_value=bundled),
mock.patch.object(lxc_apps, "_load_runtime_verified_overrides", return_value=overrides),
mock.patch.object(lxc_apps.urllib.request, "urlopen", return_value=response),
):
lxc_apps._tracking_hints_cache = None
lxc_apps._tracking_hints_ts = 0
result = lxc_apps._fetch_tracking_hints()
finally:
lxc_apps._tracking_hints_cache = previous_cache
lxc_apps._tracking_hints_ts = previous_ts
self.assertEqual(result["traefik"]["installed_via"], "binary")
self.assertEqual(result["traefik"]["binary_path"], "/usr/bin/traefik")
self.assertNotIn("file_path", result["traefik"])
def test_runtime_override_promotes_live_detector_and_preserves_helper_marker(self):
catalog = {
"traefik": {
"installed_via": "file",
"file_path": "/root/.traefik",
"file_regex": VERSION_RE,
}
}
v2 = {
"apps": {
"traefik": {
"detectors": [
{
"installed_via": "file",
"file_path": "/root/.traefik",
"file_regex": VERSION_RE,
}
]
}
}
}
overrides = {
"apps": {
"traefik": {
"operational": True,
"detector": {
"installed_via": "binary",
"binary_path": "/usr/bin/traefik",
"binary_args": ["version"],
"installed_regex": TRAEFIK_LIVE_RE,
"repo": "traefik/traefik",
"tag_regex": VERSION_RE,
},
}
}
}
with tempfile.TemporaryDirectory() as temp_dir:
path = Path(temp_dir) / "overrides.json"
path.write_text(__import__("json").dumps(overrides), encoding="utf-8")
result = app_tracking_generator.apply_runtime_overrides(catalog, v2, path)
self.assertEqual(result["promoted_to_v1"], ["traefik"])
self.assertEqual(catalog["traefik"]["installed_via"], "binary")
self.assertEqual(catalog["traefik"]["binary_path"], "/usr/bin/traefik")
self.assertEqual(
catalog["traefik"]["file_fallbacks"],
[{"path": "/root/.traefik", "regex": VERSION_RE, "source": "helper_marker"}],
)
def test_old_helper_marker_migrates_to_new_live_catalog_detector(self):
"""A catalog upgrade repairs already-saved helper sidecars too."""
app = {
"id": "app-traefik",
"name": "Traefik",
"helper_slug": "traefik",
"installed_via": "file",
"file_path": "/root/.traefik",
"file_regex": VERSION_RE,
"repo": "traefik/traefik",
"tag_regex": VERSION_RE,
"update_command": "custom updater owned by the operator",
"ports": [{"port": 8080}],
}
hint = {
"installed_via": "binary",
"binary_path": "/usr/bin/traefik",
"binary_args": ["version"],
"installed_regex": TRAEFIK_LIVE_RE,
"repo": "traefik/traefik",
"tag_regex": VERSION_RE,
"file_fallbacks": [{"path": "/root/.traefik", "regex": VERSION_RE}],
}
with (
mock.patch.object(lxc_apps, "_fetch_tracking_hints", return_value={"traefik": hint}),
mock.patch.object(
lxc_apps,
"_pct_exec",
return_value=(0, "Version: 3.7.12\n", ""),
) as run,
):
installed, error, healed = lxc_apps._detect_with_alt_healing(117, app)
self.assertEqual(installed, "3.7.12")
self.assertIsNone(error)
self.assertTrue(healed)
self.assertEqual(app["installed_via"], "binary")
self.assertEqual(app["binary_path"], "/usr/bin/traefik")
self.assertNotIn("file_path", app)
self.assertEqual(app["update_command"], "custom updater owned by the operator")
run.assert_called_once_with(117, ["/usr/bin/traefik", "version"])
def test_helper_marker_remains_safe_fallback_when_live_probe_is_missing(self):
app = {
"helper_slug": "example",
"installed_via": "file",
"file_path": "/root/.example",
"file_regex": VERSION_RE,
}
hint = {
"installed_via": "binary",
"binary_path": "/usr/bin/example",
"binary_args": ["version"],
"installed_regex": TRAEFIK_LIVE_RE,
"file_fallbacks": [{"path": "/root/.example", "regex": VERSION_RE}],
}
def probe(_vmid, argv, **_kwargs):
if argv[0] == "/usr/bin/example":
return 127, "", "not found"
self.assertEqual(argv, ["cat", "/root/.example"])
return 0, "1.2.3\n", ""
with (
mock.patch.object(lxc_apps, "_fetch_tracking_hints", return_value={"example": hint}),
mock.patch.object(lxc_apps, "_pct_exec", side_effect=probe),
):
installed, error, healed = lxc_apps._detect_with_alt_healing(117, app)
self.assertEqual(installed, "1.2.3")
self.assertIsNone(error)
self.assertFalse(healed)
self.assertEqual(app["installed_via"], "file")
def test_successful_custom_update_invalidates_only_its_manual_version_claim(self):
with tempfile.TemporaryDirectory() as temp_dir, mock.patch.object(
lxc_apps, "_APPS_DIR", temp_dir
):
sidecar = {
"vmid": 210,
"apps": [
{
"id": "manual-app",
"installed_via": "manual",
"installed_version": "0.9.0",
"state": {},
},
{
"id": "binary-app",
"installed_via": "binary",
"binary_path": "/usr/bin/example",
"state": {},
},
],
}
self.assertTrue(lxc_apps._write_sidecar(210, sidecar))
self.assertEqual(lxc_apps.mark_manual_versions_unverified(210, ["manual-app"]), 1)
persisted = lxc_apps._read_sidecar(210)
manual = persisted["apps"][0]
binary = persisted["apps"][1]
self.assertTrue(manual["manual_version_needs_confirmation"])
self.assertNotIn("manual_version_needs_confirmation", binary)
with mock.patch.object(lxc_apps, "_fetch_tracking_hints", return_value={}):
installed, error, healed = lxc_apps._detect_with_alt_healing(210, manual)
self.assertIsNone(installed)
self.assertIsNone(error)
self.assertFalse(healed)
if __name__ == "__main__":
unittest.main()
+13 -3
View File
@@ -4613,13 +4613,23 @@
"website": "https://github.com/javedh-dev/tracktor"
},
"traefik": {
"binary_args": [
"version"
],
"binary_path": "/usr/bin/traefik",
"default_ports": [
8080
],
"file_path": "/root/.traefik",
"file_regex": "(?i)(?:v|release[-_/]?)?(\\d+(?:\\.\\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)",
"file_fallbacks": [
{
"path": "/root/.traefik",
"regex": "(?i)(?:v|release[-_/]?)?(\\d+(?:\\.\\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)",
"source": "helper_marker"
}
],
"github_source": "releases",
"installed_via": "file",
"installed_regex": "(?m)^Version:\\s*(\\d+(?:\\.\\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)",
"installed_via": "binary",
"logo": "https://cdn.jsdelivr.net/gh/selfhst/icons@main/webp/traefik.webp",
"repo": "traefik/traefik",
"tag_regex": "(?i)(?:v|release[-_/]?)?(\\d+(?:\\.\\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)",
+22
View File
@@ -213,6 +213,28 @@
}
]
},
"traefik": {
"operational": true,
"name": "Traefik",
"install_scope": [
"community-script",
"manual-if-same-path"
],
"evidence": [
"community-confirmed: native Traefik binary reports its installed version"
],
"detector": {
"installed_via": "binary",
"binary_path": "/usr/bin/traefik",
"binary_args": [
"version"
],
"installed_regex": "(?m)^Version:\\s*(\\d+(?:\\.\\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)",
"repo": "traefik/traefik",
"github_source": "releases",
"tag_regex": "(?i)(?:v|release[-_/]?)?(\\d+(?:\\.\\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)"
}
},
"vaultwarden": {
"detector": {
"installed_via": "binary",