713 lines
26 KiB
Python
713 lines
26 KiB
Python
"""Bounded verification for signed GovOPlaN runtime distributions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from datetime import UTC, datetime
|
|
import hashlib
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import socket
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
from typing import Any, Mapping
|
|
from urllib.parse import urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
MAX_MANIFEST_BYTES = 4 * 1024 * 1024
|
|
MAX_KEYRING_BYTES = 1024 * 1024
|
|
MAX_OFFLINE_INDEX_BYTES = 4 * 1024 * 1024
|
|
MAX_OFFLINE_IMAGE_BYTES = 16 * 1024 * 1024 * 1024
|
|
SHA256 = re.compile(r"^[0-9a-f]{64}$")
|
|
DIGEST_IMAGE = re.compile(r"^[^@\s]+@sha256:[0-9a-f]{64}$")
|
|
TOKEN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+!-]{0,127}$")
|
|
MODULE_ID = re.compile(r"^[a-z][a-z0-9_]{1,63}$")
|
|
KEY_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
|
|
PLATFORMS = ("linux/amd64", "linux/arm64")
|
|
MANIFEST_FILENAME = "distribution-manifest.json"
|
|
KEYRING_FILENAME = "distribution-keyring.json"
|
|
|
|
|
|
class DistributionError(ValueError):
|
|
"""Distribution evidence is absent, malformed, or untrusted."""
|
|
|
|
|
|
def canonical_signed_payload(payload: Mapping[str, Any]) -> bytes:
|
|
unsigned = dict(payload)
|
|
unsigned.pop("signatures", None)
|
|
return json.dumps(
|
|
unsigned,
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
|
|
|
|
def canonical_json(payload: Mapping[str, Any]) -> bytes:
|
|
return (
|
|
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
|
).encode("utf-8")
|
|
|
|
|
|
def load_bounded_json(path: Path, *, maximum_bytes: int) -> dict[str, Any]:
|
|
encoded = read_bounded_bytes(path, maximum_bytes=maximum_bytes)
|
|
try:
|
|
value = json.loads(encoded)
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise DistributionError(f"trusted JSON file is malformed: {path}") from exc
|
|
if not isinstance(value, dict):
|
|
raise DistributionError(f"trusted JSON root must be an object: {path}")
|
|
return value
|
|
|
|
|
|
def read_bounded_bytes(path: Path, *, maximum_bytes: int) -> bytes:
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(path, flags)
|
|
except OSError as exc:
|
|
raise DistributionError(f"cannot open trusted JSON file: {path}") from exc
|
|
try:
|
|
opened = os.fstat(descriptor)
|
|
if not stat.S_ISREG(opened.st_mode) or opened.st_size > maximum_bytes:
|
|
raise DistributionError(
|
|
f"trusted JSON file is invalid or too large: {path}"
|
|
)
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
while True:
|
|
chunk = os.read(descriptor, min(64 * 1024, maximum_bytes + 1 - total))
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
total += len(chunk)
|
|
if total > maximum_bytes:
|
|
raise DistributionError(f"trusted JSON file is too large: {path}")
|
|
final = os.fstat(descriptor)
|
|
if (opened.st_dev, opened.st_ino, opened.st_size, opened.st_mtime_ns) != (
|
|
final.st_dev,
|
|
final.st_ino,
|
|
final.st_size,
|
|
final.st_mtime_ns,
|
|
):
|
|
raise DistributionError(f"trusted JSON file changed while read: {path}")
|
|
finally:
|
|
os.close(descriptor)
|
|
return b"".join(chunks)
|
|
|
|
|
|
def fetch_bounded_https(
|
|
url: str,
|
|
*,
|
|
maximum_bytes: int,
|
|
timeout_seconds: float = 15.0,
|
|
allow_private_host: bool = False,
|
|
) -> bytes:
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme != "https" or not parsed.hostname:
|
|
raise DistributionError("distribution downloads require an absolute HTTPS URL")
|
|
if parsed.username or parsed.password or parsed.fragment:
|
|
raise DistributionError(
|
|
"distribution URL must not contain credentials or a fragment"
|
|
)
|
|
if not allow_private_host:
|
|
_require_public_host(parsed.hostname)
|
|
request = Request(url, headers={"Accept": "application/json"})
|
|
try:
|
|
with urlopen(request, timeout=timeout_seconds) as response: # noqa: S310
|
|
final = urlsplit(response.geturl())
|
|
if final.scheme != "https":
|
|
raise DistributionError("distribution redirect left HTTPS")
|
|
declared = response.headers.get("Content-Length")
|
|
if declared and int(declared) > maximum_bytes:
|
|
raise DistributionError("distribution download exceeds its size limit")
|
|
value = response.read(maximum_bytes + 1)
|
|
except DistributionError:
|
|
raise
|
|
except (OSError, ValueError) as exc:
|
|
raise DistributionError(f"distribution download failed: {exc}") from exc
|
|
if len(value) > maximum_bytes:
|
|
raise DistributionError("distribution download exceeds its size limit")
|
|
return value
|
|
|
|
|
|
def decode_json_bytes(value: bytes, *, label: str) -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(value)
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise DistributionError(f"{label} is not valid JSON") from exc
|
|
if not isinstance(payload, dict):
|
|
raise DistributionError(f"{label} root must be an object")
|
|
return payload
|
|
|
|
|
|
def validate_manifest(
|
|
payload: Mapping[str, Any],
|
|
*,
|
|
expected_channel: str | None = None,
|
|
now: datetime | None = None,
|
|
) -> None:
|
|
_exact_keys(
|
|
payload,
|
|
required={
|
|
"schema_version",
|
|
"channel",
|
|
"sequence",
|
|
"version",
|
|
"issued_at",
|
|
"expires_at",
|
|
"revoked",
|
|
"deployer",
|
|
"images",
|
|
"dependencies",
|
|
"composition",
|
|
"signatures",
|
|
},
|
|
label="distribution manifest",
|
|
)
|
|
if payload.get("schema_version") != "1":
|
|
raise DistributionError("unsupported distribution manifest schema_version")
|
|
channel = _token(payload.get("channel"), "channel", maximum=32, pattern=MODULE_ID)
|
|
if expected_channel is not None and channel != expected_channel:
|
|
raise DistributionError(
|
|
f"distribution channel is {channel!r}, expected {expected_channel!r}"
|
|
)
|
|
if (
|
|
isinstance(payload.get("sequence"), bool)
|
|
or not isinstance(payload.get("sequence"), int)
|
|
or int(payload["sequence"]) < 1
|
|
):
|
|
raise DistributionError("distribution sequence must be a positive integer")
|
|
_token(payload.get("version"), "version", maximum=128, pattern=TOKEN)
|
|
issued = _datetime(payload.get("issued_at"), "issued_at")
|
|
expires = _datetime(payload.get("expires_at"), "expires_at")
|
|
current = (now or datetime.now(UTC)).astimezone(UTC)
|
|
if expires <= issued:
|
|
raise DistributionError("distribution expiry must be after issuance")
|
|
if issued > current:
|
|
raise DistributionError("distribution is not valid yet")
|
|
if expires <= current:
|
|
raise DistributionError("distribution manifest has expired")
|
|
if payload.get("revoked") is not False:
|
|
raise DistributionError("distribution manifest is revoked")
|
|
|
|
deployer = _object(payload.get("deployer"), "deployer")
|
|
_exact_keys(deployer, required={"url", "sha256"}, label="deployer")
|
|
_https_url(deployer.get("url"), "deployer.url")
|
|
_sha256(deployer.get("sha256"), "deployer.sha256")
|
|
|
|
images = _object(payload.get("images"), "images")
|
|
if set(images) != {"api", "web"}:
|
|
raise DistributionError("images must contain exactly api and web")
|
|
for name in ("api", "web"):
|
|
_validate_image(_object(images[name], f"images.{name}"), f"images.{name}")
|
|
|
|
dependencies = _object(payload.get("dependencies"), "dependencies")
|
|
if not dependencies:
|
|
raise DistributionError("dependencies must not be empty")
|
|
for name, reference in dependencies.items():
|
|
if MODULE_ID.fullmatch(str(name)) is None:
|
|
raise DistributionError(f"invalid dependency name: {name!r}")
|
|
_digest_image(reference, f"dependencies.{name}")
|
|
|
|
composition = _object(payload.get("composition"), "composition")
|
|
_exact_keys(
|
|
composition,
|
|
required={"sha256", "module_ids", "packages"},
|
|
label="composition",
|
|
)
|
|
_sha256(composition.get("sha256"), "composition.sha256")
|
|
module_ids = _string_array(composition.get("module_ids"), "module_ids")
|
|
if any(MODULE_ID.fullmatch(item) is None for item in module_ids):
|
|
raise DistributionError("composition.module_ids contains an invalid id")
|
|
packages = composition.get("packages")
|
|
if not isinstance(packages, list) or not packages:
|
|
raise DistributionError("composition.packages must be a non-empty array")
|
|
seen_packages: set[str] = set()
|
|
for index, item in enumerate(packages):
|
|
package = _object(item, f"composition.packages[{index}]")
|
|
_exact_keys(
|
|
package,
|
|
required={"name", "version", "wheel_sha256"},
|
|
label=f"composition.packages[{index}]",
|
|
)
|
|
name = _token(
|
|
package.get("name"),
|
|
f"composition.packages[{index}].name",
|
|
maximum=128,
|
|
pattern=re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$"),
|
|
)
|
|
if name in seen_packages:
|
|
raise DistributionError("composition contains duplicate packages")
|
|
seen_packages.add(name)
|
|
_token(
|
|
package.get("version"),
|
|
f"composition.packages[{index}].version",
|
|
maximum=128,
|
|
pattern=TOKEN,
|
|
)
|
|
_sha256(
|
|
package.get("wheel_sha256"),
|
|
f"composition.packages[{index}].wheel_sha256",
|
|
)
|
|
signatures = payload.get("signatures")
|
|
if not isinstance(signatures, list) or not signatures:
|
|
raise DistributionError("distribution manifest has no signatures")
|
|
seen_signatures: set[str] = set()
|
|
for index, item in enumerate(signatures):
|
|
signature = _object(item, f"signatures[{index}]")
|
|
_exact_keys(
|
|
signature,
|
|
required={"key_id", "algorithm", "value"},
|
|
label=f"signatures[{index}]",
|
|
)
|
|
key_id = _token(
|
|
signature.get("key_id"),
|
|
f"signatures[{index}].key_id",
|
|
maximum=128,
|
|
pattern=KEY_ID,
|
|
)
|
|
if key_id in seen_signatures:
|
|
raise DistributionError("distribution contains duplicate signatures")
|
|
seen_signatures.add(key_id)
|
|
if signature.get("algorithm") != "ed25519":
|
|
raise DistributionError("distribution signature algorithm must be ed25519")
|
|
_signature_bytes(signature.get("value"), f"signatures[{index}].value")
|
|
|
|
|
|
def verify_manifest(
|
|
payload: Mapping[str, Any],
|
|
keyring: Mapping[str, Any],
|
|
*,
|
|
expected_channel: str | None = None,
|
|
now: datetime | None = None,
|
|
openssl: str = "openssl",
|
|
) -> str:
|
|
current = (now or datetime.now(UTC)).astimezone(UTC)
|
|
validate_manifest(payload, expected_channel=expected_channel, now=current)
|
|
return verify_signed_document(
|
|
payload,
|
|
keyring,
|
|
purpose="govoplan-runtime-distribution",
|
|
label="distribution",
|
|
now=current,
|
|
openssl=openssl,
|
|
)
|
|
|
|
|
|
def verify_signed_document(
|
|
payload: Mapping[str, Any],
|
|
keyring: Mapping[str, Any],
|
|
*,
|
|
purpose: str,
|
|
label: str,
|
|
now: datetime,
|
|
openssl: str = "openssl",
|
|
) -> str:
|
|
keys = _trusted_keys(keyring, now=now, purpose=purpose, label=label)
|
|
signed = canonical_signed_payload(payload)
|
|
failures: list[str] = []
|
|
signatures = payload.get("signatures")
|
|
if not isinstance(signatures, list) or not signatures:
|
|
raise DistributionError(f"{label} has no signatures")
|
|
for index, raw in enumerate(signatures):
|
|
item = _object(raw, f"{label}.signatures[{index}]")
|
|
_exact_keys(
|
|
item,
|
|
required={"key_id", "algorithm", "value"},
|
|
label=f"{label}.signatures[{index}]",
|
|
)
|
|
key_id = str(item["key_id"])
|
|
if KEY_ID.fullmatch(key_id) is None or item.get("algorithm") != "ed25519":
|
|
raise DistributionError(f"{label} signature is invalid")
|
|
public_key = keys.get(key_id)
|
|
if public_key is None:
|
|
continue
|
|
signature = _signature_bytes(item["value"], "signature.value")
|
|
try:
|
|
_openssl_verify(
|
|
signed,
|
|
signature,
|
|
public_key,
|
|
openssl=openssl,
|
|
)
|
|
except DistributionError as exc:
|
|
failures.append(f"{key_id}: {exc}")
|
|
continue
|
|
return key_id
|
|
detail = (
|
|
"; ".join(failures) if failures else "no signature used an active trusted key"
|
|
)
|
|
raise DistributionError(f"{label} signature verification failed: {detail}")
|
|
|
|
|
|
def verify_manifest_binding(
|
|
payload: Mapping[str, Any],
|
|
*,
|
|
channel: str,
|
|
version: str,
|
|
api_image: str,
|
|
web_image: str,
|
|
enabled_modules: tuple[str, ...],
|
|
composition_sha256: str,
|
|
dependencies: Mapping[str, str],
|
|
) -> None:
|
|
if payload.get("channel") != channel or payload.get("version") != version:
|
|
raise DistributionError(
|
|
"stored manifest does not match release channel/version"
|
|
)
|
|
images = _object(payload.get("images"), "images")
|
|
if _object(images.get("api"), "images.api").get("index") != api_image:
|
|
raise DistributionError("stored manifest does not match API image")
|
|
if _object(images.get("web"), "images.web").get("index") != web_image:
|
|
raise DistributionError("stored manifest does not match Web image")
|
|
composition = _object(payload.get("composition"), "composition")
|
|
if composition.get("sha256") != composition_sha256:
|
|
raise DistributionError("stored manifest composition digest does not match")
|
|
available_modules = set(_string_array(composition.get("module_ids"), "module_ids"))
|
|
missing = sorted(set(enabled_modules) - available_modules)
|
|
if missing:
|
|
raise DistributionError(
|
|
"enabled modules are absent from runtime composition: " + ", ".join(missing)
|
|
)
|
|
manifest_dependencies = _object(payload.get("dependencies"), "dependencies")
|
|
for name, reference in dependencies.items():
|
|
if manifest_dependencies.get(name) != reference:
|
|
raise DistributionError(
|
|
f"stored manifest does not match dependency image {name!r}"
|
|
)
|
|
|
|
|
|
def verify_offline_image_index(
|
|
index: Mapping[str, Any],
|
|
*,
|
|
root: Path,
|
|
expected_references: tuple[str, ...],
|
|
) -> tuple[Path, ...]:
|
|
_exact_keys(index, required={"schema_version", "images"}, label="offline index")
|
|
if index.get("schema_version") != "1":
|
|
raise DistributionError("unsupported offline image index schema")
|
|
images = index.get("images")
|
|
if not isinstance(images, list):
|
|
raise DistributionError("offline image index images must be an array")
|
|
references: dict[str, Path] = {}
|
|
for item in images:
|
|
value = _object(item, "offline image")
|
|
_exact_keys(
|
|
value,
|
|
required={"reference", "archive", "sha256"},
|
|
label="offline image",
|
|
)
|
|
reference = _digest_image(value.get("reference"), "offline image reference")
|
|
archive_value = value.get("archive")
|
|
if not isinstance(archive_value, str) or not archive_value:
|
|
raise DistributionError("offline image archive must be a relative path")
|
|
archive_relative = Path(archive_value)
|
|
if archive_relative.is_absolute() or ".." in archive_relative.parts:
|
|
raise DistributionError("offline image archive must stay inside its bundle")
|
|
archive = root / archive_relative
|
|
if reference in references:
|
|
raise DistributionError("offline image index contains duplicate references")
|
|
if _sha256_regular_file(
|
|
archive, maximum_bytes=MAX_OFFLINE_IMAGE_BYTES
|
|
) != _sha256(value.get("sha256"), "offline image sha256"):
|
|
raise DistributionError(f"offline image archive digest mismatch: {archive}")
|
|
references[reference] = archive
|
|
missing = sorted(set(expected_references) - set(references))
|
|
if missing:
|
|
raise DistributionError(
|
|
"offline image bundle is incomplete: " + ", ".join(missing)
|
|
)
|
|
return tuple(references[item] for item in expected_references)
|
|
|
|
|
|
def file_sha256(path: Path, *, maximum_bytes: int = MAX_MANIFEST_BYTES) -> str:
|
|
return _sha256_regular_file(path, maximum_bytes=maximum_bytes)
|
|
|
|
|
|
def _validate_image(value: Mapping[str, Any], label: str) -> None:
|
|
_exact_keys(
|
|
value,
|
|
required={"index", "platforms", "sbom", "provenance"},
|
|
label=label,
|
|
)
|
|
_digest_image(value.get("index"), f"{label}.index")
|
|
platforms = _object(value.get("platforms"), f"{label}.platforms")
|
|
if set(platforms) != set(PLATFORMS):
|
|
raise DistributionError(f"{label}.platforms must cover amd64 and arm64")
|
|
for platform, reference in platforms.items():
|
|
_digest_image(reference, f"{label}.platforms.{platform}")
|
|
_validate_artifact(_object(value.get("sbom"), f"{label}.sbom"), f"{label}.sbom")
|
|
_validate_artifact(
|
|
_object(value.get("provenance"), f"{label}.provenance"),
|
|
f"{label}.provenance",
|
|
)
|
|
|
|
|
|
def _validate_artifact(value: Mapping[str, Any], label: str) -> None:
|
|
_exact_keys(value, required={"url", "sha256"}, label=label)
|
|
_https_url(value.get("url"), f"{label}.url")
|
|
_sha256(value.get("sha256"), f"{label}.sha256")
|
|
|
|
|
|
def _trusted_keys(
|
|
keyring: Mapping[str, Any],
|
|
*,
|
|
now: datetime,
|
|
purpose: str,
|
|
label: str,
|
|
) -> dict[str, str]:
|
|
_exact_keys(
|
|
keyring,
|
|
required={"schema_version", "purpose", "keys"},
|
|
label=f"{label} keyring",
|
|
)
|
|
if keyring.get("schema_version") != "1":
|
|
raise DistributionError(f"unsupported {label} keyring schema_version")
|
|
if keyring.get("purpose") != purpose:
|
|
raise DistributionError(f"{label} keyring has the wrong purpose")
|
|
values = keyring.get("keys")
|
|
if not isinstance(values, list) or not values:
|
|
raise DistributionError(f"{label} keyring contains no keys")
|
|
trusted: dict[str, str] = {}
|
|
for index, item in enumerate(values):
|
|
key = _object(item, f"keyring.keys[{index}]")
|
|
_exact_keys(
|
|
key,
|
|
required={
|
|
"key_id",
|
|
"algorithm",
|
|
"status",
|
|
"public_key_pem",
|
|
"not_before",
|
|
"expires_at",
|
|
},
|
|
label=f"keyring.keys[{index}]",
|
|
)
|
|
key_id = _token(
|
|
key.get("key_id"),
|
|
f"keyring.keys[{index}].key_id",
|
|
maximum=128,
|
|
pattern=KEY_ID,
|
|
)
|
|
if key_id in trusted:
|
|
raise DistributionError(f"{label} keyring contains duplicate key ids")
|
|
if key.get("algorithm") != "ed25519":
|
|
raise DistributionError(f"{label} key must use ed25519")
|
|
if key.get("status") not in {"active", "retired", "revoked"}:
|
|
raise DistributionError(f"{label} key has an invalid status")
|
|
not_before = _datetime(key.get("not_before"), "key.not_before")
|
|
expires = _datetime(key.get("expires_at"), "key.expires_at")
|
|
public_key = key.get("public_key_pem")
|
|
if (
|
|
not isinstance(public_key, str)
|
|
or len(public_key.encode("utf-8")) > 8192
|
|
or "BEGIN PUBLIC KEY" not in public_key
|
|
):
|
|
raise DistributionError(f"{label} key has an invalid public key")
|
|
if key.get("status") == "active" and not_before <= now < expires:
|
|
trusted[key_id] = public_key
|
|
if not trusted:
|
|
raise DistributionError(f"{label} keyring has no currently active keys")
|
|
return trusted
|
|
|
|
|
|
def _openssl_verify(
|
|
payload: bytes,
|
|
signature: bytes,
|
|
public_key: str,
|
|
*,
|
|
openssl: str,
|
|
) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="govoplan-distribution-verify-") as value:
|
|
root = Path(value)
|
|
payload_path = root / "payload.json"
|
|
signature_path = root / "signature.bin"
|
|
key_path = root / "public.pem"
|
|
payload_path.write_bytes(payload)
|
|
signature_path.write_bytes(signature)
|
|
key_path.write_text(public_key, encoding="utf-8")
|
|
try:
|
|
completed = subprocess.run(
|
|
[
|
|
openssl,
|
|
"pkeyutl",
|
|
"-verify",
|
|
"-pubin",
|
|
"-inkey",
|
|
str(key_path),
|
|
"-rawin",
|
|
"-in",
|
|
str(payload_path),
|
|
"-sigfile",
|
|
str(signature_path),
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise DistributionError("OpenSSL Ed25519 verifier is unavailable") from exc
|
|
if completed.returncode != 0:
|
|
raise DistributionError("Ed25519 signature is invalid")
|
|
|
|
|
|
def _signature_bytes(value: object, label: str) -> bytes:
|
|
if not isinstance(value, str) or len(value) > 256:
|
|
raise DistributionError(f"{label} is invalid")
|
|
try:
|
|
decoded = base64.b64decode(value, validate=True)
|
|
except (ValueError, base64.binascii.Error) as exc:
|
|
raise DistributionError(f"{label} is not valid base64") from exc
|
|
if len(decoded) != 64:
|
|
raise DistributionError(f"{label} is not an Ed25519 signature")
|
|
return decoded
|
|
|
|
|
|
def _require_public_host(hostname: str) -> None:
|
|
try:
|
|
addresses = {
|
|
value[4][0]
|
|
for value in socket.getaddrinfo(hostname, 443, type=socket.SOCK_STREAM)
|
|
}
|
|
except OSError as exc:
|
|
raise DistributionError(
|
|
f"distribution host cannot be resolved: {hostname}"
|
|
) from exc
|
|
if not addresses:
|
|
raise DistributionError("distribution host resolved to no addresses")
|
|
for value in addresses:
|
|
address = ipaddress.ip_address(value)
|
|
if not address.is_global:
|
|
raise DistributionError(
|
|
"distribution host resolves to a non-public address"
|
|
)
|
|
|
|
|
|
def _sha256_regular_file(path: Path, *, maximum_bytes: int) -> str:
|
|
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
|
|
try:
|
|
descriptor = os.open(path, flags)
|
|
except OSError as exc:
|
|
raise DistributionError(f"cannot open immutable artifact: {path}") from exc
|
|
digest = hashlib.sha256()
|
|
try:
|
|
opened = os.fstat(descriptor)
|
|
if not stat.S_ISREG(opened.st_mode) or opened.st_size > maximum_bytes:
|
|
raise DistributionError(
|
|
f"immutable artifact is invalid or too large: {path}"
|
|
)
|
|
while True:
|
|
chunk = os.read(descriptor, 1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
digest.update(chunk)
|
|
final = os.fstat(descriptor)
|
|
if (opened.st_dev, opened.st_ino, opened.st_size, opened.st_mtime_ns) != (
|
|
final.st_dev,
|
|
final.st_ino,
|
|
final.st_size,
|
|
final.st_mtime_ns,
|
|
):
|
|
raise DistributionError(f"immutable artifact changed while read: {path}")
|
|
finally:
|
|
os.close(descriptor)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _object(value: object, label: str) -> dict[str, Any]:
|
|
if not isinstance(value, dict) or not all(isinstance(key, str) for key in value):
|
|
raise DistributionError(f"{label} must be an object")
|
|
return value
|
|
|
|
|
|
def _exact_keys(
|
|
value: Mapping[str, Any],
|
|
*,
|
|
required: set[str],
|
|
label: str,
|
|
) -> None:
|
|
missing = sorted(required - set(value))
|
|
extra = sorted(set(value) - required)
|
|
if missing or extra:
|
|
detail = []
|
|
if missing:
|
|
detail.append("missing " + ", ".join(missing))
|
|
if extra:
|
|
detail.append("unknown " + ", ".join(extra))
|
|
raise DistributionError(f"{label} has invalid fields: {'; '.join(detail)}")
|
|
|
|
|
|
def _token(
|
|
value: object,
|
|
label: str,
|
|
*,
|
|
maximum: int,
|
|
pattern: re.Pattern[str],
|
|
) -> str:
|
|
if (
|
|
not isinstance(value, str)
|
|
or len(value) > maximum
|
|
or pattern.fullmatch(value) is None
|
|
):
|
|
raise DistributionError(f"{label} is invalid")
|
|
return value
|
|
|
|
|
|
def _datetime(value: object, label: str) -> datetime:
|
|
if not isinstance(value, str) or len(value) > 64:
|
|
raise DistributionError(f"{label} must be an RFC3339 timestamp")
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise DistributionError(f"{label} must be an RFC3339 timestamp") from exc
|
|
if parsed.tzinfo is None:
|
|
raise DistributionError(f"{label} must include a timezone")
|
|
return parsed.astimezone(UTC)
|
|
|
|
|
|
def _sha256(value: object, label: str) -> str:
|
|
if not isinstance(value, str) or SHA256.fullmatch(value) is None:
|
|
raise DistributionError(f"{label} must be a lowercase SHA-256 digest")
|
|
return value
|
|
|
|
|
|
def _digest_image(value: object, label: str) -> str:
|
|
if (
|
|
not isinstance(value, str)
|
|
or len(value) > 300
|
|
or DIGEST_IMAGE.fullmatch(value) is None
|
|
):
|
|
raise DistributionError(f"{label} must be an OCI image pinned by sha256")
|
|
return value
|
|
|
|
|
|
def _https_url(value: object, label: str) -> str:
|
|
if not isinstance(value, str) or len(value) > 2048:
|
|
raise DistributionError(f"{label} must be an HTTPS URL")
|
|
parsed = urlsplit(value)
|
|
if (
|
|
parsed.scheme != "https"
|
|
or not parsed.netloc
|
|
or parsed.username
|
|
or parsed.password
|
|
):
|
|
raise DistributionError(f"{label} must be an HTTPS URL without credentials")
|
|
return value
|
|
|
|
|
|
def _string_array(value: object, label: str) -> tuple[str, ...]:
|
|
if (
|
|
not isinstance(value, list)
|
|
or len(value) > 1024
|
|
or any(not isinstance(item, str) for item in value)
|
|
or len(set(value)) != len(value)
|
|
):
|
|
raise DistributionError(f"{label} must be an array of unique strings")
|
|
return tuple(value)
|