Files
ProxMenux/.github/scripts/generate_app_tracking_catalog.py
T
MacRimi 06f41f5792 feat(lxc-updates): revamp Updates + Options card + curated hints
Updates tab
- Unified OS + Application update card with per-section Apply buttons
  and a combined "Apply OS + <app>" footer button
- Helper-scripts install detection: uses helper_slug from managed_installs
  cache (hostname fuzzy-match against helpers_cache) so the button
  surfaces even when /usr/bin/update was removed
- Runs the community-scripts helper INSIDE the CT via pct exec so
  build.func picks the silent update path (PHS_SILENT=1) instead of the
  install menu — works with and without /usr/bin/update
- HELPER_SLUG env passthrough from backend to apply_updates.sh: falls
  back to constructing the ct/<slug>.sh URL when the CT no longer
  carries the marker file
- Post-apply state refresh via managed_installs.check_for_updates(force)
  in the /applied hook so the badge updates without a manual reload

Options card
- Rewrote as view / edit mode split with a single Edit button
- Unified apply defaults (snapshot + storage + restart) shared by
  manual and scheduled runs
- Scheduled updates (M5): cron picker + preset dropdown + What-to-update
  target + Delete schedule button, wired to a background scheduler thread
  that fires apply_updates.sh headless with the schedule's env vars
- External host cron detection with variant + scope reporting
  (tteck-legacy / community-scripts / custom, OS-only), shown as an
  informational chip only in edit mode

App tab editor
- Multi-app registration with per-app upstream tracking method
  (github / http_json / docker_hub)
- Card-contrast pattern in edit mode (bg-card + bg-background inputs)
- Auto-heal for missing installed_version via alt_detectors +
  file_fallbacks

Curated tracking hints (M6)
- Add http_json upstream for Plex (plex.tv API)
- Add binary+github hints for Emby (MediaBrowser/Emby.Releases) and
  PhotoPrism (photoprism/photoprism)
- Extend CI merge whitelist with upstream_type / upstream_url /
  upstream_json_path / docker_image

Tab reorder
- LXC modal tabs: Status | App | Updates | Mounts | Backups | Firewall

apply_updates.sh
- New helper execution path: parse ct/<slug>.sh URL, run inside CT
  with PHS_SILENT=1, respecting HELPER_SLUG fallback when
  /usr/bin/update is missing
2026-08-09 00:49:44 +02:00

1350 lines
54 KiB
Python

