platform-ci/publisher_frozen/gitops_writeback.py

"""Pin verified, published image tags in one enrolled GitOps kustomization with ONE compare-and-swap commit.

Runs only after every image of a release was verified and published (and read back from the registry). Following the
cluster-gitops rules, automation changes only the existing `newTag` of existing `images[]` entries: the entry must
already exist, already point at the enrolled registry image through `newName`, and carry no `digest`. The target
repository, branch, file and entry names come from the central enrollment; nothing from an artifact or run chooses
them. One commit updates all images or none, so a lost race or refusal writes nothing and a partial pin cannot exist.
This module never claims the new tags are deployed.
"""

import base64
import hashlib
import json
from pathlib import Path
import re
import urllib.error

from promote_container import write_receipt
from release_candidate import inspect_candidate

REGISTRY = 'git.kof22.com/'
DIGEST = re.compile(r'sha256:[0-9a-f]{64}')
TAG = re.compile(r'ci-[0-9a-f]{12}-run-([1-9][0-9]*)')
NAME = re.compile(r'[a-z0-9][a-z0-9.-]*(:[0-9]+)?(/[a-z0-9][a-z0-9._-]*)+')
ITEM = re.compile(r'^( *)- name: (\S+) *$')
KEY = re.compile(r'^( *)([A-Za-z]+): (\S+) *$')
MAX_FILE_BYTES = 256 * 1024
CONFLICT = (409, 422)


def _entries(lines):
    starts = [i for i, line in enumerate(lines) if line.rstrip() == 'images:']
    if len(starts) != 1:
        raise ValueError('Expected exactly one top-level images list')
    items, end = [], len(lines)
    for i in range(starts[0] + 1, len(lines)):
        line = lines[i]
        if line.strip() and not line.startswith((' ', '-', '#')):
            end = i
            break
        if ITEM.match(line):
            items.append(i)
        elif line.lstrip().startswith('- '):
            raise ValueError('Unsupported images entry form')
    return list(zip(items, items[1:] + [end]))


def set_image_tags(text, updates):
    """Set the existing newTag of each enrolled entry; return the new text.

    updates: {kustomize entry name: (registry name its newName must already be, new tag)}. Only the newTag line of
    each enrolled entry changes; every other byte is kept. Anything this narrow parser does not recognise is refused.
    """
    if '\r' in text:
        raise ValueError('Unsupported line endings')
    lines = text.split('\n')
    bounds = _entries(lines)
    for name, (new_name, tag) in updates.items():
        if not NAME.fullmatch(new_name) or not TAG.fullmatch(tag):
            raise ValueError('Invalid image pin')
        chosen = [(a, b) for a, b in bounds if ITEM.match(lines[a]).group(2) == name]
        if len(chosen) != 1:
            raise ValueError('Enrolled image is not exactly one images entry')
        start, stop = chosen[0]
        indent = ' ' * (len(ITEM.match(lines[start]).group(1)) + 2)
        seen = {}
        for number in range(start + 1, stop):
            line = lines[number]
            if not line.strip() or line.lstrip().startswith('#'):
                continue
            match = KEY.match(line)
            if (not match or match.group(1) != indent or match.group(2) not in ('newName', 'newTag', 'digest') or
                    match.group(2) in seen):
                raise ValueError('Unsupported or duplicate keys in the enrolled images entry')
            seen[match.group(2)] = number
        if 'digest' in seen:
            raise ValueError('A digest would override the tag; automation changes only an existing newTag')
        if 'newName' not in seen or KEY.match(lines[seen['newName']]).group(3) != new_name:
            raise ValueError('Enrolled entry must already point at the enrolled registry image')
        if 'newTag' not in seen:
            raise ValueError('Automation changes only an existing newTag')
        existing = TAG.fullmatch(KEY.match(lines[seen['newTag']]).group(3))
        if existing and int(existing.group(1)) > int(TAG.fullmatch(tag).group(1)):
            raise ValueError('Refusing to move a pin back to an older run')
        lines[seen['newTag']] = indent + 'newTag: ' + tag
    return '\n'.join(lines)


