Files
Qortal-Nextcloud-Integration/scripts/check-nextcloud-image-update.sh

409 lines
14 KiB
Bash
Executable File

#!/usr/bin/env bash
set -euo pipefail
env_file=".env.devprod"
compose_file=""
app_service="app"
image_override=""
include_newer_major="0"
timeout_seconds="8"
usage() {
cat <<'USAGE'
Usage: ./scripts/check-nextcloud-image-update.sh [options]
Checks Docker Hub for the latest published Nextcloud tag matching the
configured major track and image flavor, then prints a terminal notice.
Options:
--env-file <path> Env file to read NEXTCLOUD_IMAGE from (default: .env.devprod)
--compose-file <path> Compose file used to inspect the running app container
--service <name> Compose service to inspect when --compose-file is set (default: app)
--image <image> Override NEXTCLOUD_IMAGE for this check only
--include-newer-major Also report newer published major tracks with the same image flavor
--timeout <seconds> HTTP timeout for Docker Hub API requests (default: 8)
-h, --help Show this help
USAGE
}
read_env_value() {
local key="$1"
local line=""
line="$(grep -m1 -E "^${key}=" "${env_file}" || true)"
if [[ -z "${line}" ]]; then
return 1
fi
echo "${line#*=}"
}
while [[ $# -gt 0 ]]; do
case "$1" in
--env-file)
env_file="${2:-}"
shift 2
;;
--compose-file)
compose_file="${2:-}"
shift 2
;;
--service)
app_service="${2:-}"
shift 2
;;
--image)
image_override="${2:-}"
shift 2
;;
--include-newer-major)
include_newer_major="1"
shift
;;
--timeout)
timeout_seconds="${2:-}"
shift 2
;;
-h|--help)
usage
exit 0
;;
*)
echo "Unknown option: $1"
usage
exit 1
;;
esac
done
if [[ ! "${timeout_seconds}" =~ ^[0-9]+$ ]] || [[ "${timeout_seconds}" -lt 1 ]]; then
echo "Invalid --timeout value: ${timeout_seconds}"
exit 1
fi
resolved_image="${image_override:-${NEXTCLOUD_IMAGE:-}}"
if [[ -z "${resolved_image}" && -f "${env_file}" ]]; then
resolved_image="$(read_env_value "NEXTCLOUD_IMAGE" || true)"
fi
resolved_image="${resolved_image:-nextcloud:34-apache}"
running_image=""
running_image_id=""
running_image_os=""
running_image_arch=""
running_image_variant=""
running_repo_digests="[]"
if [[ -n "${compose_file}" && -f "${compose_file}" && -f "${env_file}" ]] && command -v docker >/dev/null 2>&1; then
container_id="$(docker compose -f "${compose_file}" --env-file "${env_file}" ps -q "${app_service}" 2>/dev/null | head -n1 || true)"
if [[ -n "${container_id}" ]]; then
running_image="$(docker inspect --format '{{.Config.Image}}' "${container_id}" 2>/dev/null || true)"
running_image_id="$(docker inspect --format '{{.Image}}' "${container_id}" 2>/dev/null || true)"
if [[ -n "${running_image_id}" ]]; then
running_repo_digests="$(docker image inspect --format '{{json .RepoDigests}}' "${running_image_id}" 2>/dev/null || true)"
running_repo_digests="${running_repo_digests:-[]}"
running_image_os="$(docker image inspect --format '{{.Os}}' "${running_image_id}" 2>/dev/null || true)"
running_image_arch="$(docker image inspect --format '{{.Architecture}}' "${running_image_id}" 2>/dev/null || true)"
running_image_variant="$(docker image inspect --format '{{.Variant}}' "${running_image_id}" 2>/dev/null || true)"
fi
fi
fi
if ! command -v python3 >/dev/null 2>&1; then
echo "[nextcloud-image-check] Skipped: python3 is required for Docker Hub tag checks."
exit 0
fi
python3 - "${resolved_image}" "${running_image}" "${running_image_id}" "${running_repo_digests}" "${running_image_os}" "${running_image_arch}" "${running_image_variant}" "${include_newer_major}" "${timeout_seconds}" <<'PY'
import json
import re
import sys
import urllib.error
import urllib.request
configured_image = sys.argv[1].strip()
running_image = sys.argv[2].strip()
running_image_id = sys.argv[3].strip()
try:
running_repo_digests = json.loads(sys.argv[4])
except Exception:
running_repo_digests = []
running_image_os = sys.argv[5].strip()
running_image_arch = sys.argv[6].strip()
running_image_variant = sys.argv[7].strip()
include_newer_major = sys.argv[8].strip() == "1"
timeout = int(sys.argv[9])
prefix = "[nextcloud-image-check]"
def log(message: str) -> None:
print(f"{prefix} {message}")
def parse_image_reference(ref: str):
if "@" in ref:
return None, "digest references are not supported"
tag = "latest"
repo_ref = ref
last_slash = ref.rfind("/")
last_colon = ref.rfind(":")
if last_colon > last_slash:
repo_ref = ref[:last_colon]
tag = ref[last_colon + 1 :]
path_parts = repo_ref.split("/")
if len(path_parts) == 1:
namespace = "library"
repository = path_parts[0]
elif len(path_parts) == 2 and "." not in path_parts[0] and ":" not in path_parts[0] and path_parts[0] != "localhost":
namespace, repository = path_parts
else:
return None, "only Docker Hub library/user images are supported"
return {
"repo_ref": repo_ref,
"namespace": namespace,
"repository": repository,
"tag": tag,
}, None
def parse_tag(tag: str):
match = re.fullmatch(r"(?P<major>\d+)(?:\.(?P<minor>\d+))?(?:\.(?P<patch>\d+))?(?P<suffix>-[A-Za-z0-9._-]+)?", tag)
if not match:
return None
minor = match.group("minor")
patch = match.group("patch")
return {
"major": int(match.group("major")),
"minor": int(minor) if minor is not None else None,
"patch": int(patch) if patch is not None else None,
"suffix": match.group("suffix") or "",
}
def docker_hub_tags_url(parsed_ref):
return (
"https://hub.docker.com/v2/repositories/"
f"{parsed_ref['namespace']}/{parsed_ref['repository']}/tags?page_size=100"
)
def docker_hub_tag_url(parsed_ref, tag):
return (
"https://hub.docker.com/v2/repositories/"
f"{parsed_ref['namespace']}/{parsed_ref['repository']}/tags/{tag}"
)
def find_latest_matching_tag(parsed_ref, parsed_tag, same_major=True):
namespace = parsed_ref["namespace"]
repository = parsed_ref["repository"]
major = parsed_tag["major"]
suffix = parsed_tag["suffix"]
best = None
best_tag = None
url = docker_hub_tags_url(parsed_ref)
while url:
with urllib.request.urlopen(url, timeout=timeout) as response:
payload = json.load(response)
for entry in payload.get("results", []):
name = str(entry.get("name", "")).strip()
parsed_candidate = parse_tag(name)
if parsed_candidate is None:
continue
if same_major and parsed_candidate["major"] != major:
continue
if parsed_candidate["suffix"] != suffix:
continue
version_key = (
parsed_candidate["major"],
parsed_candidate["minor"] if parsed_candidate["minor"] is not None else -1,
parsed_candidate["patch"] if parsed_candidate["patch"] is not None else -1,
)
if best is None or version_key > best:
best = version_key
best_tag = name
url = payload.get("next")
return best, best_tag
def docker_hub_tag_digests(parsed_ref, tag, image_os, image_arch, image_variant):
with urllib.request.urlopen(docker_hub_tag_url(parsed_ref, tag), timeout=timeout) as response:
payload = json.load(response)
digests = set()
top_level_digest = str(payload.get("digest") or "").strip()
if top_level_digest.startswith("sha256:"):
digests.add(top_level_digest)
images = payload.get("images", [])
platform_matched = False
for image_info in images if isinstance(images, list) else []:
digest = str(image_info.get("digest") or "").strip()
if not digest.startswith("sha256:"):
continue
if image_os and image_arch:
same_platform = (
str(image_info.get("os") or "") == image_os
and str(image_info.get("architecture") or "") == image_arch
and (
not image_variant
or str(image_info.get("variant") or "") == image_variant
)
)
if same_platform:
platform_matched = True
digests.add(digest)
else:
digests.add(digest)
if image_os and image_arch and not platform_matched:
for image_info in images if isinstance(images, list) else []:
digest = str(image_info.get("digest") or "").strip()
if digest.startswith("sha256:"):
digests.add(digest)
return digests
def describe_platform():
if not running_image_os or not running_image_arch:
return ""
value = f"{running_image_os}/{running_image_arch}"
if running_image_variant:
value += f"/{running_image_variant}"
return value
def version_tuple(parsed_tag):
return (
parsed_tag["major"],
parsed_tag["minor"] if parsed_tag["minor"] is not None else -1,
parsed_tag["patch"] if parsed_tag["patch"] is not None else -1,
)
def report_newer_major_status(label, parsed_ref, parsed_tag, image, same_major_best_tag):
if not include_newer_major:
return
try:
best_any, best_any_tag = find_latest_matching_tag(
parsed_ref,
parsed_tag,
same_major=False,
)
except Exception as exc:
log(f"Skipped newer major check for {label}: {exc}")
return
if best_any is not None and best_any_tag != same_major_best_tag and version_tuple(parsed_tag) < best_any:
latest_major_image = f"{parsed_ref['repo_ref']}:{best_any_tag}"
log(f"Newer major track available: {latest_major_image} ({label}: {image})")
log(f"Major upgrade command: ./upgrade-devprod-nextcloud.sh --nextcloud-image {latest_major_image}")
def report_tag_status(label, image):
parsed_ref, parse_error = parse_image_reference(image)
if parsed_ref is None:
log(f"Skipped {label}: {parse_error} ({image})")
return None, None
parsed_tag = parse_tag(parsed_ref["tag"])
if parsed_tag is None:
log(f"Skipped {label}: unsupported tag pattern '{parsed_ref['tag']}' for {image}")
return parsed_ref, None
try:
best, best_tag = find_latest_matching_tag(parsed_ref, parsed_tag)
except urllib.error.HTTPError as exc:
log(f"Skipped {label}: Docker Hub lookup failed with HTTP {exc.code} for {parsed_ref['namespace']}/{parsed_ref['repository']}")
return parsed_ref, parsed_tag
except urllib.error.URLError as exc:
reason = getattr(exc, "reason", exc)
log(f"Skipped {label}: Docker Hub lookup failed ({reason})")
return parsed_ref, parsed_tag
except Exception as exc: # pragma: no cover - best effort status output
log(f"Skipped {label}: unexpected Docker Hub lookup error ({exc})")
return parsed_ref, parsed_tag
if best_tag is None:
log(
f"No published Docker Hub tags found for {parsed_ref['repo_ref']} "
f"matching major {parsed_tag['major']}{parsed_tag['suffix']}"
)
return parsed_ref, parsed_tag
latest_image = f"{parsed_ref['repo_ref']}:{best_tag}"
current_is_floating = parsed_tag["minor"] is None or parsed_tag["patch"] is None
if parsed_ref["tag"] == best_tag:
log(f"{label.capitalize()} image is already on the newest published matching tag: {image}")
report_newer_major_status(label, parsed_ref, parsed_tag, image, best_tag)
return parsed_ref, parsed_tag
if current_is_floating:
log(f"Published latest matching tag: {latest_image} ({label} track: {image})")
log(f"Upgrade test command: ./recreate-devprod.sh --nextcloud-image {latest_image}")
log(f"To persist after testing, set NEXTCLOUD_IMAGE={latest_image} in .env.devprod")
report_newer_major_status(label, parsed_ref, parsed_tag, image, best_tag)
return parsed_ref, parsed_tag
if version_tuple(parsed_tag) < best:
log(f"Newer published matching tag available: {latest_image} ({label}: {image})")
log(f"Upgrade test command: ./recreate-devprod.sh --nextcloud-image {latest_image}")
log(f"To persist after testing, set NEXTCLOUD_IMAGE={latest_image} in .env.devprod")
else:
log(f"{label.capitalize()} image is already on the newest published matching tag: {image}")
report_newer_major_status(label, parsed_ref, parsed_tag, image, best_tag)
return parsed_ref, parsed_tag
def digest_values(repo_digests):
values = set()
for item in repo_digests if isinstance(repo_digests, list) else []:
digest = str(item).rsplit("@", 1)[-1].strip()
if digest.startswith("sha256:"):
values.add(digest)
return values
if running_image:
log(f"Running app container image: {running_image}")
if running_image_id:
log(f"Running app local image ID: {running_image_id}")
if running_image != configured_image:
log(f"Configured image for next recreate: {configured_image}")
configured_ref, configured_tag = report_tag_status("configured", configured_image)
if running_image and running_image != configured_image:
running_ref, running_tag = report_tag_status("running", running_image)
else:
running_ref, running_tag = configured_ref, configured_tag
if running_image and running_ref is not None and running_tag is not None:
try:
remote_digests = docker_hub_tag_digests(
running_ref,
running_ref["tag"],
running_image_os,
running_image_arch,
running_image_variant,
)
except Exception as exc:
log(f"Skipped running digest check for {running_image}: {exc}")
else:
local_digests = digest_values(running_repo_digests)
if not local_digests:
log(f"Running digest check inconclusive: local RepoDigests are unavailable for {running_image}")
elif not remote_digests:
log(f"Running digest check inconclusive: registry digest metadata is unavailable for {running_image}")
elif local_digests & remote_digests:
suffix = f" ({describe_platform()})" if describe_platform() else ""
log(f"Running local image digest matches the current registry digest for {running_image}{suffix}")
else:
log(
f"Registry digest for {running_image} differs from the local running image. "
"A pull/recreate may update Nextcloud even though the tag name is unchanged."
)
PY