#!/usr/bin/env python3
"""Build a verified ProxMenux LXC application-version tracking catalog.
The source of truth is a pinned snapshot of community-scripts/ProxmoxVE,
downloaded through the GitHub REST API. Only ct/*.sh launchers are considered.
An operational hint is emitted only when two independent pieces of the helper
scripts agree:
* file: the LXC update script reads the cache written by the shared deploy
helper, and the matching install script deploys the same app/repository; or
* dpkg/apk: the package is installed by the install script and checked or
explicitly upgraded by the LXC update script.
Everything else is retained in the audit report rather than guessed into the
runtime JSON. The script uses only Python's standard library and runs on macOS.
"""
from __future__ import annotations
import argparse
import hashlib
import io
import json
import os
import re
import shutil
import sys
import tarfile
import tempfile
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Iterable
DEFAULT_REPOSITORY = "community-scripts/ProxmoxVE"
DEFAULT_REF = "main"
API_VERSION = "2022-11-28"
USER_AGENT = "ProxMenux-app-tracking-catalog/1.0"
# Both the helper cache and common GitHub tags are handled. The first capture
# group is deliberately the normalized version consumed by lxc_apps.py.
DEFAULT_VERSION_REGEX = (
r"(?i)(?:v|release[-_/]?)?"
r"(\d+(?:\.\d+){1,3}(?:[-+._][0-9A-Za-z.-]+)?)"
)
VERSION_FORMATS: list[tuple[re.Pattern[str], str]] = [
(re.compile(r'(?i)^v?\d{6,14}$'), r"(?i)v?(\d{6,14})"),
(
re.compile(r'(?i)^\d{6,14}-[0-9a-f]{6,40}$'),
r"(?i)(\d{6,14}(?:-[0-9a-f]{6,40})?)",
),
(re.compile(r'(?i)^r\d{4,}$'), r"(?i)r?(\d{4,})"),
(
re.compile(r'(?i)^SQUID_\d+(?:_\d+){1,3}$'),
r"(?i)(?:SQUID_)?(\d+(?:[._]\d+){1,3})",
),
(
re.compile(r'(?i)^release\.\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}Z(?:\.\d+)?$'),
r"(?i)(?:release\.)?(\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}Z(?:\.\d+)?)",
),
]
APP_RE = re.compile(r'^\s*APP=["\']([^"\']+)["\']', re.MULTILINE)
CHECK_RE = re.compile(
r'\bcheck_for_gh_release\s+["\']([^"\']+)["\']\s+["\']'
r'([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)["\']'
)
FETCH_RE = re.compile(
r'\bfetch_and_deploy_gh_release\s+["\']([^"\']+)["\']\s+["\']'
r'([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)["\']'
)
HEADER_GITHUB_RE = re.compile(
r'Github:\s*https?://github\.com/([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)',
re.IGNORECASE,
)
HEADER_SOURCE_RE = re.compile(r'^\s*#\s*Source:\s*(https?://\S+)', re.IGNORECASE | re.MULTILINE)
EXPLICIT_VERSION_FILE_RE = re.compile(
r'(?:>|tee\s+)(?:["\']?)'
r'(?:~|\$HOME|\$\{HOME\})/(\.[A-Za-z0-9_.-]+)'
)
DOCKER_IMAGE_RE = re.compile(
r'(?:docker\s+(?:run|pull)\b[\s\S]{0,800}?)'
r'((?:ghcr\.io|docker\.io|quay\.io|lscr\.io)/[A-Za-z0-9_./-]+:[A-Za-z0-9_.-]+)',
re.IGNORECASE,
)
DOCKER_NAME_RE = re.compile(r'\bdocker\s+run\b[\s\S]{0,1200}?--name(?:=|\s+)([A-Za-z0-9_.-]+)', re.IGNORECASE)
EXECSTART_RE = re.compile(r'^\s*ExecStart=(/[A-Za-z0-9_./+@:-]+)', re.MULTILINE)
EXISTENCE_PATH_RE = re.compile(r'\[\[?[^\n]{0,80}?!?\s+-[fx]\s+(/[A-Za-z0-9_./+@:-]+)')
PACKAGE_TOKEN_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9.+:@_-]*$')
class CatalogError(RuntimeError):
pass
class GitHubClient:
def __init__(self, token: str | None = None) -> None:
self.token = token or os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
self.rate_remaining: str | None = None
def request(self, url: str, *, accept: str = "application/vnd.github+json") -> bytes:
headers = {
"Accept": accept,
"X-GitHub-Api-Version": API_VERSION,
"User-Agent": USER_AGENT,
}
if self.token:
headers["Authorization"] = f"Bearer {self.token}"
req = urllib.request.Request(url, headers=headers)
try:
with urllib.request.urlopen(req, timeout=90) as response:
self.rate_remaining = response.headers.get("X-RateLimit-Remaining")
return response.read()
except urllib.error.HTTPError as exc:
remaining = exc.headers.get("X-RateLimit-Remaining")
if exc.code == 403 and remaining == "0":
raise CatalogError(
"GitHub API rate limit exhausted. Set GITHUB_TOKEN (or GH_TOKEN) "
"and run again."
) from exc
raise CatalogError(f"GitHub API HTTP {exc.code} for {url}") from exc
except urllib.error.URLError as exc:
raise CatalogError(f"Cannot reach GitHub API: {exc}") from exc
def json(self, path: str) -> Any:
url = path if path.startswith("https://") else f"https://api.github.com{path}"
return json.loads(self.request(url).decode("utf-8"))
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def safe_extract_tar(payload: bytes, destination: Path) -> None:
destination.mkdir(parents=True, exist_ok=True)
root = destination.resolve()
with tarfile.open(fileobj=io.BytesIO(payload), mode="r:gz") as archive:
members = archive.getmembers()
for member in members:
parts = Path(member.name).parts
if len(parts) < 2:
continue
relative = Path(*parts[1:])
target = (destination / relative).resolve()
if target != root and root not in target.parents:
raise CatalogError(f"Unsafe path in GitHub tarball: {member.name}")
if member.issym() or member.islnk() or member.isdev():
raise CatalogError(f"Unsupported link/device in GitHub tarball: {member.name}")
if not (member.isfile() or member.isdir()):
continue
member.name = str(relative)
if member.name != ".":
# Python 3.9 (the system Python on several macOS releases)
# predates tarfile's `filter=` argument. Paths and special
# members have already been validated above.
archive.extract(member, destination)
def obtain_snapshot(
client: GitHubClient,
repository: str,
ref: str,
cache_dir: Path,
) -> tuple[Path, str, str]:
commit = client.json(f"/repos/{repository}/commits/{ref}")
commit_sha = str(commit.get("sha") or "")
if not re.fullmatch(r"[0-9a-f]{40}", commit_sha):
raise CatalogError(f"Unexpected commit SHA for {repository}@{ref}")
snapshot_dir = cache_dir / repository.replace("/", "--") / commit_sha
marker = snapshot_dir / ".snapshot-complete"
if marker.exists() and (snapshot_dir / "ct").is_dir():
return snapshot_dir, commit_sha, marker.read_text(encoding="utf-8").strip()
payload = client.request(
f"https://api.github.com/repos/{repository}/tarball/{commit_sha}",
accept="application/vnd.github+json",
)
archive_sha = hashlib.sha256(payload).hexdigest()
temp_parent = snapshot_dir.parent
temp_parent.mkdir(parents=True, exist_ok=True)
temp_dir = Path(tempfile.mkdtemp(prefix=f"{commit_sha}.tmp-", dir=temp_parent))
try:
safe_extract_tar(payload, temp_dir)
(temp_dir / ".snapshot-complete").write_text(archive_sha + "\n", encoding="utf-8")
if snapshot_dir.exists():
shutil.rmtree(snapshot_dir)
temp_dir.rename(snapshot_dir)
except Exception:
shutil.rmtree(temp_dir, ignore_errors=True)
raise
return snapshot_dir, commit_sha, archive_sha
def normalize(value: str) -> str:
return re.sub(r"[^a-z0-9]", "", value.lower())
def cache_key(app: str) -> str:
# Mirrors tools.func: lower-case then `tr -d ' '`.
return app.lower().replace(" ", "")
def version_regex_for_tag(tag: str) -> str:
if not tag or re.search(DEFAULT_VERSION_REGEX, tag):
return DEFAULT_VERSION_REGEX
for matcher, pattern in VERSION_FORMATS:
if matcher.fullmatch(tag):
return pattern
return DEFAULT_VERSION_REGEX
def read_text(path: Path | None) -> str:
if path is None or not path.is_file():
return ""
return path.read_text(encoding="utf-8", errors="replace")
def marker_stores_url(text: str, marker: str) -> bool:
marker_pattern = re.escape(marker)
for line in text.splitlines():
if not re.search(rf'(?:~|\$HOME|\$\{{HOME\}})/{marker_pattern}\b', line):
continue
variable = re.search(r'echo\s+["\']?\$\{?([A-Za-z_][A-Za-z0-9_]*)', line)
if not variable:
continue
assignment = re.search(
rf'^\s*{re.escape(variable.group(1))}=([^\n]*(?:\n(?![A-Za-z_][A-Za-z0-9_]*=)[^\n]*){{0,3}})',
text,
re.MULTILINE,
)
if assignment and re.search(r"grep\s+-o[^\n]*https?://|DownloadLocation", assignment.group(0), re.IGNORECASE):
return True
return False
def unique_pairs(items: Iterable[tuple[str, str]]) -> list[tuple[str, str]]:
result: list[tuple[str, str]] = []
seen: set[tuple[str, str]] = set()
for item in items:
key = (item[0], item[1].lower())
if key not in seen:
seen.add(key)
result.append(item)
return result
def extract_command_blocks(text: str) -> list[str]:
lines = text.splitlines()
blocks: list[str] = []
i = 0
while i < len(lines):
line = lines[i].strip()
block = line
while block.rstrip().endswith("\\") and i + 1 < len(lines):
block = block.rstrip()[:-1] + " " + lines[i + 1].strip()
i += 1
blocks.append(block)
i += 1
return blocks
def packages_from_command(text: str, manager: str) -> set[str]:
packages: set[str] = set()
command_re = (
re.compile(r'\b(?:apt|apt-get)\b[^\n]*?\b(?:install|upgrade)\b\s+(.+)$')
if manager == "dpkg"
else re.compile(r'\bapk\b[^\n]*?\b(?:add|upgrade)\b\s+(.+)$')
)
for block in extract_command_blocks(text):
match = command_re.search(block)
if not match:
continue
for token in re.split(r"\s+", match.group(1)):
token = token.strip("'\"")
if (
not token
or token.startswith("-")
or token.startswith("$")
or "/" in token
or "=" in token
or not PACKAGE_TOKEN_RE.fullmatch(token)
):
continue
packages.add(token)
return packages
def checked_packages(text: str, manager: str) -> set[str]:
patterns = (
[r'\bdpkg\s+-s\s+([A-Za-z0-9.+:@_-]+)', r'\bdpkg-query\b[^\n]*?\s([A-Za-z0-9.+:@_-]+)\s*(?:[>&]|$)']
if manager == "dpkg"
else [r'\bapk\s+info\b[^\n]*?\s([A-Za-z0-9.+:@_-]+)\s*(?:[>&]|$)']
)
result: set[str] = set()
for pattern in patterns:
result.update(re.findall(pattern, text))
return result
def load_helper_catalog(path: Path | None) -> dict[str, dict[str, Any]]:
if path is None or not path.is_file():
return {}
raw = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(raw, list):
raise CatalogError("helpers_cache.json must contain a list")
result: dict[str, dict[str, Any]] = {}
for item in raw:
if not isinstance(item, dict) or item.get("type") != "lxc":
continue
slug = str(item.get("slug") or "")
if slug and slug not in result:
result[slug] = item
return result
@dataclass
class Candidate:
app: str
repo: str
score: int
reasons: list[str] = field(default_factory=list)
install_fetch: bool = False
update_fetch: bool = False
def select_release_candidate(
slug: str,
app_name: str,
launcher: str,
installer: str,
) -> tuple[Candidate | None, list[dict[str, Any]], str | None]:
checks = unique_pairs(CHECK_RE.findall(launcher))
update_fetches = {(a, r.lower()) for a, r in FETCH_RE.findall(launcher)}
install_fetches = {(a, r.lower()) for a, r in FETCH_RE.findall(installer)}
candidates: list[Candidate] = []
for app, repo in checks:
key = (app, repo.lower())
score = 0
identity_score = 0
reasons: list[str] = []
n_app, n_slug, n_name = normalize(app), normalize(slug), normalize(app_name)
n_repo = normalize(repo.split("/", 1)[1])
if n_app == n_slug:
score += 100
identity_score += 100
reasons.append("check app name matches LXC slug")
elif n_app and (n_app in n_slug or n_slug in n_app):
score += 55
identity_score += 55
reasons.append("check app name closely matches LXC slug")
if n_app == n_name:
score += 45
identity_score += 45
reasons.append("check app name matches APP label")
if n_repo == n_slug or (n_repo and (n_repo in n_slug or n_slug in n_repo)):
score += 30
identity_score += 30
reasons.append("repository name matches LXC slug")
install_fetch = key in install_fetches
update_fetch = key in update_fetches
if install_fetch:
score += 80
reasons.append("matching deploy call exists in install script")
if update_fetch:
score += 30
reasons.append("matching deploy call exists in update script")
if identity_score == 0:
# Auxiliary components (Ollama inside Open WebUI, web vault
# assets inside Vaultwarden, etc.) must never become the primary
# application merely because their deploy helper is present.
score -= 1000
reasons.append("does not identify the primary LXC application")
candidates.append(Candidate(app, repo, score, reasons, install_fetch, update_fetch))
candidates.sort(key=lambda candidate: candidate.score, reverse=True)
audit_candidates = [
{
"app": c.app,
"repo": c.repo,
"cache_file": f"/root/.{cache_key(c.app)}",
"score": c.score,
"install_fetch": c.install_fetch,
"update_fetch": c.update_fetch,
"reasons": c.reasons,
}
for c in candidates
]
if not candidates:
return None, audit_candidates, "no literal check_for_gh_release call"
winner = candidates[0]
if winner.score < 70:
return None, audit_candidates, "no release check identifies the primary LXC application"
if len(candidates) > 1 and winner.score == candidates[1].score:
return None, audit_candidates, "ambiguous primary GitHub application"
if not winner.install_fetch:
return None, audit_candidates, "version cache is not proven to exist immediately after installation"
return winner, audit_candidates, None
def select_install_only_release(
slug: str,
app_name: str,
installer: str,
header_repos: list[str],
) -> Candidate | None:
candidates: list[Candidate] = []
for app, repo in unique_pairs(FETCH_RE.findall(installer)):
n_app, n_slug, n_name = normalize(app), normalize(slug), normalize(app_name)
n_repo = normalize(repo.split("/", 1)[1])
score = 0
reasons: list[str] = []
if n_app == n_slug:
score += 100
reasons.append("deploy app matches LXC slug")
elif n_app and (n_app in n_slug or n_slug in n_app):
score += 50
reasons.append("deploy app closely matches LXC slug")
if n_app == n_name:
score += 45
reasons.append("deploy app matches APP label")
if n_repo == n_slug or (n_repo and (n_repo in n_slug or n_slug in n_repo)):
score += 30
reasons.append("repository name matches LXC slug")
if repo.lower() in {item.lower() for item in header_repos}:
score += 20
reasons.append("repository matches script header")
candidates.append(Candidate(app, repo, score, reasons, install_fetch=True))
candidates.sort(key=lambda candidate: candidate.score, reverse=True)
if not candidates or candidates[0].score < 70:
return None
if len(candidates) > 1 and candidates[0].score == candidates[1].score:
return None
return candidates[0]
def choose_package(
slug: str,
app_name: str,
launcher: str,
installer: str,
manager: str,
) -> tuple[str | None, dict[str, Any]]:
installed = packages_from_command(installer, manager)
updated = packages_from_command(launcher, manager)
checked = checked_packages(launcher, manager)
raw_proven = installed & (updated | checked)
# Some official repositories are updated with a plain `apt upgrade` or
# `apk upgrade`, so the package is not repeated in the update command.
# Accept only an exact app/slug match in that case; dependencies remain
# excluded.
has_generic_upgrade = bool(
re.search(r'\b(?:apt|apt-get)\b[^\n]*\bupgrade\b', launcher)
if manager == "dpkg"
else re.search(r'\bapk\b[^\n]*\bupgrade\b', launcher)
)
if has_generic_upgrade:
expected = {normalize(slug), normalize(app_name)}
raw_proven.update(package for package in installed if normalize(package) in expected)
expected = {normalize(slug), normalize(app_name)}
def is_app_package(package: str) -> bool:
normalized = normalize(package)
return any(
candidate and (
normalized == candidate
or normalized in candidate
or candidate in normalized
)
for candidate in expected
)
proven = {package for package in raw_proven if is_app_package(package)}
evidence = {
"installed_packages": sorted(installed),
"updated_packages": sorted(updated),
"checked_packages": sorted(checked),
"proven_packages": sorted(proven),
"rejected_unrelated_packages": sorted(raw_proven - proven),
"generic_upgrade": has_generic_upgrade,
}
if not proven:
return None, evidence
def score(package: str) -> tuple[int, int, str]:
n_pkg = normalize(package)
n_slug = normalize(slug)
n_name = normalize(app_name)
value = 0
if n_pkg == n_slug:
value += 100
elif n_pkg in n_slug or n_slug in n_pkg:
value += 45
if n_pkg == n_name:
value += 50
if package in checked:
value += 30
return value, -len(package), package
ranked = sorted(proven, key=score, reverse=True)
if len(ranked) > 1 and score(ranked[0])[:2] == score(ranked[1])[:2]:
return None, evidence
return ranked[0], evidence
def repo_from_helper(item: dict[str, Any]) -> str:
repo = str(item.get("github_repo") or "").strip()
if re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repo):
return repo
raw = str(item.get("github") or "").strip()
match = re.search(r'(?:github\.com/)?([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)', raw)
return match.group(1) if match else ""
def build_catalog(
source: Path,
helpers: dict[str, dict[str, Any]],
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
catalog: dict[str, Any] = {}
v2_apps: dict[str, Any] = {}
records: list[dict[str, Any]] = []
launchers = sorted((source / "ct").glob("*.sh"))
tools_func = source / "misc" / "tools.func"
tools_text = read_text(tools_func)
shared_helper_verified = bool(
re.search(r'local version_file="\$HOME/\.\$\{app_lc\}"', tools_text)
and re.search(r'echo "\$version" >"\$version_file"', tools_text)
and re.search(r'local current_file="\$HOME/\.\$\{app_lc\}"', tools_text)
)
if not shared_helper_verified:
raise CatalogError("Could not verify the shared GitHub release cache contract in misc/tools.func")
for launcher_path in launchers:
slug = launcher_path.stem
launcher = read_text(launcher_path)
installer_path = source / "install" / f"{slug}-install.sh"
installer = read_text(installer_path)
app_match = APP_RE.search(launcher)
app_name = app_match.group(1) if app_match else slug
helper_item = helpers.get(slug, {})
helper_repo = repo_from_helper(helper_item)
helper_version = str(helper_item.get("github_version") or "").strip()
version_regex = version_regex_for_tag(helper_version)
header_repos = HEADER_GITHUB_RE.findall(launcher + "\n" + installer)
official_sources = sorted(set(HEADER_SOURCE_RE.findall(launcher + "\n" + installer)))
docker_images = sorted(set(DOCKER_IMAGE_RE.findall(installer)))
docker_names = sorted(set(DOCKER_NAME_RE.findall(installer)))
launcher_version_files = set(EXPLICIT_VERSION_FILE_RE.findall(launcher))
installer_version_files = set(EXPLICIT_VERSION_FILE_RE.findall(installer))
explicit_files = sorted(launcher_version_files | installer_version_files)
relevant_markers = sorted(
marker
for marker in launcher_version_files & installer_version_files
if normalize(marker.lstrip(".")) in {normalize(slug), normalize(app_name)}
and not marker_stores_url(launcher, marker)
and not marker_stores_url(installer, marker)
)
package_evidence: dict[str, Any] = {}
package_detectors: list[dict[str, Any]] = []
for manager in ("dpkg", "apk"):
package, evidence = choose_package(slug, app_name, launcher, installer, manager)
package_evidence[manager] = evidence
if package and helper_repo:
upstream_verified = bool(
helper_version and re.search(version_regex, helper_version)
)
package_detectors.append(
{
"installed_via": manager,
"package": package,
"repo": helper_repo,
"github_source": "releases",
"tag_regex": version_regex,
"install_scope": ["community-script", "manual-if-same-package"],
"verification": (
"verified-static"
if upstream_verified
else "candidate-needs-upstream-verification"
),
"evidence": {
"install": f"install/{installer_path.name}: package installation",
"update": f"ct/{launcher_path.name}: package check or named upgrade",
},
}
)
binary_paths = sorted(set(EXECSTART_RE.findall(installer)) & set(EXISTENCE_PATH_RE.findall(launcher)))
binary_detectors = [
{
"installed_via": "binary",
"binary_path": path,
"binary_args": ["--version"],
"repo": helper_repo or None,
"github_source": "releases",
"tag_regex": version_regex,
"install_scope": ["community-script", "manual-if-same-path"],
"verification": "candidate-needs-version-probe",
"evidence": {
"install": f"install/{installer_path.name}: systemd ExecStart",
"update": f"ct/{launcher_path.name}: installation existence check",
},
}
for path in binary_paths
]
docker_detectors: list[dict[str, Any]] = []
if docker_names or docker_images:
# Keep all discovered data when a script has multiple containers;
# pairing by shell position is intentionally left to an override.
docker_detectors.append(
{
"installed_via": "docker",
"container_names": docker_names,
"images": docker_images,
"version_sources": [
{"type": "oci_label", "name": "org.opencontainers.image.version"},
{"type": "image_ref_tag"},
],
"repo": helper_repo or None,
"github_source": "releases",
"tag_regex": version_regex,
"install_scope": ["community-script", "manual-docker"],
"verification": "requires-detector-change",
"evidence": {"install": f"install/{installer_path.name}: docker run/pull"},
}
)
record: dict[str, Any] = {
"slug": slug,
"name": app_name,
"launcher": f"ct/{launcher_path.name}",
"installer": f"install/{installer_path.name}" if installer_path.is_file() else None,
"status": "excluded",
"method": None,
"reason": None,
"helper_repo": helper_repo or None,
"helper_upstream_version": helper_version or None,
"header_repositories": header_repos,
"official_sources": official_sources,
"docker_images": docker_images,
"docker_names": docker_names,
"explicit_version_files": [f"/root/{item}" for item in explicit_files],
}
release, release_candidates, release_error = select_release_candidate(
slug, app_name, launcher, installer
)
install_only_release = select_install_only_release(
slug, app_name, installer, header_repos
) if release is None else None
record["release_candidates"] = release_candidates
v2_detectors: list[dict[str, Any]] = []
if release is not None:
hint = {
"installed_via": "file",
"file_path": f"/root/.{cache_key(release.app)}",
"file_regex": DEFAULT_VERSION_REGEX,
"repo": release.repo,
"github_source": "releases",
"tag_regex": version_regex,
}
hint["file_regex"] = version_regex
catalog[slug] = hint
v2_detectors.append(
{
**hint,
"install_scope": ["community-script"],
"verification": "verified-static",
"evidence": {
"install": f"install/{installer_path.name}: matching deploy helper",
"update": f"ct/{launcher_path.name}: matching release check",
"contract": "misc/tools.func: shared version-cache contract",
},
}
)
v2_detectors.extend(package_detectors)
v2_detectors.extend(binary_detectors)
v2_detectors.extend(docker_detectors)
record.update(
{
"status": "verified",
"method": "file",
"reason": "install deploy and update check share the same helper cache",
"selected": hint,
"evidence": {
"install": f"install/{installer_path.name}: fetch_and_deploy_gh_release({release.app}, {release.repo})",
"update": f"ct/{launcher_path.name}: check_for_gh_release({release.app}, {release.repo})",
"contract": "misc/tools.func writes and reads /root/.<normalized-app>",
},
}
)
records.append(record)
v2_apps[slug] = {
"name": app_name,
"repo": release.repo,
"official_sources": official_sources,
"detectors": v2_detectors,
}
continue
if len(relevant_markers) == 1 and helper_repo and helper_version:
marker = relevant_markers[0]
hint = {
"installed_via": "file",
"file_path": f"/root/{marker}",
"file_regex": version_regex,
"repo": helper_repo,
"github_source": "releases",
"tag_regex": version_regex,
}
catalog[slug] = hint
v2_detectors.append(
{
**hint,
"install_scope": ["community-script"],
"verification": "verified-static",
"evidence": {
"install": f"install/{installer_path.name}: writes {hint['file_path']}",
"update": f"ct/{launcher_path.name}: writes {hint['file_path']}",
},
}
)
v2_detectors.extend(package_detectors)
v2_detectors.extend(binary_detectors)
v2_detectors.extend(docker_detectors)
record.update(
{
"status": "verified",
"method": "file",
"reason": "install and update scripts write the same app-specific version marker",
"selected": hint,
"evidence": v2_detectors[0]["evidence"],
}
)
records.append(record)
v2_apps[slug] = {
"name": app_name,
"repo": helper_repo,
"official_sources": official_sources,
"detectors": v2_detectors,
}
continue
if install_only_release is not None:
v2_detectors.append(
{
"installed_via": "file",
"file_path": f"/root/.{cache_key(install_only_release.app)}",
"file_regex": version_regex,
"repo": install_only_release.repo,
"github_source": "releases",
"tag_regex": version_regex,
"install_scope": ["community-script"],
"verification": "candidate-install-cache-may-stale",
"evidence": {
"install": f"install/{installer_path.name}: deploy helper writes the version cache",
"limitation": "no matching update check proves that later updates refresh this cache",
},
}
)
package_selected = False
for detector in package_detectors:
manager = detector["installed_via"]
if not package_selected and detector["verification"] == "verified-static":
hint = {key: value for key, value in detector.items() if key in {
"installed_via", "package", "repo", "github_source", "tag_regex"
}}
catalog[slug] = hint
record.update(
{
"status": "verified",
"method": manager,
"reason": "package is present in both install and update/check paths",
"selected": hint,
"evidence": package_evidence[manager],
}
)
package_selected = True
record["package_evidence"] = package_evidence
if package_selected:
v2_detectors.extend(package_detectors)
v2_detectors.extend(binary_detectors)
v2_detectors.extend(docker_detectors)
records.append(record)
v2_apps[slug] = {
"name": app_name,
"repo": helper_repo or None,
"official_sources": official_sources,
"detectors": v2_detectors,
}
continue
if docker_images or re.search(r'\bsetup_docker\b|\bdocker\s+(?:run|compose|pull)\b', installer):
record["reason"] = "Docker installation requires detector support not present in lxc_apps.py"
record["required_detector"] = "docker"
elif release_error:
record["reason"] = release_error
elif not installer_path.is_file():
record["reason"] = "no matching install script"
elif helper_repo and any(
item["proven_packages"] for item in package_evidence.values()
):
record["reason"] = "package detector needs upstream repository/release verification"
elif not helper_repo:
record["reason"] = "no verified upstream GitHub repository for package tracking"
else:
record["reason"] = "no supported detection method could be proven from both scripts"
records.append(record)
v2_detectors.extend(package_detectors)
v2_detectors.extend(binary_detectors)
v2_detectors.extend(docker_detectors)
v2_apps[slug] = {
"name": app_name,
"repo": helper_repo or (header_repos[0] if header_repos else None),
"official_sources": official_sources,
"detectors": v2_detectors,
}
status_counts: dict[str, int] = {}
method_counts: dict[str, int] = {}
reason_counts: dict[str, int] = {}
tag_validation = {"matched": 0, "missing": 0, "mismatched": []}
for record in records:
status_counts[record["status"]] = status_counts.get(record["status"], 0) + 1
method = record.get("method") or "none"
method_counts[method] = method_counts.get(method, 0) + 1
reason = record.get("reason") or "none"
reason_counts[reason] = reason_counts.get(reason, 0) + 1
if record.get("status") == "verified":
upstream = record.get("helper_upstream_version")
pattern = (record.get("selected") or {}).get("tag_regex")
if not upstream:
tag_validation["missing"] += 1
elif pattern and re.search(pattern, upstream):
tag_validation["matched"] += 1
else:
tag_validation["mismatched"].append(
{"slug": record["slug"], "version": upstream, "tag_regex": pattern}
)
audit = {
"summary": {
"lxc_launchers": len(launchers),
"operational_hints": len(catalog),
"coverage_percent": round((len(catalog) / len(launchers) * 100), 2) if launchers else 0,
"status_counts": status_counts,
"method_counts": method_counts,
"reason_counts": reason_counts,
"shared_release_cache_contract_verified": shared_helper_verified,
"helper_upstream_tag_validation": tag_validation,
},
"records": records,
}
v2 = {
"schema_version": 2,
"detector_policy": {
"strategy": "try detectors in order and retain the first successful detector",
"operational_verification": ["verified-static", "verified-runtime"],
"non_operational_verification": [
"candidate-needs-version-probe",
"requires-detector-change",
"candidate-needs-runtime-validation",
"candidate-install-cache-may-stale",
"candidate-needs-upstream-verification",
"candidate-helper-marker",
],
},
"apps": dict(sorted(v2_apps.items())),
}
return dict(sorted(catalog.items())), audit, v2
def compare_existing(generated: dict[str, Any], existing_path: Path | None) -> dict[str, Any]:
if existing_path is None or not existing_path.is_file():
return {"existing_file": None, "added": sorted(generated), "removed": [], "changed": []}
existing = json.loads(existing_path.read_text(encoding="utf-8"))
if not isinstance(existing, dict):
raise CatalogError("Existing catalog must be a JSON object")
return {
"existing_file": str(existing_path),
"added": sorted(set(generated) - set(existing)),
"removed": sorted(set(existing) - set(generated)),
"changed": sorted(
slug for slug in set(existing) & set(generated) if existing[slug] != generated[slug]
),
"unchanged": sorted(
slug for slug in set(existing) & set(generated) if existing[slug] == generated[slug]
),
}
def demote_generic_helper_markers(
catalog: dict[str, Any],
v2: dict[str, Any],
audit: dict[str, Any],
) -> list[str]:
"""Remove generic /root/.app caches from the operational catalog.
Even when install and update scripts both write the marker, it records
helper/update state rather than interrogating the installed application.
Runtime checks also found these files absent on legacy and manually
updated LXC. They remain useful candidates/fallbacks in v2, not verified
primary detectors.
"""
demoted: list[str] = []
apps = v2.get("apps", {})
for record in audit.get("records", []):
slug = record.get("slug")
hint = catalog.get(slug)
if not isinstance(hint, dict):
continue
path = hint.get("file_path")
if hint.get("installed_via") != "file" or not isinstance(path, str):
continue
if not re.fullmatch(r"/root/\.[A-Za-z0-9_.-]+", path):
continue
catalog.pop(slug, None)
record["status"] = "candidate"
record["reason"] = "generic helper marker is not guaranteed on legacy/manual installations"
for detector in (apps.get(slug) or {}).get("detectors", []):
if detector.get("installed_via") == "file" and detector.get("file_path") == path:
detector["verification"] = "candidate-helper-marker"
detector["limitation"] = (
"Observed absent on legacy/manual LXC; use only as fallback or after runtime probe"
)
demoted.append(slug)
return sorted(demoted)
def verify_upstream_releases(
client: GitHubClient,
catalog: dict[str, Any],
cache_dir: Path,
) -> dict[str, Any]:
"""Verify repositories and current tags directly with the GitHub API."""
repos = sorted({hint["repo"] for hint in catalog.values() if hint.get("repo")})
if len(repos) > 40 and not client.token:
raise CatalogError(
f"--verify-upstream needs GITHUB_TOKEN or GH_TOKEN for {len(repos)} repositories "
"(the anonymous GitHub API limit is only 60 requests/hour)."
)
cache_file = cache_dir / "upstream-releases.json"
cache_file.parent.mkdir(parents=True, exist_ok=True)
try:
cache = json.loads(cache_file.read_text(encoding="utf-8"))
if not isinstance(cache, dict):
cache = {}
except (OSError, json.JSONDecodeError):
cache = {}
now = int(time.time())
results: dict[str, Any] = {}
for index, repo in enumerate(repos, start=1):
cached = cache.get(repo, {})
if isinstance(cached, dict) and now - int(cached.get("fetched_at", 0)) < 24 * 3600:
results[repo] = cached
continue
tag = ""
source = "releases"
error = ""
try:
payload = client.json(f"/repos/{repo}/releases/latest")
if isinstance(payload, dict):
tag = str(payload.get("tag_name") or payload.get("name") or "").strip()
except CatalogError as exc:
error = str(exc)
try:
tags = client.json(f"/repos/{repo}/tags?per_page=30")
if isinstance(tags, list) and tags and isinstance(tags[0], dict):
tag = str(tags[0].get("name") or "").strip()
source = "tags"
error = ""
except CatalogError as tag_exc:
error = f"release: {exc}; tags: {tag_exc}"
results[repo] = {
"tag": tag or None,
"source": source,
"error": error or None,
"fetched_at": now,
}
if index % 25 == 0:
print(f"Verified upstream repositories: {index}/{len(repos)}", file=sys.stderr)
cache_file.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8")
matched: list[dict[str, str]] = []
mismatched: list[dict[str, str]] = []
unavailable: list[dict[str, str]] = []
for slug, hint in catalog.items():
result = results.get(hint.get("repo"), {})
tag = result.get("tag")
if not tag:
unavailable.append({"slug": slug, "repo": hint.get("repo", ""), "error": result.get("error") or "no tag"})
elif re.search(hint["tag_regex"], tag):
matched.append({"slug": slug, "repo": hint["repo"], "tag": tag})
else:
mismatched.append(
{"slug": slug, "repo": hint["repo"], "tag": tag, "tag_regex": hint["tag_regex"]}
)
return {
"repositories_queried": len(repos),
"matched": len(matched),
"mismatched": mismatched,
"unavailable": unavailable,
"results": results,
}
def merge_existing_as_runtime_candidates(v2: dict[str, Any], existing_path: Path | None) -> dict[str, Any]:
"""Retain hand-curated/manual-install hints without declaring them proven.
Existing entries are valuable for official/manual layouts, but static
analysis found that several no longer match current Community Scripts.
They therefore enter v2 as runtime-validation candidates and never enter
the compatible v1 output automatically.
"""
result = {"merged": [], "unmatched": [], "skipped_duplicates": []}
if existing_path is None or not existing_path.is_file():
return result
raw = json.loads(existing_path.read_text(encoding="utf-8"))
if not isinstance(raw, dict):
return result
apps = v2.get("apps", {})
allowed = {
"installed_via", "package", "file_path", "file_regex", "binary_path",
"repo", "github_source", "tag_regex",
}
for slug, hint in raw.items():
if slug not in apps or not isinstance(hint, dict):
result["unmatched"].append(slug)
continue
detector = {key: value for key, value in hint.items() if key in allowed}
if not detector.get("installed_via"):
continue
signature = json.dumps(detector, sort_keys=True)
existing_signatures = {
json.dumps({key: value for key, value in item.items() if key in allowed}, sort_keys=True)
for item in apps[slug]["detectors"]
}
if signature in existing_signatures:
result["skipped_duplicates"].append(slug)
continue
detector.update(
{
"install_scope": ["manual", "legacy-catalog"],
"verification": "candidate-needs-runtime-validation",
"evidence": {"catalog": str(existing_path)},
}
)
apps[slug]["detectors"].append(detector)
result["merged"].append(slug)
for key in result:
result[key].sort()
return result
def enrich_catalog_metadata(
catalog: dict[str, Any],
v2: dict[str, Any],
helpers: dict[str, dict[str, Any]],
existing_path: Path | None,
) -> dict[str, int]:
"""Add presentation metadata without weakening detector verification.
Community Scripts provides one primary port and a curated logo. Existing
manual `default_ports` take precedence because they may describe multi-port
applications. Detector fields and their evidence remain untouched.
"""
existing: dict[str, Any] = {}
if existing_path and existing_path.is_file():
try:
payload = json.loads(existing_path.read_text(encoding="utf-8"))
if isinstance(payload, dict):
existing = payload
except (OSError, json.JSONDecodeError):
pass
counts = {"apps_with_ports": 0, "apps_with_logos": 0, "selfhst_logos": 0}
apps = v2.get("apps", {})
for slug, app in apps.items():
helper = helpers.get(slug, {})
prior = existing.get(slug, {}) if isinstance(existing.get(slug), dict) else {}
ports: list[int] = []
prior_ports = prior.get("default_ports")
if isinstance(prior_ports, list):
for value in prior_ports:
try:
port = int(value)
except (TypeError, ValueError):
continue
if 1 <= port <= 65535 and port not in ports:
ports.append(port)
if not ports:
raw_port = helper.get("port")
if isinstance(raw_port, int) and 1 <= raw_port <= 65535:
ports.append(raw_port)
logo = str(prior.get("logo") or helper.get("logo") or "").strip()
if logo and not re.match(r"^https://[A-Za-z0-9.-]+/", logo):
logo = ""
website = str(helper.get("website") or "").strip()
if ports:
app["default_ports"] = ports
counts["apps_with_ports"] += 1
if logo:
app["logo"] = logo
app["logo_source"] = (
"selfh.st/icons via jsDelivr"
if "cdn.jsdelivr.net/gh/selfhst/icons@" in logo
else "community-scripts catalog"
)
counts["apps_with_logos"] += 1
if app["logo_source"].startswith("selfh.st"):
counts["selfhst_logos"] += 1
if website.startswith("https://"):
app["website"] = website
# v1 only contains operationally verified apps. Extra metadata is
# ignored safely by validate_config but is available to suggestions/UI.
if slug in catalog:
if ports:
catalog[slug]["default_ports"] = ports
if logo:
catalog[slug]["logo"] = logo
if website.startswith("https://"):
catalog[slug]["website"] = website
return counts
def apply_runtime_overrides(
catalog: dict[str, Any],
v2: dict[str, Any],
overrides_path: Path | None,
) -> dict[str, Any]:
"""Apply detectors proven against real containers.
The generated/static catalog is intentionally conservative. This optional
overlay promotes only detectors carrying runtime evidence. Unsupported
future methods (for example ``python_dist`` or ``docker_label``) are kept
in v2 but are not written to the current-compatible v1 catalog.
"""
result: dict[str, Any] = {
"file": str(overrides_path) if overrides_path else None,
"promoted_to_v1": [],
"v2_only": [],
"invalid": [],
}
if overrides_path is None or not overrides_path.is_file():
return result
raw = json.loads(overrides_path.read_text(encoding="utf-8"))
apps_raw = raw.get("apps") if isinstance(raw, dict) else None
if not isinstance(apps_raw, dict):
raise CatalogError("runtime overrides must contain an 'apps' object")
supported_v1 = {"dpkg", "apk", "file", "binary"}
v2_apps = v2.get("apps", {})
detector_keys = {
"installed_via", "package", "file_path", "file_regex",
"binary_path", "binary_args", "python_path", "distribution",
"container_name", "label", "repo", "github_source", "tag_regex",
"installed_regex",
}
passthrough_keys = {
"file_fallbacks", "alt_detectors", "default_ports", "logo", "website",
}
for slug, spec in apps_raw.items():
if not isinstance(spec, dict) or not isinstance(spec.get("detector"), dict):
result["invalid"].append(slug)
continue
detector = {k: v for k, v in spec["detector"].items() if k in detector_keys}
method = detector.get("installed_via")
if not isinstance(method, str) or not method:
result["invalid"].append(slug)
continue
evidence = spec.get("evidence") if isinstance(spec.get("evidence"), list) else []
v2_detector = {
**detector,
"install_scope": spec.get("install_scope") or ["runtime-observed"],
"verification": "verified-runtime",
"evidence": evidence,
}
app = v2_apps.get(slug)
if not isinstance(app, dict):
result["invalid"].append(slug)
continue
app.setdefault("detectors", []).insert(0, v2_detector)
app["runtime_evidence"] = evidence
if bool(spec.get("remove_from_v1")):
catalog.pop(slug, None)
operational = bool(spec.get("operational", True))
if operational and method in supported_v1:
# Preserve presentation metadata already enriched from helpers.
presentation_source = dict(app)
presentation_source.update(catalog.get(slug, {}))
presentation = {
key: value
for key, value in presentation_source.items()
if key in {"default_ports", "logo", "website"}
}
hint = {k: v for k, v in detector.items() if k not in {
"binary_args", "python_path", "distribution", "container_name",
"label", "installed_regex",
}}
for key in passthrough_keys:
if key in spec:
hint[key] = spec[key]
hint.update(presentation)
catalog[slug] = hint
result["promoted_to_v1"].append(slug)
else:
result["v2_only"].append(slug)
for key in ("promoted_to_v1", "v2_only", "invalid"):
result[key].sort()
return result
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--repository", default=DEFAULT_REPOSITORY)
parser.add_argument("--ref", default=DEFAULT_REF)
parser.add_argument(
"--source-dir",
type=Path,
help="Analyze an existing checkout/snapshot instead of downloading through GitHub API",
)
parser.add_argument(
"--cache-dir",
type=Path,
default=Path.home() / ".cache" / "proxmenux-app-tracking",
)
parser.add_argument("--helpers-cache", type=Path)
parser.add_argument("--existing", type=Path)
parser.add_argument(
"--runtime-overrides",
type=Path,
help="JSON overlay with detectors verified against real LXC installations",
)
parser.add_argument(
"--include-helper-markers",
action="store_true",
help="Keep generic /root/.app helper caches in v1 (not recommended for legacy/manual LXC)",
)
parser.add_argument("--output", type=Path, default=Path("app_tracking_hints.generated.json"))
parser.add_argument("--audit-output", type=Path, default=Path("app_tracking_hints.audit.json"))
parser.add_argument("--v2-output", type=Path, default=Path("app_tracking_catalog.v2.json"))
parser.add_argument(
"--verify-upstream",
action="store_true",
help="Verify every repository's current release/tag directly through GitHub API (token recommended)",
)
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
started = time.time()
client = GitHubClient()
commit_sha = "local-source"
archive_sha = "not-applicable"
try:
if args.source_dir:
source = args.source_dir.resolve()
if re.fullmatch(r"[0-9a-fA-F]{40}", source.name):
commit_sha = source.name.lower()
else:
source, commit_sha, archive_sha = obtain_snapshot(
client, args.repository, args.ref, args.cache_dir.expanduser().resolve()
)
if not (source / "ct").is_dir() or not (source / "misc" / "tools.func").is_file():
raise CatalogError(f"Not a valid ProxmoxVE source tree: {source}")
helpers = load_helper_catalog(args.helpers_cache)
catalog, audit, v2 = build_catalog(source, helpers)
audit["demoted_helper_markers"] = (
[] if args.include_helper_markers else demote_generic_helper_markers(catalog, v2, audit)
)
audit["metadata"] = enrich_catalog_metadata(catalog, v2, helpers, args.existing)
audit["v2_existing_candidates"] = merge_existing_as_runtime_candidates(v2, args.existing)
audit["runtime_overrides"] = apply_runtime_overrides(
catalog, v2, args.runtime_overrides
)
if args.verify_upstream:
audit["github_upstream_verification"] = verify_upstream_releases(
client, catalog, args.cache_dir.expanduser().resolve()
)
audit["provenance"] = {
"repository": args.repository,
"ref": args.ref,
"commit_sha": commit_sha,
"archive_sha256": archive_sha,
"source_dir": str(source),
"generated_at_unix": int(time.time()),
"generator_sha256": sha256_file(Path(__file__).resolve()),
"github_api_rate_remaining": client.rate_remaining,
"helpers_cache": str(args.helpers_cache) if args.helpers_cache else None,
}
audit["existing_comparison"] = compare_existing(catalog, args.existing)
audit["summary"]["operational_hints"] = len(catalog)
audit["summary"]["coverage_percent"] = round(
len(catalog) / max(1, audit["summary"]["lxc_launchers"]) * 100, 2
)
method_counts: dict[str, int] = {}
for hint in catalog.values():
method = str(hint.get("installed_via") or "none")
method_counts[method] = method_counts.get(method, 0) + 1
method_counts["none"] = max(
0, audit["summary"]["lxc_launchers"] - len(catalog)
)
audit["summary"]["method_counts"] = method_counts
audit["summary"]["elapsed_seconds"] = round(time.time() - started, 3)
args.output.parent.mkdir(parents=True, exist_ok=True)
args.audit_output.parent.mkdir(parents=True, exist_ok=True)
args.v2_output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(json.dumps(catalog, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
args.audit_output.write_text(json.dumps(audit, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
v2["provenance"] = audit["provenance"]
args.v2_output.write_text(json.dumps(v2, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
except (CatalogError, OSError, json.JSONDecodeError) as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
summary = audit["summary"]
print(f"Pinned source: {args.repository}@{commit_sha}")
print(f"LXC launchers analyzed: {summary['lxc_launchers']}")
print(f"Verified operational hints: {summary['operational_hints']} ({summary['coverage_percent']}%)")
print(f"Methods: {summary['method_counts']}")
print(f"Catalog: {args.output.resolve()}")
print(f"Audit: {args.audit_output.resolve()}")
print(f"Multi-detector catalog v2: {args.v2_output.resolve()}")
return 0
if __name__ == "__main__":
raise SystemExit(main())