gitea-dynamic-runner/src/gitea_dynamic_runner/opensandbox_identity.py

"""Reconcile OpenSandbox Pod UIDs to narrowly scoped SPIRE entries."""

from __future__ import annotations

import asyncio
import hashlib
import logging
import os
from pathlib import Path

from aiohttp import ClientResponseError

from .sandbox_kubernetes import SandboxKubernetesClient, allocated_pod_name


LOG = logging.getLogger(__name__)
KUBERNETES_API = os.environ.get("KUBERNETES_API", "https://kubernetes.default.svc")
KUBERNETES_TOKEN_FILE = Path(
    os.environ.get(
        "KUBERNETES_TOKEN_FILE",
        "/var/run/secrets/kubernetes.io/serviceaccount/token",
    )
)
KUBERNETES_CA_FILE = Path(
    os.environ.get(
        "KUBERNETES_CA_FILE",
        "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt",
    )
)
NAMESPACE = os.environ.get("OPENSANDBOX_NAMESPACE", "opensandbox")
SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top")
SPIRE_CLUSTER_NAME = os.environ.get("SPIRE_CLUSTER_NAME", "sandbox-kata")
SPIRE_CLASS_NAME = os.environ.get("SPIRE_CLASS_NAME", "spire-mgmt-spire")
RUNNER_UID = int(os.environ.get("RUNNER_UID", "2000"))
RECONCILE_INTERVAL = int(os.environ.get("RECONCILE_INTERVAL", "2"))


def entry_name(sandbox_id: str) -> str:
    suffix = hashlib.sha256(sandbox_id.encode()).hexdigest()[:12]
    return f"gitea-ci-{suffix}"


def sandbox_name(document: dict[str, object]) -> str | None:
    metadata = document.get("metadata")
    value = metadata.get("name") if isinstance(metadata, dict) else None
    return value if isinstance(value, str) and value else None


def is_runner_sandbox(document: dict[str, object]) -> bool:
    metadata = document.get("metadata")
    labels = metadata.get("labels") if isinstance(metadata, dict) else None
    return isinstance(labels, dict) and labels.get("ci.ddupan.top/runner") == "true"


def task_environment(document: dict[str, object]) -> dict[str, str]:
    spec = document.get("spec")
    task = spec.get("taskTemplate") if isinstance(spec, dict) else None
    task_spec = task.get("spec") if isinstance(task, dict) else None
    process = task_spec.get("process") if isinstance(task_spec, dict) else None
    values = process.get("env") if isinstance(process, dict) else None
    result: dict[str, str] = {}
    if not isinstance(values, list):
        return result
    for item in values:
        if not isinstance(item, dict):
            continue
        name, value = item.get("name"), item.get("value")
        if isinstance(name, str) and isinstance(value, str):
            result[name] = value
    return result


def identity_entry(
    *, sandbox_id: str, pod_uid: str, spiffe_id: str
) -> dict[str, object]:
    expected_prefix = f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/"
    if not spiffe_id.startswith(expected_prefix):
        raise ValueError("runner SPIFFE ID is outside the CI namespace")
    return {
        "apiVersion": "spire.spiffe.io/v1alpha1",
        "kind": "ClusterStaticEntry",
        "metadata": {
            "name": entry_name(sandbox_id),
            "labels": {
                "app.kubernetes.io/name": "gitea-dynamic-runner",
                "app.kubernetes.io/component": "opensandbox-identity",
                "ci.ddupan.top/sandbox-id": sandbox_id,
            },
        },
        "spec": {
            "className": SPIRE_CLASS_NAME,
            "parentID": (
                f"spiffe://{SPIFFE_TRUST_DOMAIN}/spire/agent/k8s_psat/"
                f"{SPIRE_CLUSTER_NAME}/pod/{pod_uid}"
            ),
            "spiffeID": spiffe_id,
            "selectors": [f"unix:uid:{RUNNER_UID}"],
        },
    }


async def reconcile(client: SandboxKubernetesClient) -> None:
    sandboxes = await client.list_batchsandboxes()
    live_names = {
        name
        for document in sandboxes
        if is_runner_sandbox(document) and (name := sandbox_name(document))
    }
    entries = await client.list_entries()
    existing = {
        name
        for document in entries
        if (name := sandbox_name(document)) is not None
    }

    for document in sandboxes:
        if not is_runner_sandbox(document):
            continue
        name = sandbox_name(document)
        pod_name = allocated_pod_name(document)
        spiffe_id = task_environment(document).get("CI_SPIFFE_ID")
        if not name or not pod_name or not spiffe_id or entry_name(name) in existing:
            continue
        pod = await client.get_pod(pod_name)
        metadata = pod.get("metadata") if isinstance(pod, dict) else None
        pod_uid = metadata.get("uid") if isinstance(metadata, dict) else None
        if not isinstance(pod_uid, str) or not pod_uid:
            continue
        try:
            await client.create_entry(
                identity_entry(
                    sandbox_id=name,
                    pod_uid=pod_uid,
                    spiffe_id=spiffe_id,
                )
            )
        except ClientResponseError as error:
            if error.status != 409:
                raise
        LOG.info(
            "SPIRE entry ready sandbox=%s pod=%s pod_uid=%s spiffe_id=%s",
            name,
            pod_name,
            pod_uid,
            spiffe_id,
        )

    for document in entries:
        metadata = document.get("metadata")
        labels = metadata.get("labels") if isinstance(metadata, dict) else None
        sandbox_id = labels.get("ci.ddupan.top/sandbox-id") if isinstance(labels, dict) else None
        name = metadata.get("name") if isinstance(metadata, dict) else None
        if (
            isinstance(name, str)
            and isinstance(sandbox_id, str)
            and sandbox_id not in live_names
        ):
            await client.delete_entry(name)
            LOG.info("removed stale SPIRE entry=%s sandbox=%s", name, sandbox_id)


async def main() -> None:
    async with SandboxKubernetesClient(
        api_url=KUBERNETES_API,
        token_file=KUBERNETES_TOKEN_FILE,
        ca_file=KUBERNETES_CA_FILE,
        namespace=NAMESPACE,
    ) as client:
        while True:
            try:
                await reconcile(client)
            except Exception:
                LOG.exception("OpenSandbox identity reconcile failed")
            await asyncio.sleep(RECONCILE_INTERVAL)


def cli() -> None:
    logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
    asyncio.run(main())


if __name__ == "__main__":
    cli()