226 lines
9.6 KiB
Python
226 lines
9.6 KiB
Python
"""Update timestamp and release manifests using only Python 3.10+.
|
|
|
|
Without --write, calculate changes without saving them. Never commit or push.
|
|
PowerShell statements in manifests are not executed.
|
|
"""
|
|
|
|
import argparse
|
|
from datetime import timezone
|
|
from email.utils import parsedate_to_datetime
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urldefrag, urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
BUCKET = Path(__file__).resolve().parents[1] / 'bucket'
|
|
TIMESTAMP_REGEX = r'(?<version>\d{8}T\d{6}Z)'
|
|
|
|
|
|
def open_url(url, method='GET', headers=None):
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password:
|
|
raise ValueError('Nur HTTPS-URLs ohne eingebettete Zugangsdaten sind erlaubt.')
|
|
return urlopen(Request(url, method=method, headers={
|
|
'User-Agent': 'Medelexis-Scoop-Updater', 'Cache-Control': 'no-cache',
|
|
'Accept-Encoding': 'identity', **(headers or {})}), timeout=180)
|
|
|
|
|
|
def metadata(headers):
|
|
return {key: headers.get(key, '') for key in ('Last-Modified', 'ETag')}
|
|
|
|
|
|
def read_head(url):
|
|
with open_url(url, 'HEAD') as response:
|
|
return metadata(response.headers)
|
|
|
|
|
|
def release_version(checkver):
|
|
if checkver.get('jsonpath') != '$.tag_name' or not checkver.get('regex'):
|
|
raise ValueError('Unterstützt werden Last-Modified oder jsonpath $.tag_name mit regex.')
|
|
with open_url(checkver['url']) as response:
|
|
tag = json.load(response)['tag_name']
|
|
pattern = checkver['regex'].replace('(?<version>', '(?P<version>')
|
|
match = re.search(pattern, tag)
|
|
if not match:
|
|
raise ValueError('Release-Tag passt nicht zum Versionsmuster.')
|
|
version = match.groupdict().get('version') or match.group(1)
|
|
if not re.fullmatch(r'[0-9A-Za-z][0-9A-Za-z._+-]*', version):
|
|
raise ValueError('Ungültige Release-Version.')
|
|
return version
|
|
|
|
|
|
def download_file(url, observed, target):
|
|
conditions = {}
|
|
if observed['ETag'] and not observed['ETag'].startswith('W/'):
|
|
conditions['If-Match'] = observed['ETag']
|
|
elif observed['Last-Modified']:
|
|
conditions['If-Unmodified-Since'] = observed['Last-Modified']
|
|
size = 0
|
|
prefix = b''
|
|
with open_url(url, headers=conditions) as response:
|
|
if metadata(response.headers) != observed:
|
|
raise ValueError('Datei hat sich zwischen HEAD und GET geändert; erneut versuchen.')
|
|
if response.headers.get('Content-Type', '').split(';')[0].lower() in (
|
|
'text/html', 'application/json'):
|
|
raise ValueError('Download liefert HTML/JSON statt einer Programmdatei.')
|
|
expected = response.headers.get('Content-Length')
|
|
total = int(expected) if expected is not None else None
|
|
last_report = time.monotonic()
|
|
if total is not None:
|
|
print(f' Download gestartet: {total / 1048576:.1f} MiB.', flush=True)
|
|
else:
|
|
print(' Download gestartet; Gesamtgrösse unbekannt.', flush=True)
|
|
with target.open('wb') as output:
|
|
while chunk := response.read(1024 * 1024):
|
|
if not size:
|
|
prefix = chunk[:4]
|
|
output.write(chunk)
|
|
size += len(chunk)
|
|
now = time.monotonic()
|
|
if now - last_report >= 5:
|
|
progress = f'{size / 1048576:.1f} MiB'
|
|
if total:
|
|
progress += f' / {total / 1048576:.1f} MiB ({size / total:.0%})'
|
|
print(f' Heruntergeladen: {progress}', flush=True)
|
|
last_report = now
|
|
if not size or (expected is not None and size != int(expected)):
|
|
raise ValueError('Leerer oder unvollständiger Download.')
|
|
suffix = Path(urlsplit(url).path).suffix.lower()
|
|
if suffix == '.exe' and not prefix.startswith(b'MZ'):
|
|
raise ValueError('Download hat keine Windows-EXE-Signatur.')
|
|
if suffix == '.zip' and not prefix.startswith(b'PK'):
|
|
raise ValueError('Download hat keine ZIP-Signatur.')
|
|
print(f' Download abgeschlossen: {size / 1048576:.1f} MiB.', flush=True)
|
|
return size
|
|
|
|
|
|
def hash_file(path):
|
|
total = path.stat().st_size
|
|
print(f' SHA-256-Berechnung gestartet: {total / 1048576:.1f} MiB.', flush=True)
|
|
digest = hashlib.sha256()
|
|
size = 0
|
|
last_report = time.monotonic()
|
|
with path.open('rb') as source:
|
|
while chunk := source.read(1024 * 1024):
|
|
digest.update(chunk)
|
|
size += len(chunk)
|
|
now = time.monotonic()
|
|
if now - last_report >= 5:
|
|
print(f' Gehasht: {size / 1048576:.1f} / {total / 1048576:.1f} MiB '
|
|
f'({size / total:.0%})', flush=True)
|
|
last_report = now
|
|
print(' SHA-256-Berechnung abgeschlossen.', flush=True)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def download_hash(url, observed):
|
|
with tempfile.TemporaryDirectory(prefix='scoop-update-') as directory:
|
|
target = Path(directory) / 'download.bin'
|
|
size = download_file(url, observed, target)
|
|
return hash_file(target), size
|
|
|
|
|
|
def save_manifest(path, manifest):
|
|
original = path.read_bytes()
|
|
newline = '\r\n' if b'\r\n' in original else '\n'
|
|
content = json.dumps(manifest, ensure_ascii=False, indent=4) + '\n'
|
|
temporary = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(dir=path.parent, suffix='.tmp', delete=False) as output:
|
|
temporary = Path(output.name)
|
|
output.write(content.replace('\n', newline).encode('utf-8'))
|
|
temporary.replace(path)
|
|
finally:
|
|
if temporary is not None:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def update_manifest(path, write=False):
|
|
manifest = json.loads(path.read_text(encoding='utf-8-sig'))
|
|
checkver, autoupdate = manifest.get('checkver'), manifest.get('autoupdate')
|
|
if not checkver or not autoupdate:
|
|
print(f'{path.stem}: kein automatisches Update konfiguriert; übersprungen.')
|
|
return False
|
|
if not isinstance(checkver, dict) or not isinstance(autoupdate.get('url'), str):
|
|
raise ValueError('Diese checkver/autoupdate-Konfiguration wird noch nicht unterstützt.')
|
|
|
|
template = autoupdate['url']
|
|
observed = None
|
|
if checkver.get('regex') == TIMESTAMP_REGEX:
|
|
source = urldefrag(manifest['url'])[0]
|
|
print(f'{path.stem}: HEAD-Anfrage an {source} ...', flush=True)
|
|
observed = read_head(source)
|
|
if not observed['Last-Modified']:
|
|
raise ValueError('Last-Modified-Header fehlt.')
|
|
date = parsedate_to_datetime(observed['Last-Modified'])
|
|
if date.tzinfo is None:
|
|
raise ValueError('Last-Modified enthält keine Zeitzone.')
|
|
version = date.astimezone(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
|
if urldefrag(template.replace('$version', version))[0] != source:
|
|
raise ValueError('Zeitstempel-Updates benötigen eine feste Download-URL.')
|
|
elif 'script' not in checkver:
|
|
print(f'{path.stem}: Release-Version abfragen: {checkver.get("url", "")} ...', flush=True)
|
|
version = release_version(checkver)
|
|
else:
|
|
raise ValueError('Dieses checkver-Skript wird nicht ausgeführt oder interpretiert.')
|
|
|
|
if version == manifest['version']:
|
|
print(f'{path.stem}: {version} unverändert; kein Download.')
|
|
return False
|
|
url = template.replace('$version', version)
|
|
if '$' in url:
|
|
raise ValueError('Unterstützt wird nur der Platzhalter $version.')
|
|
source = urldefrag(url)[0]
|
|
if observed is None:
|
|
print(f'{path.stem}: HEAD-Anfrage an {source} ...', flush=True)
|
|
observed = read_head(source)
|
|
print(f'{path.stem}: neue Version {version}; GET-Download ...', flush=True)
|
|
print(' Für den SHA-256-Hash muss die gesamte Datei heruntergeladen werden. '
|
|
'Danach wird der Hash berechnet. Bei grossen Dateien kann dies mehrere Minuten dauern, '
|
|
'auch im Probelauf.', flush=True)
|
|
digest, size = download_hash(source, observed)
|
|
print(f'{path.stem}: {manifest["version"]} -> {version}; {size} Bytes; SHA-256 {digest}')
|
|
manifest.update(version=version, url=url, hash=digest)
|
|
if write:
|
|
save_manifest(path, manifest)
|
|
print(f'{path.stem}: Manifest gespeichert.')
|
|
else:
|
|
print(f'{path.stem}: Probelauf; Manifest bleibt unverändert.')
|
|
return True
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--bucket', type=Path, default=BUCKET)
|
|
parser.add_argument('--app', action='append', help='Manifestname; mehrfach möglich, sonst alle.')
|
|
parser.add_argument('--write', action='store_true', help='Geprüfte Änderungen speichern.')
|
|
args = parser.parse_args(argv)
|
|
if not args.bucket.is_dir():
|
|
parser.error('Bucket-Verzeichnis existiert nicht.')
|
|
if args.app and any(not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]*', app) for app in args.app):
|
|
parser.error('--app erwartet einen Manifestnamen ohne Pfad oder .json-Endung.')
|
|
paths = ([args.bucket / f'{app}.json' for app in args.app] if args.app else
|
|
sorted(args.bucket.glob('*.json')))
|
|
errors = 0
|
|
for path in paths:
|
|
try:
|
|
update_manifest(path, args.write)
|
|
except HTTPError as error:
|
|
print(f'{path.stem}: HTTP {error.code}; Manifest unverändert.', file=sys.stderr)
|
|
errors += 1
|
|
except (ValueError, KeyError, TypeError, OSError, URLError) as error:
|
|
print(f'{path.stem}: {error}; Manifest unverändert.', file=sys.stderr)
|
|
errors += 1
|
|
return int(errors > 0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|