doran/scripts/hetzner/forgejo-bootstrap.py

#!/usr/bin/env python3
import argparse
import base64
import json
import ssl
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

import yaml
from nacl import encoding, public


class ForgejoClient:
    def __init__(self, base_url: str, username: str, password: str):
        self.base_url = base_url.rstrip('/')
        credentials = base64.b64encode(f'{username}:{password}'.encode()).decode()
        self.headers = {
            'Authorization': f'Basic {credentials}',
            'Accept': 'application/json',
            'Content-Type': 'application/json',
        }
        self.ssl_context = ssl.create_default_context()

    def request(self, method: str, path: str, payload=None, expected=(200, 201, 204)):
        url = f'{self.base_url}{path}'
        data = None
        if payload is not None:
            data = json.dumps(payload).encode()
        req = urllib.request.Request(url, data=data, method=method)
        for key, value in self.headers.items():
            req.add_header(key, value)
        try:
            with urllib.request.urlopen(req, context=self.ssl_context) as response:
                body = response.read().decode() if response.length != 0 else ''
                if response.status not in expected:
                    raise RuntimeError(f'{method} {path} returned {response.status}: {body}')
                if not body:
                    return None
                return json.loads(body)
        except urllib.error.HTTPError as exc:
            body = exc.read().decode()
            if exc.code not in expected:
                raise RuntimeError(f'{method} {path} returned {exc.code}: {body}') from exc
            return json.loads(body) if body else None

    def get_repo(self, owner: str, repo: str):
        try:
            return self.request('GET', f'/api/v1/repos/{owner}/{repo}')
        except RuntimeError as exc:
            if ' returned 404:' in str(exc):
                return None
            raise

    def create_repo(self, name: str, private: bool):
        return self.request('POST', '/api/v1/user/repos', {
            'name': name,
            'private': private,
            'auto_init': False,
            'default_branch': 'main',
        }, expected=(201,))

    def upsert_variable(self, owner: str, repo: str, name: str, value: str):
        try:
            self.request('POST', f'/api/v1/repos/{owner}/{repo}/actions/variables/{urllib.parse.quote(name)}', {
                'value': value,
            }, expected=(201,))
        except RuntimeError as exc:
            if ' returned 409:' not in str(exc) and ' returned 422:' not in str(exc):
                raise
            self.request('PUT', f'/api/v1/repos/{owner}/{repo}/actions/variables/{urllib.parse.quote(name)}', {
                'value': value,
            }, expected=(201, 204))

    def upsert_secret(self, owner: str, repo: str, name: str, value: str):
        key = self.request('GET', f'/api/v1/repos/{owner}/{repo}/actions/secrets/public-key')
        public_key = public.PublicKey(key['key'].encode(), encoder=encoding.Base64Encoder)
        sealed_box = public.SealedBox(public_key)
        encrypted_value = base64.b64encode(sealed_box.encrypt(value.encode())).decode()
        self.request('PUT', f'/api/v1/repos/{owner}/{repo}/actions/secrets/{urllib.parse.quote(name)}', {
            'encrypted_value': encrypted_value,
            'key_id': key['key_id'],
        }, expected=(201, 204))


def render_ci_kubeconfig(source: Path) -> str:
    config = yaml.safe_load(source.read_text())
    clusters = config.get('clusters') or []
    if not clusters:
        raise SystemExit('kubeconfig does not contain any clusters')
    clusters[0]['cluster']['server'] = 'https://kubernetes.default.svc:443'
    return yaml.safe_dump(config, sort_keys=False)


def main():
    parser = argparse.ArgumentParser(description='Bootstrap Forgejo repo secrets/variables for CI/CD')
    parser.add_argument('--forgejo-url', required=True)
    parser.add_argument('--admin-username', required=True)
    parser.add_argument('--admin-password', required=True)
    parser.add_argument('--repo-owner', required=True)
    parser.add_argument('--repo-name', required=True)
    parser.add_argument('--repo-private', action='store_true', default=False)
    parser.add_argument('--kubeconfig', required=True)
    parser.add_argument('--registry-username', required=True)
    parser.add_argument('--registry-password', required=True)
    parser.add_argument('--registry-host', required=True)
    parser.add_argument('--project-name', required=True)
    parser.add_argument('--project-namespace', required=True)
    parser.add_argument('--project-deployments', required=True)
    args = parser.parse_args()

    client = ForgejoClient(args.forgejo_url, args.admin_username, args.admin_password)
    repo = client.get_repo(args.repo_owner, args.repo_name)
    if repo is None:
        created = client.create_repo(args.repo_name, args.repo_private)
        print(f'created repo {created["full_name"]}')
    else:
        print(f'repo already exists: {repo["full_name"]}')

    ci_kubeconfig = render_ci_kubeconfig(Path(args.kubeconfig))
    client.upsert_secret(args.repo_owner, args.repo_name, 'KUBECONFIG_B64', base64.b64encode(ci_kubeconfig.encode()).decode())
    client.upsert_secret(args.repo_owner, args.repo_name, 'REGISTRY_USERNAME', args.registry_username)
    client.upsert_secret(args.repo_owner, args.repo_name, 'REGISTRY_PASSWORD', args.registry_password)
    print('upserted repo action secrets')

    client.upsert_variable(args.repo_owner, args.repo_name, 'REGISTRY_HOST', args.registry_host)
    client.upsert_variable(args.repo_owner, args.repo_name, 'PROJECT_NAME', args.project_name)
    client.upsert_variable(args.repo_owner, args.repo_name, 'PROJECT_NAMESPACE', args.project_namespace)
    client.upsert_variable(args.repo_owner, args.repo_name, 'PROJECT_DEPLOYMENTS', args.project_deployments)
    print('upserted repo action variables')


if __name__ == '__main__':
    main()