mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-09-15 11:16:50 +00:00
341 lines
15 KiB
Python
341 lines
15 KiB
Python
"""Resolving the version a Docker image pull would install.
|
|
|
|
The registry answers the question the Updates tab actually asks: not "what is
|
|
the newest upstream release" but "what do I get if I re-pull this tag". These
|
|
tests pin the rules that keep that number honest — and the cases where the
|
|
honest answer is no number at all.
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
|
if str(SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS_DIR))
|
|
|
|
import lxc_apps
|
|
|
|
|
|
def _blob(payload: dict) -> tuple[bytes, str]:
|
|
body = json.dumps(payload).encode("utf-8")
|
|
return body, "sha256:" + hashlib.sha256(body).hexdigest()
|
|
|
|
|
|
def _image_index(platform_digest: str) -> dict:
|
|
"""An index shaped like the real ones: platforms plus attestations."""
|
|
return {
|
|
"mediaType": "application/vnd.oci.image.index.v1+json",
|
|
"manifests": [
|
|
{"digest": platform_digest,
|
|
"platform": {"os": "linux", "architecture": "amd64"}},
|
|
{"digest": "sha256:" + "b" * 64,
|
|
"platform": {"os": "linux", "architecture": "arm64", "variant": "v8"}},
|
|
{"digest": "sha256:" + "c" * 64,
|
|
"platform": {"os": "unknown", "architecture": "unknown"},
|
|
"annotations": {"vnd.docker.reference.type": "attestation-manifest"}},
|
|
],
|
|
}
|
|
|
|
|
|
def _manifest(config_digest: str) -> dict:
|
|
return {
|
|
"mediaType": "application/vnd.oci.image.manifest.v1+json",
|
|
"config": {
|
|
"mediaType": "application/vnd.oci.image.config.v1+json",
|
|
"digest": config_digest,
|
|
},
|
|
}
|
|
|
|
|
|
class RemoteImageVersionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
with lxc_apps._docker_remote_config_lock:
|
|
lxc_apps._docker_remote_config_cache.clear()
|
|
|
|
def _wire(self, remote_version="3.1.0", labels=None, config_media_type=None):
|
|
"""Serve a three-hop index -> manifest -> config walk from memory."""
|
|
config_payload = {"config": {"Labels": labels if labels is not None else {
|
|
"org.opencontainers.image.version": remote_version,
|
|
}}}
|
|
config_body, config_digest = _blob(config_payload)
|
|
manifest = _manifest(config_digest)
|
|
if config_media_type is not None:
|
|
manifest["config"]["mediaType"] = config_media_type
|
|
manifest_body, manifest_digest = _blob(manifest)
|
|
index_body, index_digest = _blob(_image_index(manifest_digest))
|
|
documents = {
|
|
index_digest: index_body,
|
|
manifest_digest: manifest_body,
|
|
config_digest: config_body,
|
|
}
|
|
calls = []
|
|
|
|
def fake_request(url, headers, method="HEAD", token=None, max_bytes=0,
|
|
follow_redirect=True):
|
|
calls.append(url)
|
|
digest = url.rsplit("/", 1)[-1]
|
|
if digest in documents:
|
|
return {}, documents[digest], "token", None
|
|
return None, None, token, f"registry HTTP 404 for {digest}"
|
|
|
|
return index_digest, calls, mock.patch.object(lxc_apps, "_registry_request", fake_request)
|
|
|
|
def _item(self, index_digest, **overrides):
|
|
item = {
|
|
"api_host": "ghcr.io",
|
|
"repository": "immich-app/immich-server",
|
|
"reference": "ghcr.io/immich-app/immich-server:release",
|
|
"remote_digest": index_digest,
|
|
"installed_version": "3.0.1",
|
|
"installed_version_source": "image_label:org.opencontainers.image.version",
|
|
"platform": {"os": "linux", "architecture": "amd64", "variant": ""},
|
|
}
|
|
item.update(overrides)
|
|
return item
|
|
|
|
def test_resolves_version_from_the_remote_image_config(self):
|
|
index_digest, calls, patch = self._wire()
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest))
|
|
self.assertEqual(version, "3.1.0")
|
|
self.assertEqual(source, "remote_image_label:org.opencontainers.image.version")
|
|
self.assertEqual(len(calls), 3, "index, platform manifest and config blob")
|
|
|
|
def test_attestation_manifest_is_never_selected(self):
|
|
"""Its config is a provenance document, not an image."""
|
|
digest = lxc_apps._select_platform_manifest(
|
|
_image_index("sha256:" + "a" * 64),
|
|
{"os": "linux", "architecture": "amd64"},
|
|
)
|
|
self.assertEqual(digest, "sha256:" + "a" * 64)
|
|
|
|
def test_arm64_v8_matches_bare_arm64(self):
|
|
digest = lxc_apps._select_platform_manifest(
|
|
_image_index("sha256:" + "a" * 64),
|
|
{"os": "linux", "architecture": "arm64", "variant": ""},
|
|
)
|
|
self.assertEqual(digest, "sha256:" + "b" * 64)
|
|
|
|
def test_missing_platform_yields_no_version(self):
|
|
"""No build for this host means a pull would fail; claim nothing."""
|
|
index_digest, _calls, patch = self._wire()
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest, platform={"os": "linux", "architecture": "riscv64"}))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "remote_platform_missing")
|
|
|
|
def test_same_version_rebuilt_reports_no_number(self):
|
|
index_digest, _calls, patch = self._wire(remote_version="3.0.1")
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "version_not_comparable")
|
|
|
|
def test_lower_remote_version_is_rejected(self):
|
|
index_digest, _calls, patch = self._wire(remote_version="2.9.0")
|
|
with patch:
|
|
version, _source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest))
|
|
self.assertIsNone(version)
|
|
|
|
def test_linuxserver_build_suffix_is_an_update(self):
|
|
"""LinuxServer bumps the -lsNNN build; the app version may not move.
|
|
|
|
Label shape taken from the live lscr.io/linuxserver/radarr image.
|
|
"""
|
|
index_digest, _calls, patch = self._wire(
|
|
labels={"org.opencontainers.image.version": "6.3.0.10514-ls314",
|
|
"build_version": "Linuxserver.io version:- 6.3.0.10514-ls314"})
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest, installed_version="6.3.0.10514-ls313"))
|
|
self.assertEqual(version, "6.3.0.10514-ls314")
|
|
self.assertEqual(source, "remote_image_label:org.opencontainers.image.version")
|
|
|
|
def test_label_key_must_match_the_installed_side(self):
|
|
"""Two publishers' label schemes are not a comparison."""
|
|
index_digest, _calls, patch = self._wire()
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest, installed_version_source="image_label:Version"))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "version_source_mismatch")
|
|
|
|
def test_config_without_labels_reports_no_version(self):
|
|
index_digest, _calls, patch = self._wire(labels={})
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "remote_no_labels")
|
|
|
|
def test_moving_tag_label_is_rejected(self):
|
|
index_digest, _calls, patch = self._wire(
|
|
labels={"org.opencontainers.image.version": "main"})
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "remote_no_labels")
|
|
|
|
def test_non_image_artifact_is_unsupported(self):
|
|
index_digest, _calls, patch = self._wire(
|
|
config_media_type="application/vnd.cncf.helm.config.v1+json")
|
|
with patch:
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item(index_digest))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "remote_unsupported_manifest")
|
|
|
|
def test_tampered_document_is_discarded(self):
|
|
"""Documents are content-addressed; a mismatch is not trusted."""
|
|
def fake_request(url, headers, method="HEAD", token=None, max_bytes=0,
|
|
follow_redirect=True):
|
|
return {}, b'{"mediaType": "application/vnd.oci.image.manifest.v1+json"}', "t", None
|
|
|
|
with mock.patch.object(lxc_apps, "_registry_request", fake_request):
|
|
version, source = lxc_apps._docker_available_version_from_registry(
|
|
self._item("sha256:" + "f" * 64))
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "remote_fetch_error")
|
|
|
|
def test_registry_error_leaves_the_digest_verdict_alone(self):
|
|
def fake_request(url, headers, method="HEAD", token=None, max_bytes=0,
|
|
follow_redirect=True):
|
|
return None, None, None, "registry network error: timed out"
|
|
|
|
item = self._item("sha256:" + "e" * 64, update_available=True)
|
|
with mock.patch.object(lxc_apps, "_registry_request", fake_request):
|
|
version, source = lxc_apps._docker_available_version_from_registry(item)
|
|
self.assertIsNone(version)
|
|
self.assertEqual(source, "remote_fetch_error")
|
|
self.assertIs(item["update_available"], True, "the digest verdict stands on its own")
|
|
|
|
def test_second_image_with_the_same_digest_is_served_from_cache(self):
|
|
index_digest, calls, patch = self._wire()
|
|
with patch:
|
|
lxc_apps._docker_available_version_from_registry(self._item(index_digest))
|
|
lxc_apps._docker_available_version_from_registry(self._item(index_digest))
|
|
self.assertEqual(len(calls), 3, "the second lookup must not hit the registry")
|
|
|
|
|
|
class RegistryRedirectTests(unittest.TestCase):
|
|
def test_authorization_is_dropped_on_the_cdn_hop(self):
|
|
"""Signed storage URLs reject a second auth mechanism."""
|
|
seen = []
|
|
|
|
def fake_open(url, headers, method, max_bytes):
|
|
seen.append((url, dict(headers)))
|
|
if "blobs" in url:
|
|
return {"Location": "https://cdn.example/blob"}, None, 307, "https://cdn.example/blob", None
|
|
return {}, b"{}", 200, None, None
|
|
|
|
with mock.patch.object(lxc_apps, "_registry_open", fake_open):
|
|
lxc_apps._registry_request(
|
|
"https://ghcr.io/v2/x/y/blobs/sha256:abc",
|
|
{"User-Agent": "ProxMenux-Monitor", "Authorization": "Bearer secret"},
|
|
method="GET", max_bytes=1024,
|
|
)
|
|
self.assertEqual(len(seen), 2)
|
|
self.assertIn("Authorization", seen[0][1])
|
|
self.assertNotIn("Authorization", seen[1][1])
|
|
|
|
def test_digest_request_keeps_its_contract(self):
|
|
def fake_open(url, headers, method, max_bytes):
|
|
return {"Docker-Content-Digest": "sha256:deadbeef"}, None, 200, None, None
|
|
|
|
with mock.patch.object(lxc_apps, "_registry_open", fake_open):
|
|
digest, error = lxc_apps._registry_digest_request("https://ghcr.io/v2/x/y/manifests/latest", {})
|
|
self.assertEqual(digest, "sha256:deadbeef")
|
|
self.assertIsNone(error)
|
|
|
|
|
|
class ContainerIdentityTests(unittest.TestCase):
|
|
"""A container's catalog identity comes from a detector that declares it."""
|
|
|
|
HINTS = {
|
|
"immich": {
|
|
"installed_via": "file",
|
|
"file_path": "/root/.immich",
|
|
"logo": "https://example.invalid/immich.webp",
|
|
"alt_detectors": [
|
|
{"installed_via": "docker_label", "container_name": "immich_server",
|
|
"label": "org.opencontainers.image.version"},
|
|
],
|
|
},
|
|
"netalertx": {
|
|
"installed_via": "docker_label",
|
|
"container_name": "netalertx",
|
|
"label": "org.opencontainers.image.version",
|
|
},
|
|
"postgresql": {"installed_via": "dpkg", "package": "postgresql"},
|
|
}
|
|
|
|
def test_index_covers_primary_and_alternate_detectors(self):
|
|
with mock.patch.object(lxc_apps, "_fetch_tracking_hints", lambda: self.HINTS):
|
|
index = lxc_apps._docker_container_slug_index()
|
|
self.assertEqual(index.get("immich_server"), "immich")
|
|
self.assertEqual(index.get("netalertx"), "netalertx")
|
|
|
|
def test_unclaimed_container_names_stay_out(self):
|
|
"""A container merely called postgres is not an application claim."""
|
|
with mock.patch.object(lxc_apps, "_fetch_tracking_hints", lambda: self.HINTS):
|
|
index = lxc_apps._docker_container_slug_index()
|
|
self.assertNotIn("postgres", index)
|
|
|
|
def test_declared_container_resolves_a_name_heuristics_cannot(self):
|
|
with mock.patch.object(lxc_apps, "_fetch_tracking_hints", lambda: self.HINTS), \
|
|
mock.patch.object(lxc_apps, "_catalog_lookup", lambda slug: {"name": "Immich"} if slug == "immich" else None):
|
|
meta = lxc_apps._docker_service_catalog_meta(
|
|
"immich-server", "immich_server",
|
|
"ghcr.io/immich-app/immich-server:release",
|
|
)
|
|
self.assertEqual(meta["slug"], "immich")
|
|
self.assertEqual(meta["name"], "Immich")
|
|
self.assertEqual(meta["logo_url"], "https://example.invalid/immich.webp")
|
|
|
|
|
|
class UpdateNotificationWordingTests(unittest.TestCase):
|
|
def test_alert_leads_with_the_application_name(self):
|
|
payload = lxc_apps._docker_stack_notification_payload(
|
|
102,
|
|
{"state": {}},
|
|
{"images": [{
|
|
"reference": "vaultwarden/server:latest",
|
|
"display_name": "Vaultwarden",
|
|
"installed_version": "1.37.2",
|
|
"available_version": "1.38.0",
|
|
"update_available": True,
|
|
"remote_digest": "sha256:" + "a" * 64,
|
|
}]},
|
|
"vaultwarden",
|
|
)
|
|
self.assertIsNotNone(payload)
|
|
self.assertIn("• Vaultwarden (vaultwarden/server:latest): 1.37.2 → 1.38.0", payload["details"])
|
|
|
|
def test_unnamed_image_keeps_its_reference(self):
|
|
payload = lxc_apps._docker_stack_notification_payload(
|
|
102,
|
|
{"state": {}},
|
|
{"images": [{
|
|
"reference": "valkey/valkey:8-bookworm",
|
|
"installed_version": None,
|
|
"available_version": None,
|
|
"update_available": True,
|
|
"remote_digest": "sha256:" + "b" * 64,
|
|
}]},
|
|
"immich",
|
|
)
|
|
self.assertIn("• valkey/valkey:8-bookworm: new registry digest", payload["details"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|