128 lines
6.3 KiB
Python
128 lines
6.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Fail closed before publishing a release that supports legacy adoption.
|
|
|
|
The inventory contains public version and signed-manifest hashes only. The
|
|
installed production verifier authenticates the downloaded manifest bytes;
|
|
the frontend identity then comes from that signed document, never inventory.
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
import public_update_prepare as production
|
|
|
|
BASE = 'https://source.archipelago-foundation.org/lfg2025/archy/releases/download'
|
|
BASELINE = ('1.9.1-alpha', '1.9.0-alpha', *(f'1.8.{n}-alpha' for n in range(22, 12, -1)))
|
|
VERSION = re.compile(r'[0-9]+\.[0-9]+\.[0-9]+(?:-[A-Za-z0-9.]+)?\Z')
|
|
SHA = re.compile(r'[0-9a-f]{64}\Z')
|
|
|
|
|
|
def strict_pairs(items):
|
|
return production.pairs(items)
|
|
|
|
|
|
def public_get(url, expected_sha, expected_size=None):
|
|
"""Hash the full HTTPS body without retaining another large archive."""
|
|
parsed = urllib.parse.urlsplit(url)
|
|
if (parsed.scheme != 'https' or parsed.netloc != 'source.archipelago-foundation.org'
|
|
or not parsed.path.startswith('/lfg2025/archy/releases/download/')
|
|
or parsed.query or parsed.fragment):
|
|
raise ValueError('legacy asset URL escaped the fixed public origin')
|
|
if not SHA.fullmatch(expected_sha) or (expected_size is not None and
|
|
(type(expected_size) is not int or not 0 < expected_size <= production.MAX_COMPONENT)):
|
|
raise ValueError('invalid signed legacy asset identity')
|
|
maximum = expected_size if expected_size is not None else production.MAX_DOCUMENT
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
digest = hashlib.sha256()
|
|
count = 0
|
|
deadline = time.monotonic() + 180
|
|
with opener.open(url, timeout=30) as response:
|
|
if response.status != 200:
|
|
raise ValueError('public legacy asset did not return HTTP 200')
|
|
final = urllib.parse.urlsplit(response.geturl())
|
|
if final.scheme != 'https' or final.netloc != parsed.netloc:
|
|
raise ValueError('legacy asset redirect escaped the fixed public origin')
|
|
chunks = [] if expected_size is None else None
|
|
while block := response.read(1024 * 1024):
|
|
if time.monotonic() > deadline:
|
|
raise TimeoutError('public legacy asset download deadline exceeded')
|
|
count += len(block)
|
|
if count > maximum:
|
|
raise ValueError('legacy asset exceeded signed size')
|
|
digest.update(block)
|
|
if chunks is not None:
|
|
chunks.append(block)
|
|
if (expected_size is not None and count != expected_size) or digest.hexdigest() != expected_sha:
|
|
raise ValueError('public legacy asset differs from signed identity')
|
|
return b''.join(chunks) if chunks is not None else None
|
|
|
|
|
|
def check(version, inventory, *, get=public_get, authenticate=production.authenticate):
|
|
if not VERSION.fullmatch(version):
|
|
raise ValueError('invalid target release version')
|
|
document = json.loads(Path(inventory).read_text(), object_pairs_hook=strict_pairs)
|
|
if document.get('schema') != 1 or document.get('target_version') != version:
|
|
raise ValueError('legacy compatibility inventory targets a different release')
|
|
entries = document.get('supported')
|
|
if not isinstance(entries, list) or len(entries) < len(BASELINE):
|
|
raise ValueError('legacy compatibility inventory omits baseline support')
|
|
versions = [entry.get('version') for entry in entries if isinstance(entry, dict)]
|
|
if len(versions) != len(entries) or len(set(versions)) != len(entries) or not set(BASELINE).issubset(versions):
|
|
raise ValueError('legacy compatibility inventory omits or duplicates baseline support')
|
|
for entry in entries:
|
|
old_version, manifest_sha = entry['version'], entry['manifest_sha256']
|
|
if not VERSION.fullmatch(old_version) or not SHA.fullmatch(manifest_sha):
|
|
raise ValueError('invalid historical release identity')
|
|
alias = f'{BASE}/v{version}/legacy-manifest-{old_version}.json'
|
|
raw = get(alias, manifest_sha)
|
|
authenticate(raw) # fixed installed verifier and release root; no CLI trust override
|
|
old = json.loads(raw, object_pairs_hook=strict_pairs)
|
|
if old.get('version') != old_version or old.get('signed_by') != production.RELEASE_ROOT:
|
|
raise ValueError('historical alias is not the declared signed release')
|
|
components = old.get('components')
|
|
if not isinstance(components, list):
|
|
raise ValueError('historical signed component list missing')
|
|
frontend = [c for c in components if isinstance(c, dict) and
|
|
c.get('name') == f'archipelago-frontend-{old_version}.tar.gz']
|
|
binary = [c for c in components if isinstance(c, dict) and c.get('name') == 'archipelago']
|
|
if len(frontend) != 1 or len(binary) != 1:
|
|
raise ValueError('historical signed code pair missing')
|
|
item = frontend[0]
|
|
signed_url, size, sha = item.get('download_url'), item.get('size_bytes'), item.get('sha256')
|
|
if not isinstance(signed_url, str) or not isinstance(sha, str):
|
|
raise ValueError('historical signed frontend identity missing')
|
|
fallback = f'{BASE}/v{version}/legacy-frontend-{old_version}.tar.gz'
|
|
# The alias is mandatory even when the historical URL still works: a
|
|
# release asset omission must fail before the fleet sees this update.
|
|
get(fallback, sha, size)
|
|
original_available = True
|
|
try:
|
|
get(signed_url, sha, size)
|
|
except urllib.error.HTTPError as error:
|
|
if error.code not in (404, 410):
|
|
raise
|
|
original_available = False
|
|
print(f'PASS legacy {old_version}: signed manifest alias and full frontend fallback; '
|
|
f'original {"verified" if original_available else "unavailable (404/410)"}')
|
|
return len(entries)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('target_version')
|
|
parser.add_argument('inventory', type=Path)
|
|
args = parser.parse_args()
|
|
count = check(args.target_version, args.inventory)
|
|
print(f'PASS {count} historical release compatibility assets for v{args.target_version}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|