mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-09-14 18:56:52 +00:00
MacRimi
0251f77331
- Generate and ship a verified 389-app tracking catalog with 23 runtime overrides, fallback detectors, ports, logos, and Docker Hub tag previews. - Support modern Proxmox VE Helper-Scripts markers, historical installations, and official or manual app deployments. - Rework the LXC App and Updates tabs with cached suggestions, explicit discovery, version tracking, web links, custom updaters, and complete i18n. - Add independent OS, app, Docker Engine, Docker image, bulk, and scheduled update targets. - Add digest-based Docker inventory, Compose dependency grouping, safe standalone-container recreation with rollback, and package-scoped Docker Engine updates. - Refresh per-LXC caches after lifecycle and update tasks, then emit idempotent notifications based on the verified final state. - Harden Coral USB recovery by removing orphaned gasket DKMS registrations and validating that dpkg is healthy before reporting success.
1424 lines
58 KiB
Python
1424 lines
58 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
|
|
)
|
|
# Some current helpers do not install through
|
|
# fetch_and_deploy_gh_release, but explicitly write the release to
|
|
# the same marker used by check_for_gh_release. Actual Budget is a
|
|
# representative example (npm install + ~/.actualbudget). Recover
|
|
# the repository from the release check only when that marker maps
|
|
# to one unique candidate; never guess it from a generic header.
|
|
marker_release_repo: str | None = None
|
|
if len(relevant_markers) == 1:
|
|
marker_path = f"/root/{relevant_markers[0]}".lower()
|
|
marker_repos = {
|
|
str(candidate.get("repo") or "")
|
|
for candidate in release_candidates
|
|
if str(candidate.get("cache_file") or "").lower() == marker_path
|
|
and candidate.get("repo")
|
|
}
|
|
if len(marker_repos) == 1:
|
|
marker_release_repo = marker_repos.pop()
|
|
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 marker_release_repo:
|
|
marker = relevant_markers[0]
|
|
hint = {
|
|
"installed_via": "file",
|
|
"file_path": f"/root/{marker}",
|
|
"file_regex": version_regex,
|
|
"repo": marker_release_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": marker_release_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 modern helper markers when compatibility-only output is wanted.
|
|
|
|
Current community-scripts installers maintain ``/root/.<app>`` as their
|
|
version contract, so these are valid detectors for modern helper-owned
|
|
containers. They are not universal: legacy helpers and official/manual
|
|
installs may not have them. The default conservative mode therefore keeps
|
|
them in v2; production generation can opt in with
|
|
``--include-helper-markers`` and the runtime reports their distinct source.
|
|
"""
|
|
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"] = "modern 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"] = (
|
|
"Valid for modern helper installs; absent on legacy/manual LXC"
|
|
)
|
|
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. The compatible
|
|
catalog now supports every detector implemented by ``lxc_apps.py``;
|
|
retaining an older dpkg/file/binary-only allow-list silently discarded
|
|
proven Python and Docker detectors.
|
|
"""
|
|
result: dict[str, Any] = {
|
|
"file": str(overrides_path) if overrides_path else None,
|
|
"promoted_to_v1": [],
|
|
"v2_only": [],
|
|
"runtime_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", "python_dist",
|
|
"docker_label", "docker_exec", "command", "manual",
|
|
}
|
|
v2_apps = v2.get("apps", {})
|
|
detector_keys = {
|
|
"installed_via", "package", "file_path", "file_regex",
|
|
"binary_path", "binary_args", "python_path", "distribution",
|
|
"container_name", "label", "command_argv", "installed_version",
|
|
"repo", "github_source", "tag_regex", "installed_regex",
|
|
"upstream_type", "upstream_url", "upstream_json_path", "docker_image",
|
|
}
|
|
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):
|
|
# Official/manual and nested Docker applications do not
|
|
# necessarily have a community-scripts ct/<slug>.sh launcher.
|
|
# A runtime-proven override is sufficient to create their v2
|
|
# entry; rejecting it here silently reintroduced the old
|
|
# helper-only limitation.
|
|
display_name = str(spec.get("name") or "").strip()
|
|
if not display_name:
|
|
display_name = slug.replace("-", " ").replace("_", " ").title()
|
|
app = {
|
|
"name": display_name,
|
|
"repo": detector.get("repo"),
|
|
"official_sources": [],
|
|
"detectors": [],
|
|
}
|
|
v2_apps[slug] = app
|
|
result["runtime_only"].append(slug)
|
|
|
|
# Preserve every statically-proven modern community-scripts marker
|
|
# when a stronger runtime detector becomes primary. This gives new
|
|
# helper installs their official /root/.<app> checker while keeping
|
|
# package/binary/python detectors first for legacy/manual installs.
|
|
helper_markers: list[dict[str, str]] = []
|
|
for candidate in app.get("detectors", []):
|
|
marker_path = candidate.get("file_path")
|
|
marker_regex = candidate.get("file_regex")
|
|
if (
|
|
candidate.get("installed_via") == "file"
|
|
and isinstance(marker_path, str)
|
|
and re.fullmatch(r"/root/\.[A-Za-z0-9_.-]+", marker_path)
|
|
and isinstance(marker_regex, str)
|
|
and marker_regex
|
|
):
|
|
helper_markers.append({
|
|
"path": marker_path,
|
|
"regex": marker_regex,
|
|
"source": "helper_marker",
|
|
})
|
|
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"}
|
|
}
|
|
# Every method-specific field is required at runtime. The old
|
|
# compatibility filter removed python_path/distribution,
|
|
# binary_args and container fields, producing catalog entries
|
|
# that validated statically but could never execute.
|
|
hint = dict(detector)
|
|
for key in passthrough_keys:
|
|
if key in spec:
|
|
hint[key] = spec[key]
|
|
fallbacks = [
|
|
dict(item) for item in hint.get("file_fallbacks", [])
|
|
if isinstance(item, dict)
|
|
]
|
|
known_paths = {
|
|
item.get("path") for item in fallbacks if isinstance(item.get("path"), str)
|
|
}
|
|
primary_path = hint.get("file_path") if hint.get("installed_via") == "file" else None
|
|
for marker in helper_markers:
|
|
if marker["path"] != primary_path and marker["path"] not in known_paths:
|
|
fallbacks.append(marker)
|
|
known_paths.add(marker["path"])
|
|
if fallbacks:
|
|
hint["file_fallbacks"] = fallbacks
|
|
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", "runtime_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())
|