def _commit_is_verified(commit, what='GitOps commit'):
    verification = commit.get('commit', {}).get('verification') or {}
    if verification.get('verified') is not True:
        raise ValueError(what + ' is not signature-verified')


def _read_file(gitops, branch, gitops_policy):
    _, repo = gitops('GET', '/repos/' + gitops_policy['repository'])
    if (repo.get('id') != gitops_policy['repository_id'] or repo.get('full_name') != gitops_policy['repository'] or
            repo.get('fork') is not False or repo.get('archived') is True):
        raise ValueError('GitOps repository differs from enrollment')
    _, head = gitops('GET', '/repos/' + gitops_policy['repository'] + '/branches/' + branch)
    if (head.get('name') != branch or head.get('protected') is not True or
            head.get('user_can_push') is False):
        raise ValueError('GitOps branch must match, be protected, and accept the publisher')
    path = gitops_policy['path']
    base = '/repos/' + gitops_policy['repository']
    _, entry = gitops('GET', base + '/contents/' + path + '?ref=' + branch)
    if (entry.get('type') != 'file' or entry.get('encoding') != 'base64' or entry.get('path') != path or
            not re.fullmatch(r'[0-9a-f]{40}', entry.get('sha', '')) or
            type(entry.get('size')) is not int or entry['size'] > MAX_FILE_BYTES):
        raise ValueError('Unexpected GitOps file entry')
    data = base64.b64decode(''.join(entry['content'].split()), validate=True)
    return entry['sha'], data.decode('utf-8'), head['commit']['id']


def release_images(policy):
    """The enrolled images of a container release: the `images` list, or the single legacy `image`."""
    if 'images' in policy:
        return policy['images']
    return [{'name': 'image', 'image': policy['image'], 'artifact': policy['artifact'],
             'kustomize_name': policy.get('gitops', {}).get('image_name')}]


def _check_results(policy, run_id, results):
    images = release_images(policy)
    if len(results) != len(images):
        raise ValueError('Every enrolled image must have a verified publication')
    for entry, result in zip(images, results):
        if (result.get('repository') != policy['repository'] or result.get('run') != run_id or
                result.get('deployment_verified') is not False or
                result.get('workflow_sha256') != policy['workflow_sha256'] or
                not DIGEST.fullmatch(str(result.get('published_digest'))) or
                not isinstance(result.get('image'), dict) or
                result['image'].get('manifest') != result.get('published_digest') or
                not TAG.fullmatch(str(result.get('published_tag'))) or
                result.get('published_image') != REGISTRY + entry['image'] + ':' + result['published_tag']):
            raise ValueError('Publication result does not match the enrolled run')
    if len({r['published_tag'] for r in results}) != 1:
        raise ValueError('All images of one release must carry the same immutable tag')


