infra/k8s/forgejo/forgejo-secret-sync/forgejo-secret-sync.py

#!/usr/bin/env python3
import subprocess
import logging
import base64
import os
import requests
import json

logging.basicConfig(level=logging.DEBUG)

with open("/var/lib/rancher/k3s/server/tls/server-ca.crt") as f:
    ca = base64.b64encode(f.read().encode()).decode()

forgejo_token = os.getenv("FORGEJO_TOKEN")


def run(cmd: list[str], stdin=None) -> str:
    logging.debug("executing %s", cmd)
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE)
    out = p.communicate(stdin)
    if p.returncode != 0:
        logging.critical("{} exited with code {}", cmd, p.returncode)
        os.exit(1)
    return out[0]


def update_cert(k8s_name: str, owner: str, repo: str):
    key = run(["openssl", "genrsa", "4096"])
    req = run(
        ["openssl", "req", "-key", "/dev/stdin", "-new", "-nodes", "-subj", f"/CN={k8s_name}"], stdin=key
    )
    cert = run(
        [
            "openssl",
            "x509",
            "-req",
            "-CA",
            "/var/lib/rancher/k3s/server/tls/client-ca.nochain.crt",
            "-CAkey",
            "/var/lib/rancher/k3s/server/tls/client-ca.key",
            "-CAcreateserial",
            "-days",
            "10",
        ],
        stdin=req,
    )

    keyb64 = base64.b64encode(key).decode()
    certb64 = base64.b64encode(cert).decode()

    kubeconfig = f"""
apiVersion: v1
clusters:
- cluster:
    certificate-authority-data: {ca}
    server: https://10.5.1.110:6443
  name: default
contexts:
- context:
    cluster: default
    user: default
  name: default
current-context: default
kind: Config
preferences: {"{}"}
users:
- name: default
  user:
    client-certificate-data: {certb64}
    client-key-data: {keyb64}
"""
    logging.info(f"updating secret for {owner}/{repo}")
    requests.put(
        f"https://git.janky.solutions/api/v1/repos/{owner}/{repo}/actions/secrets/KUBERNETES_CLIENT_CONFIG",
        data=json.dumps(
            {"data": kubeconfig},
        ),
        headers={
            "Authorization": f"token {forgejo_token}",
            "Content-Type": "application/json",
        },
    ).raise_for_status()


for entry in json.loads(os.getenv("REPO_MAPPINGS")):
    update_cert(**entry)