#!/usr/bin/env python3 """Gitea-Entwurf befüllen; erst nach exakten Paket-/Zielnachweisen freigeben.""" import argparse import hashlib import json import os from pathlib import Path import re from urllib.error import HTTPError from urllib.parse import quote, urlparse from urllib.request import Request, HTTPRedirectHandler, build_opener import distribution as dist class SafeRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): redirected = super().redirect_request(req, fp, code, msg, headers, newurl) if redirected and urlparse(newurl).netloc != urlparse(req.full_url).netloc: redirected.remove_header('Authorization') return redirected class Gitea: def __init__(self): self.base = os.environ['GITEA_SERVER_URL'].rstrip('/') dist.require(self.base == 'https://git.rfc1437.de', 'Release nur am bestehenden Origin') dist.require(os.environ['GITEA_REPOSITORY'] == 'hugo/TerminalBasic', 'Falsches Release-Repository') self.api = self.base + '/api/v1/repos/hugo/TerminalBasic' self.token = os.environ['TB_RELEASE_TOKEN'] self.open = build_opener(SafeRedirect()).open def request(self, method, path, data=None): raw = isinstance(data, bytes) headers = {'Authorization': 'token ' + self.token, 'Content-Type': 'application/octet-stream' if raw else 'application/json'} payload = data if raw or data is None else json.dumps(data).encode() with self.open(Request(self.api + path, data=payload, headers=headers, method=method), timeout=120) as response: return json.load(response) def download(self, asset): url = asset['browser_download_url'] dist.require(urlparse(url).netloc == urlparse(self.base).netloc, 'Fremder Assetserver') with self.open(Request(url, headers={'Authorization': 'token ' + self.token}), timeout=120) as response: return response.read() def release_version(tag): dist.require(re.fullmatch(r'v(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)', tag), 'Release-Tag muss vX.Y.Z entsprechen') return tag[1:] def expected(version): return [n for t in dist.TARGETS for n in (dist.name(version, t), dist.name(version, t) + '.sha256')] + ['acceptance.json'] def stage(api, directory, tag, revision): version = release_version(tag) dist.require(re.fullmatch('[0-9a-f]{40}', revision), 'Tag/Revision ungültig') dist.require(api.request('GET', '/tags/' + quote(tag, safe=''))['commit']['sha'] == revision, 'Tag zeigt auf andere Revision') names = expected(version) for target in dist.TARGETS: path = directory / dist.name(version, target) manifest = dist.verify(path, target, dist.ROOT / 'target/release/tb-template') dist.require(manifest['revision'] == revision and not manifest['source_dirty'], 'Paketrevision falsch/uncommitted') dist.require(path.with_name(path.name + '.sha256').read_text() == f'{dist.digest(path)} {path.name}\n', 'Paketprüfsumme falsch') acceptance = json.loads((directory / 'acceptance.json').read_text()) dist.require(acceptance.get('revision') == revision and acceptance.get('passed') is True, 'Grundabnahme fehlt') try: release = api.request('GET', '/releases/tags/' + quote(tag, safe='')) except HTTPError as error: if error.code != 404: raise release = api.request('POST', '/releases', dict(tag_name=tag, target_commitish=revision, name='Terminal Basic ' + version, draft=True, body='Vierzielpakete im Entwurf. Zielnachweise fehlen noch; keine Freigabe.')) dist.require(release['draft'], 'Bereits veröffentlichten Release nicht verändern') # Wiederholungen dürfen identische Uploads ergänzen, keine vorhandenen Assets ersetzen. existing = {a['name']: a for a in release.get('assets', [])} dist.require(set(existing) <= set(names), 'Unerwartete Assets im Entwurf') for name in names: data = (directory / name).read_bytes() if name in existing: dist.require(api.download(existing[name]) == data, f'Abweichendes vorhandenes Asset: {name}') else: api.request('POST', f'/releases/{release["id"]}/assets?name=' + quote(name), data) verify_uploads(api, release['id'], directory, names) return release['id'] def verify_uploads(api, release_id, directory, names): release = api.request('GET', f'/releases/{release_id}') assets = release.get('assets', []) dist.require(len(assets) == len(names) and {a['name'] for a in assets} == set(names), 'Unvollständiger Asset-Upload') for asset in assets: dist.require(hashlib.sha256(api.download(asset)).hexdigest() == dist.digest(directory / asset['name']), 'Uploadprüfsumme falsch') return release def publish(api, directory, evidence, tag, revision): version = release_version(tag) dist.require(api.request('GET', '/tags/' + quote(tag, safe=''))['commit']['sha'] == revision, 'Tag zeigt auf andere Revision') dist.gate(directory, evidence, revision, version) release = api.request('GET', '/releases/tags/' + quote(tag, safe='')) verify_uploads(api, release['id'], directory, expected(version)) dist.require(release['draft'], 'Bereits freigegebenen Release nicht verändern') api.request('PATCH', f'/releases/{release["id"]}', dict(draft=False, body=f'Geprüfte Vierzielpakete für Revision {revision}. Zugehörige Zielnachweise: {os.environ.get("TB_EVIDENCE_REVISION", "lokaler Nachweissatz")}.')) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('command', choices=['stage', 'download', 'publish']) parser.add_argument('--packages', type=Path, required=True) parser.add_argument('--tag', required=True); parser.add_argument('--revision', required=True) parser.add_argument('--evidence', type=Path) args = parser.parse_args() api = Gitea() if args.command == 'stage': stage(api, args.packages, args.tag, args.revision) elif args.command == 'publish': dist.require(args.evidence, '--evidence erforderlich') publish(api, args.packages, args.evidence, args.tag, args.revision) else: args.packages.mkdir(parents=True) # Nie lokale Kandidaten überschreiben. release = api.request('GET', '/releases/tags/' + quote(args.tag, safe='')) names = expected(release_version(args.tag)) dist.require(release['draft'], 'Nur bestehende Entwürfe freigeben') assets = release.get('assets', []) dist.require(len(assets) == len(names) and {a['name'] for a in assets} == set(names), 'Pflichtassets fehlen') for asset in assets: (args.packages / asset['name']).write_bytes(api.download(asset)) if __name__ == '__main__': main()