TerminalBasic/.gitea/scripts/release.py

#!/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 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 = tag.removeprefix('v')
    dist.require(tag == 'v' + version and 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 = tag.removeprefix('v')
    dist.require(tag == 'v' + version, 'Versionstag erforderlich')
    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(args.tag.removeprefix('v'))
        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()