def write_back(get, policy, run_id, results, reader, gitops, receipts, attempts=3):
    """Pin every published tag in the enrolled file with one commit; return the deployment receipt.

    `reader(image)` returns the read-only registry client for that image.
    """
    target = policy['gitops']
    images = release_images(policy)
    if any(not entry.get('kustomize_name') for entry in images):
        raise ValueError('Every image needs a kustomize entry name to be pinned')
    _check_results(policy, run_id, results)
    candidate = inspect_candidate(get, policy, run_id)
    # The OCI evidence sits under 'image' in each result and the enrolled names under 'images' in the candidate.
    if any(result.get(key) != value for result in results for key, value in candidate.items()
           if key not in ('image', 'images')):
        raise ValueError('Publication result differs from the current release candidate')
    directory = Path(receipts) / policy['repository']
    destination = directory / (str(run_id) + '.deploy.json')
    # Never move a pin backwards to an older run, even if that run is still eligible.
    for existing in directory.glob('*.deploy.json') if directory.is_dir() else ():
        if int(existing.name.split('.')[0]) > run_id:
            raise ValueError('A newer run was already written back')
    for entry, result in zip(images, results):
        _, _, manifest = reader(entry['image'])('GET', '/v2/' + entry['image'] + '/manifests/' + result['published_digest'])
        if 'sha256:' + hashlib.sha256(manifest).hexdigest() != result['published_digest']:
            raise ValueError('Registry does not serve the digest by content')
    tag = results[0]['published_tag']
    updates = {entry['kustomize_name']: (REGISTRY + entry['image'], tag) for entry in images}
    base = '/repos/' + target['repository']
    previous = None
    for _ in range(attempts):
        sha, text, head_id = _read_file(gitops, target['branch'], target)
        if sha == previous:
            raise ValueError('GitOps update was refused without a competing change')
        updated = set_image_tags(text, updates)
        if updated == text:
            _, found = gitops('GET', base + '/commits?sha=' + target['branch'] + '&path=' +
                              target['path'].replace('/', '%2F') + '&limit=1&stat=false')
            if len(found) != 1:
                raise ValueError('No commit found for the already-current GitOps file')
            _commit_is_verified(found[0])
            status, commit = 'already-current', found[0]['sha']
            break
        # A commit cannot be un-written: never make the first write on a branch whose head is not already
        # signature-verified. That would show server-side commit signing is not qualified, so refuse before writing.
        if not re.fullmatch(r'[0-9a-f]{40}', head_id):
            raise ValueError('Forge did not report the GitOps branch head')
        _, head_commit = gitops('GET', base + '/git/commits/' + head_id)
        _commit_is_verified(head_commit, 'GitOps branch head (server-side signing is not qualified)')
        # Everything authoritative must still hold at the instant before the only write.
        if inspect_candidate(get, policy, run_id) != candidate:
            raise ValueError('Release eligibility changed before the GitOps write')
        message = ('deploy: pin ' + ', '.join(sorted(n.rsplit('/', 1)[-1] for n in updates)) + ' to ' + tag +
                   '\n\nSource: ' + policy['repository'] + '@' + candidate['source_commit'] +
                   ' run ' + str(run_id) + '\n' + '\n'.join(
                       'Image: ' + REGISTRY + e['image'] + ':' + tag + ' ' + r['published_digest']
                       for e, r in zip(images, results)))
        try:
            _, written = gitops('PUT', base + '/contents/' + target['path'], {
                'content': base64.b64encode(updated.encode('utf-8')).decode(), 'sha': sha,
                'branch': target['branch'], 'message': message})
        except urllib.error.HTTPError as error:
            error.close()
            if error.code not in CONFLICT:
                raise
            previous = sha  # Lost the race (or refused): re-read and re-decide from scratch.
            continue
        commit = written['commit']['sha']
        if not re.fullmatch(r'[0-9a-f]{40}', commit):
            raise ValueError('Forge did not return a commit id')
        _, fetched = gitops('GET', base + '/git/commits/' + commit)
        _commit_is_verified(fetched)
        if [item.get('filename') for item in fetched.get('files') or []] != [target['path']]:
            raise ValueError('GitOps commit touched files other than the enrolled path')
        _, after = gitops('GET', base + '/contents/' + target['path'] + '?ref=' + commit)
        if base64.b64decode(''.join(after['content'].split()), validate=True).decode('utf-8') != updated:
            raise ValueError('Committed GitOps bytes differ from the intended update')
        status = 'committed'
        break
    else:
        raise ValueError('GitOps branch kept changing; no pin was written')
    receipt = {'repository': policy['repository'], 'run': run_id, 'source_commit': candidate['source_commit'],
               'gitops_repository': target['repository'], 'gitops_branch': target['branch'],
               'gitops_path': target['path'], 'tag': tag,
               'images': {e['name']: {'image': REGISTRY + e['image'] + ':' + tag, 'digest': r['published_digest']}
                          for e, r in zip(images, results)},
               'gitops_commit': commit, 'gitops_status': status, 'deployment_verified': False}
    if destination.exists() and json.loads(destination.read_text(encoding='utf-8')).get('images') != receipt['images']:
        raise ValueError('Existing deployment receipt conflicts with these images')
    write_receipt(destination, receipt)
    return receipt