179 lines
7.1 KiB
Python
179 lines
7.1 KiB
Python
"""Strict source identities for immutable, registry-backed catalog entries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
import json
|
|
import re
|
|
from urllib.parse import unquote, urlsplit
|
|
|
|
|
|
_SHA256 = re.compile(r"[0-9a-f]{64}\Z")
|
|
_COMMIT = re.compile(r"[0-9a-f]{40}(?:[0-9a-f]{24})?\Z")
|
|
_REPO = re.compile(r"govoplan-[a-z0-9-]+\Z")
|
|
_VERSION = re.compile(r"\d+\.\d+\.\d+(?:[-+][0-9A-Za-z.-]+)?\Z")
|
|
_PYTHON = re.compile(
|
|
r"(?P<name>govoplan-[a-z0-9-]+)(?:\[[a-z0-9_,.-]+\])? @ "
|
|
r"(?P<url>https://\S+)#sha256=(?P<digest>[0-9a-f]{64})\Z"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RegistrySource:
|
|
repository: str
|
|
version: str
|
|
commit: str
|
|
tag_object: str
|
|
|
|
|
|
def registry_artifact_conflicts(entries: list[object]) -> tuple[str, ...]:
|
|
"""Allow repeated module projections only when package artifacts agree."""
|
|
observed: dict[tuple[str, str], str] = {}
|
|
issues = []
|
|
for entry in entries:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
try:
|
|
source = registry_entry_source(entry)
|
|
except ValueError as exc:
|
|
issues.append(str(exc))
|
|
continue
|
|
if source is None:
|
|
continue
|
|
for kind, identity in entry["artifact_integrity"].items():
|
|
if kind not in {"python", "webui"}:
|
|
continue
|
|
key = (source.repository, kind)
|
|
encoded = json.dumps(identity, sort_keys=True, separators=(",", ":"))
|
|
if observed.setdefault(key, encoded) != encoded:
|
|
issues.append(f"conflicting {kind} registry artifacts for {source.repository}")
|
|
return tuple(issues)
|
|
|
|
|
|
def registry_entry_source(entry: dict[str, object]) -> RegistrySource | None:
|
|
"""Admit registry refs only with a complete matching artifact/source binding.
|
|
|
|
Git-backed catalogs keep their existing validation path. An HTTPS Python
|
|
requirement cannot masquerade as a source-only/non-Python entry when its
|
|
registry provenance is missing or inconsistent.
|
|
"""
|
|
|
|
ref = entry.get("python_ref")
|
|
if not isinstance(ref, str) or " @ https://" not in ref:
|
|
return None
|
|
match = _PYTHON.fullmatch(ref)
|
|
source = entry.get("source")
|
|
version = entry.get("version")
|
|
package = entry.get("python_package")
|
|
integrity = entry.get("artifact_integrity")
|
|
if (
|
|
match is None
|
|
or not isinstance(source, dict)
|
|
or not isinstance(integrity, dict)
|
|
or not isinstance(version, str)
|
|
or _VERSION.fullmatch(version) is None
|
|
or package != match.group("name")
|
|
):
|
|
raise ValueError("registry entry has no complete package/source identity")
|
|
repo = source.get("repository")
|
|
commit = source.get("commit")
|
|
tag_object = source.get("tag_object_sha")
|
|
if (
|
|
not isinstance(repo, str)
|
|
or _REPO.fullmatch(repo) is None
|
|
or repo != package
|
|
or source.get("tag") != f"v{version}"
|
|
or not isinstance(commit, str)
|
|
or _COMMIT.fullmatch(commit) is None
|
|
or not isinstance(tag_object, str)
|
|
or _COMMIT.fullmatch(tag_object) is None
|
|
):
|
|
raise ValueError("registry entry has invalid immutable tag provenance")
|
|
python = _artifact(
|
|
integrity.get("python"), ref=ref, package=package, version=version,
|
|
commit=commit,
|
|
)
|
|
if python["url"] != match.group("url") or python["sha256"] != match.group("digest"):
|
|
raise ValueError("registry Python ref differs from its artifact identity")
|
|
webui_package = entry.get("webui_package")
|
|
webui_ref = entry.get("webui_ref")
|
|
if bool(webui_package) != bool(webui_ref):
|
|
raise ValueError("registry WebUI package and ref must be declared together")
|
|
if webui_package:
|
|
if not isinstance(webui_package, str) or re.fullmatch(
|
|
r"@govoplan/[a-z0-9-]+-webui", webui_package
|
|
) is None or not isinstance(webui_ref, str):
|
|
raise ValueError("registry WebUI package identity is malformed")
|
|
if webui_package != f"@govoplan/{repo.removeprefix('govoplan-')}-webui":
|
|
raise ValueError("registry WebUI package belongs to another source repository")
|
|
webui = _artifact(
|
|
integrity.get("webui"), ref=webui_ref, package=webui_package,
|
|
version=version, commit=commit,
|
|
)
|
|
if webui_ref != webui["url"]:
|
|
raise ValueError("registry WebUI ref differs from its artifact identity")
|
|
sri = webui.get("integrity")
|
|
try:
|
|
valid_sri = isinstance(sri, str) and sri.startswith("sha512-") and len(
|
|
base64.b64decode(sri[7:], validate=True)
|
|
) == 64
|
|
except ValueError:
|
|
valid_sri = False
|
|
if not valid_sri:
|
|
raise ValueError("registry WebUI artifact needs a SHA-512 integrity identity")
|
|
elif "webui" in integrity:
|
|
raise ValueError("registry entry carries an unexpected WebUI artifact")
|
|
return RegistrySource(repo, version, commit, tag_object)
|
|
|
|
|
|
def _artifact(
|
|
value: object, *, ref: str, package: str, version: str, commit: str,
|
|
) -> dict[str, object]:
|
|
if not isinstance(value, dict):
|
|
raise ValueError("registry entry is missing artifact integrity")
|
|
url = value.get("url")
|
|
filename = value.get("filename")
|
|
digest = value.get("sha256")
|
|
size = value.get("size")
|
|
parsed = urlsplit(url) if isinstance(url, str) else None
|
|
url_filename = unquote(parsed.path.rsplit("/", 1)[-1]) if parsed else ""
|
|
decoded_parts = unquote(parsed.path).split("/") if parsed else []
|
|
expected_filenames = {url_filename}
|
|
expected_path = [package, version, url_filename]
|
|
if package.startswith("@govoplan/"):
|
|
# npm pack includes the scope in its local filename; the registry's
|
|
# immutable download URL uses the unscoped package basename.
|
|
expected_filenames = {
|
|
f"govoplan-{package.split('/', 1)[1]}-{version}.tgz",
|
|
} if url_filename == f"{package.split('/', 1)[1]}-{version}.tgz" else set()
|
|
expected_path = [*package.split("/"), "-", version, url_filename]
|
|
if (
|
|
parsed is None
|
|
or parsed.scheme != "https"
|
|
or not parsed.netloc
|
|
or parsed.username is not None
|
|
or parsed.password is not None
|
|
or parsed.fragment
|
|
or parsed.query
|
|
or any(part in {".", ".."} for part in unquote(parsed.path).split("/"))
|
|
or "%" in unquote(parsed.path)
|
|
or "\\" in unquote(parsed.path)
|
|
or any(ord(character) < 32 for character in url)
|
|
or decoded_parts[-len(expected_path):] != expected_path
|
|
or not isinstance(filename, str)
|
|
or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._+-]{0,254}", filename) is None
|
|
or filename not in expected_filenames
|
|
or not isinstance(digest, str)
|
|
or _SHA256.fullmatch(digest) is None
|
|
or not isinstance(size, int)
|
|
or isinstance(size, bool)
|
|
or not 0 < size <= 512 * 1024 * 1024
|
|
or value.get("ref") != ref
|
|
or value.get("registry_identity") != f"{package}@{version}"
|
|
or value.get("git_ref") != f"v{version}"
|
|
or value.get("source_commit") != commit
|
|
):
|
|
raise ValueError("registry artifact ref, version, bytes, or source binding is inconsistent")
|
|
return value
|