停用镜像自动发布流水线

.gitea/workflows/publish-images.yml

@@ -1,122 +0,0 @@
----
-name: publish images
-
-on:
-  push:
-    branches: [main]
-    paths:
-      - '.gitea/workflows/publish-images.yml'
-      - 'config/**'
-      - 'container/**'
-      - 'scripts/**'
-      - 'src/**'
-      - 'pyproject.toml'
-      - 'README.md'
-  workflow_dispatch:
-
-jobs:
-  publish-images:
-    name: publish-images
-    runs-on: [self-hosted, vm]
-    timeout-minutes: 45
-    permissions:
-      contents: read
-    env:
-      PUSH_REGISTRY: zot-push.ad.ddupan.top
-      PULL_REGISTRY: zot.ad.ddupan.top
-      CONTROLLER_REPOSITORY: panxiao81/gitea-dynamic-runner-controller
-      RUNNER_REPOSITORY: panxiao81/gitea-dynamic-runner-runner
-      SPIRE_AGENT_SOCKET: /run/spire/agent-sockets/spire-agent.sock
-    steps:
-      - uses: actions/checkout@v4
-
-      - name: Test source
-        shell: bash
-        run: |
-          set -euo pipefail
-          python3 -m pip install --break-system-packages -e '.[test]'
-          pytest -q
-          python3 -m compileall -q src tests
-          apt-get update
-          apt-get install --yes --no-install-recommends shellcheck
-          shellcheck scripts/*
-
-      - name: Build and publish
-        shell: bash
-        run: |
-          set -euo pipefail
-          set +x
-
-          : "${GITHUB_SHA:?GITHUB_SHA is required}"
-          image_tag="sha-${GITHUB_SHA}"
-          docker_config=$(mktemp -d)
-          jwt_file=$(mktemp)
-          cleanup() {
-            docker buildx rm ci-builder >/dev/null 2>&1 || true
-            rm -rf -- "$docker_config" "$jwt_file"
-          }
-          trap cleanup EXIT
-          export DOCKER_CONFIG="$docker_config"
-
-          /opt/spire/bin/spire-agent api fetch jwt \
-            -audience zot \
-            -socketPath "$SPIRE_AGENT_SOCKET" \
-            -output json >"$jwt_file"
-          # shellcheck disable=SC2016
-          jq -er '.[0].svids[0].svid' "$jwt_file" | \
-            docker login "$PUSH_REGISTRY" --username zot --password-stdin
-
-          docker buildx create \
-            --name ci-builder \
-            --driver docker-container \
-            --use
-
-          publish() {
-            local repository=$1
-            local dockerfile=$2
-            local metadata=$3
-            docker buildx build \
-              --builder ci-builder \
-              --platform linux/amd64 \
-              --file "$dockerfile" \
-              --tag "${PUSH_REGISTRY}/${repository}:${image_tag}" \
-              --tag "${PUSH_REGISTRY}/${repository}:main" \
-              --provenance=mode=max \
-              --sbom=true \
-              --metadata-file "$metadata" \
-              --push \
-              .
-          }
-
-          publish \
-            "$CONTROLLER_REPOSITORY" \
-            container/controller.Dockerfile \
-            controller-metadata.json
-          publish \
-            "$RUNNER_REPOSITORY" \
-            container/runner.Dockerfile \
-            runner-metadata.json
-
-          controller_digest=$(
-            # shellcheck disable=SC2016
-            jq -er '."containerimage.digest"' controller-metadata.json
-          )
-          runner_digest=$(
-            # shellcheck disable=SC2016
-            jq -er '."containerimage.digest"' runner-metadata.json
-          )
-          controller_ref="${PULL_REGISTRY}/${CONTROLLER_REPOSITORY}@${controller_digest}"
-          runner_ref="${PULL_REGISTRY}/${RUNNER_REPOSITORY}@${runner_digest}"
-
-          printf 'controller=%s\nrunner=%s\n' "$controller_ref" "$runner_ref"
-          if [[ -n "${GITHUB_STEP_SUMMARY:-}" ]]; then
-            {
-              printf '## Published images\n\n'
-              # shellcheck disable=SC2016
-              printf -- '- Controller: `%s`\n' "$controller_ref"
-              # shellcheck disable=SC2016
-              printf -- '- Runner: `%s`\n' "$runner_ref"
-              # shellcheck disable=SC2016
-              printf -- '- Source: `%s`\n' "$GITHUB_SHA"
-            } >>"$GITHUB_STEP_SUMMARY"
-          fi

README.md

@@ -17,23 +17,25 @@ runs-on: [self-hosted, vm]
 
 组件：
 
-- `controller`：接收 Gitea `workflow_job` webhook，直接调用 OpenSandbox Lifecycle
-  API，从 `ci-pod` 或 `ci-vm` Pool 创建一次性环境。
+- `controller`：接收 Gitea `workflow_job` webhook，将指定 label 的 queued job
+  发布到 NATS JetStream。
+- `worker`：领取任务、限制并发，并通过 Pod 或 microVM backend 创建一次性环境。
 - `microvm-runner-launch`：为每个任务以 direct I/O 转换出 flat qcow2 root disk、创建 NoCloud seed 和 TAP，运行
   Cloud Hypervisor，退出后完整清理。
 - `guest-runner`：在 guest 中领取一次性 runner registration token，注册 ephemeral
   runner，执行一个 job 后关机。
-- `opensandbox-identity`：在 sandbox 集群按实际 Pod UID 创建并清理临时 SPIFFE
-  entry；不持有 OpenSandbox API key、Gitea token 或 Bao 凭据。身份与 Pool 契约见
-  [`docs/opensandbox-runner.md`](docs/opensandbox-runner.md)。
 - `pod-worker`：在 Kubernetes 中创建一次性 privileged Pod；Pod 内的 workflow 使用
   host executor，Docker、BuildKit 和 kind 等工具由 pipeline 按需 setup。Runner 固定在
   支持原生 job hooks 的 3.x 版本，在 workflow 第一步前等待实际任务对应的 SVID。
 - `jwt-broker`：早期共享 Kubernetes runner 的过渡实验；目标架构不部署它，每个
   动态 Pod 或 VM 直接取得自己的 SPIFFE 身份。
 
-OpenSandbox 路径不使用 NATS。旧 Pod/microVM worker 只作为迁移期代码保留，不应重新
-部署。长期 RunnerService 协议路线见
+消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一
+durable consumer；扩容只需要增加 worker 或提高单机 capacity。
+
+当前 webhook → NATS 流程是用于尽快验证 Pod/VM 生命周期的 bootstrap 实现，不是
+长期调度接口。长期目标是让 controller 作为兼容 Gitea Runner 协议的调度器直接注册、
+声明 labels、领取 task，并把已领取 task 交给 Pod/VM executor；路线与迁移边界见
 [`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
 
 ## 开发
@@ -47,7 +49,7 @@ pytest
 
 ## 安全边界
 
-- OpenSandbox API key、webhook secret 和 Gitea registration token 只从文件读取。
+- NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。
 - registration token 不写入 seed image；worker 通过单次 nonce endpoint 交给 guest。
 - guest 启动时从仅监听 microVM bridge 的 worker endpoint 获取固定版本 Runner 和配置
   资产；基础镜像无需为 Runner 发布而重做。

container/runner.Dockerfile

@@ -14,7 +14,6 @@ COPY --from=runner /usr/local/bin/run.sh /usr/local/bin/run.sh
 COPY --from=spire /opt/spire/bin/spire-agent /opt/spire/bin/spire-agent
 COPY config/runner.yaml /etc/gitea-runner/config.yaml
 COPY --chmod=0755 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
-COPY --chmod=0755 scripts/gitea-opensandbox-runner /usr/local/libexec/gitea-opensandbox-runner
 
 VOLUME ["/data"]
 WORKDIR /

docs/opensandbox-runner.md

@@ -1,52 +0,0 @@
-# OpenSandbox runner
-
-Gitea 的 `workflow_job` webhook 只负责发现带 `self-hosted,pod` 或
-`self-hosted,vm` label 的 queued job。controller 不再把请求写入 NATS，而是携带由
-homelab ExternalSecret 挂载的 API key 直接调用 OpenSandbox Lifecycle API：
-
-```text
-http://10.60.0.13:8080/v1/sandboxes
-```
-
-请求通过 `extensions.poolRef` 选择 `ci-pod` 或 `ci-vm`。地址是 VyOS HAProxy 的
-内网 TCP frontend，backend 为 sandbox 两个节点上的固定 NodePort；不通过公网或
-Cloudflare Tunnel。OpenSandbox key 只从 OpenBao `kv/k8s/opensandbox-api.api_key`
-进入 homelab Secret，进程通过 `OPENSANDBOX_API_KEY_FILE` 读取。
-
-## 身份顺序
-
-Lifecycle 请求把稳定的 repository/task SPIFFE ID 放入 task environment。sandbox
-集群内的 `opensandbox-identity` controller 读取 BatchSandbox allocation 得到实际
-Pod UID，然后创建：
-
-- parent：`spiffe://ddupan.top/spire/agent/k8s_psat/sandbox-kata/pod/<pod-uid>`；
-- workload：`spiffe://ddupan.top/ci/<owner>/<repository>/<task>`；
-- selector：`unix:uid:2000`。
-
-job ID 只进入诊断 label，不进入业务身份。Pool 内 runner 进程固定使用 UID 2000，
-Pod 设置 `shareProcessNamespace: true`；guest-local SPIRE Agent 使用 Pod-bound PSAT，
-通过内存 emptyDir 暴露 Workload API。`gitea-opensandbox-runner` 等待精确 SVID，随后
-通过一次性 nonce URL 领取 Gitea registration token。token 不进入 Lifecycle 请求、
-BatchSandbox、镜像或 sandbox Secret。
-
-identity controller 只能读取 opensandbox namespace 的 BatchSandbox/Pod，并维护带
-自身 label 的 ClusterStaticEntry。它不持有 OpenSandbox API key、Gitea token 或 Bao
-凭据。BatchSandbox 消失后，对应 entry 在下一次 reconcile 删除；Pod 删除后，按 Pod
-UID attestation 的临时 Agent 失去父级。
-
-## Pool 契约
-
-`ci-vm` 使用 `kata-clh-runtime-rs`；`ci-pod` 使用默认 runc。两者都要求：
-
-- runner 镜像包含 Gitea Runner、Node.js action userspace、SPIRE CLI 和 identity gate；
-- runner UID 2000，SPIRE Agent 与 privileged dockerd 使用不同 UID；
-- Docker socket 通过 group 2000 共享，Docker 数据仅存在于 sandbox emptyDir；
-- `self-hosted` 必须是所有 runner labels 的前缀；
-- ephemeral/once runner 完成一项任务后退出。
-
-## 清理与恢复
-
-controller 监控 Lifecycle 状态，在终止、失败、超时或取消时调用 DELETE。API delete、
-identity entry delete 均接受对象已不存在。controller 重启时，OpenSandbox timeout
-仍是最终回收边界；后续可基于 metadata list 恢复主动监控，但不得为此重新引入消息
-队列。

pyproject.toml

@@ -16,7 +16,6 @@ test = ["pytest==8.4.2", "pytest-asyncio==1.2.0"]
 gitea-dynamic-runner-controller = "gitea_dynamic_runner.controller:main"
 gitea-dynamic-runner-pod-worker = "gitea_dynamic_runner.pod_worker:cli"
 gitea-dynamic-runner-vm-worker = "gitea_dynamic_runner.worker:cli"
-gitea-dynamic-runner-opensandbox-identity = "gitea_dynamic_runner.opensandbox_identity:cli"
 gitea-spire-jwt-broker = "gitea_dynamic_runner.jwt_broker:main"
 
 [tool.pytest.ini_options]

scripts/gitea-opensandbox-runner

@@ -1,30 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-: "${CI_SPIFFE_ID:?CI_SPIFFE_ID is required}"
-: "${SPIFFE_ENDPOINT_SOCKET:?SPIFFE_ENDPOINT_SOCKET is required}"
-: "${GITEA_RUNNER_REGISTRATION_TOKEN_URL:?GITEA_RUNNER_REGISTRATION_TOKEN_URL is required}"
-
-socket_path=${SPIFFE_ENDPOINT_SOCKET#unix://}
-token_file=$(mktemp)
-trap 'rm -f -- "$token_file"' EXIT
-export GITEA_RUNNER_REGISTRATION_TOKEN_FILE="$token_file"
-deadline=$((SECONDS + 120))
-while (( SECONDS < deadline )); do
-  if /opt/spire/bin/spire-agent api fetch x509 \
-    -socketPath "$socket_path" \
-    -output json 2>/dev/null | \
-    jq -e --arg id "$CI_SPIFFE_ID" \
-      'any(.svids[]?; .spiffe_id == $id)' >/dev/null; then
-    umask 077
-    curl --fail --silent --show-error --retry 10 --retry-all-errors \
-      --connect-timeout 2 --max-time 30 \
-      "$GITEA_RUNNER_REGISTRATION_TOKEN_URL" >"$token_file"
-    /usr/local/bin/run.sh
-    exit $?
-  fi
-  sleep 1
-done
-
-printf 'timed out waiting for OpenSandbox SPIFFE identity %s\n' "$CI_SPIFFE_ID" >&2
-exit 1

src/gitea_dynamic_runner/controller.py

@@ -1,54 +1,63 @@
 #!/usr/bin/env python3
-"""Gitea workflow_job webhook to the OpenSandbox Lifecycle API."""
-
-from __future__ import annotations
+"""Gitea workflow_job webhook to NATS JetStream producer."""
 
 import hashlib
 import hmac
 import json
 import logging
 import os
+import ssl
 from pathlib import Path
 
+import nats
 from aiohttp import web
+from nats.js.api import DiscardPolicy, RetentionPolicy, StorageType, StreamConfig
+from nats.js.errors import NotFoundError
 
-from .models import RunnerRequest
-from .opensandbox import OpenSandboxClient
-from .opensandbox_worker import OpenSandboxScheduler, RegistrationTokens
+from .models import IdentityBinding, RunnerRequest
 
 
 LOG = logging.getLogger(__name__)
-WEBHOOK_SECRET_FILE = Path(
-    os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret")
-)
-REGISTRATION_TOKEN_FILE = Path(
-    os.environ.get(
-        "REGISTRATION_TOKEN_FILE", "/run/secrets/gitea/registration-token"
-    )
-)
-OPENSANDBOX_API = os.environ.get(
-    "OPENSANDBOX_API", "http://10.60.0.13:8080"
-)
-OPENSANDBOX_API_KEY_FILE = Path(
-    os.environ.get(
-        "OPENSANDBOX_API_KEY_FILE",
-        "/run/secrets/opensandbox/api-key",
-    )
-)
+SUBJECT_PREFIX = os.environ.get("NATS_SUBJECT_PREFIX", "ci.runner")
+STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER")
+NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222")
+NATS_USER = os.environ.get("NATS_USER", "ci-producer")
+NATS_PASSWORD_FILE = Path(os.environ.get("NATS_PASSWORD_FILE", "/run/secrets/nats/password"))
+NATS_CA_FILE = os.environ.get("NATS_CA_FILE", "/etc/ssl/certs/ca-certificates.crt")
+WEBHOOK_SECRET_FILE = Path(os.environ.get("WEBHOOK_SECRET_FILE", "/run/secrets/gitea/webhook-secret"))
 
 
 def accepts(payload: object) -> tuple[bool, str | None]:
+    """Compatibility helper for callers that only need acceptance and identity."""
     request = RunnerRequest.from_webhook(payload)
     return (request is not None, str(request.job_id) if request else None)
 
 
 def valid_signature(body: bytes, signature: str) -> bool:
-    expected = hmac.new(
-        WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256
-    ).hexdigest()
+    expected = hmac.new(WEBHOOK_SECRET_FILE.read_bytes().strip(), body, hashlib.sha256).hexdigest()
     return hmac.compare_digest(signature.removeprefix("sha256="), expected)
 
 
+async def ensure_stream(js: object) -> None:
+    config = StreamConfig(
+        name=STREAM,
+        subjects=[f"{SUBJECT_PREFIX}.>"],
+        retention=RetentionPolicy.WORK_QUEUE,
+        storage=StorageType.FILE,
+        discard=DiscardPolicy.OLD,
+        max_age=24 * 60 * 60,
+        max_msgs=10_000,
+        max_bytes=256 * 1024 * 1024,
+        duplicate_window=24 * 60 * 60,
+    )
+    try:
+        await js.stream_info(STREAM)
+    except NotFoundError:
+        await js.add_stream(config=config)
+    else:
+        await js.update_stream(config=config)
+
+
 async def webhook(request: web.Request) -> web.Response:
     body = await request.read()
     if not valid_signature(body, request.headers.get("X-Gitea-Signature", "")):
@@ -61,29 +70,50 @@ async def webhook(request: web.Request) -> web.Response:
     LOG.info("workflow_job webhook received %s", context)
 
     runner_request = RunnerRequest.from_webhook(payload)
-    if runner_request is None:
-        LOG.info("workflow_job webhook ignored %s", context)
+    if runner_request is not None:
+        subject = f"{SUBJECT_PREFIX}.{runner_request.backend}"
+        await request.app["js"].publish(
+            subject,
+            runner_request.to_json(),
+            headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
+        )
+        LOG.info(
+            "runner request published subject=%s job_id=%d run_id=%d backend=%s "
+            "repository=%s job_name=%r labels=%s",
+            subject,
+            runner_request.job_id,
+            runner_request.run_id,
+            runner_request.backend,
+            runner_request.repository,
+            runner_request.job_name,
+            runner_request.labels,
+        )
+        return web.Response(status=202, text="queued\n")
+
+    binding = IdentityBinding.from_webhook(payload)
+    if binding is not None:
+        subject = f"{SUBJECT_PREFIX}.{binding.backend}.binding"
+        await request.app["js"].publish(
+            subject,
+            binding.to_json(),
+            headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"},
+        )
+        LOG.info(
+            "identity binding published subject=%s job_id=%d run_id=%d backend=%s "
+            "runner_name=%s repository=%s job_name=%r",
+            subject,
+            binding.job_id,
+            binding.run_id,
+            binding.backend,
+            binding.runner_name,
+            binding.repository,
+            binding.job_name,
+        )
+        return web.Response(status=202, text="binding queued\n")
+
+    LOG.warning("workflow_job webhook ignored %s", context)
     return web.Response(status=204)
 
-    scheduler: OpenSandboxScheduler = request.app["scheduler"]
-    try:
-        sandbox_id = await scheduler.create(runner_request)
-    except ValueError as error:
-        LOG.info("duplicate workflow_job webhook ignored: %s", error)
-        return web.Response(status=202, text="already scheduled\n")
-    except Exception:
-        LOG.exception("failed to create OpenSandbox runner %s", context)
-        raise web.HTTPServiceUnavailable(text="sandbox unavailable\n")
-    return web.Response(status=202, text=f"sandbox={sandbox_id}\n")
-
-
-async def registration_token(request: web.Request) -> web.Response:
-    scheduler: OpenSandboxScheduler = request.app["scheduler"]
-    value = await scheduler.tokens.consume(request.match_info["nonce"])
-    if value is None:
-        raise web.HTTPNotFound()
-    return web.Response(body=value, headers={"Cache-Control": "no-store"})
-
 
 def _webhook_context(payload: object) -> str:
     if not isinstance(payload, dict):
@@ -101,42 +131,37 @@ def _webhook_context(payload: object) -> str:
 
 
 async def health(request: web.Request) -> web.Response:
-    client: OpenSandboxClient = request.app["opensandbox_client"]
-    return web.Response(
-        text="ok\n" if client.session is not None else "disconnected\n",
-        status=200 if client.session is not None else 503,
+    connected = request.app["nc"].is_connected
+    return web.Response(text="ok\n" if connected else "disconnected\n", status=200 if connected else 503)
+
+
+async def nats_context(app: web.Application):
+    tls = ssl.create_default_context(cafile=NATS_CA_FILE)
+    nc = await nats.connect(
+        NATS_URL,
+        user=NATS_USER,
+        password=NATS_PASSWORD_FILE.read_text().strip(),
+        tls=tls,
+        name="microvm-runner-controller",
     )
-
-
-async def opensandbox_context(app: web.Application):
-    tokens = RegistrationTokens(REGISTRATION_TOKEN_FILE.read_bytes())
-    async with OpenSandboxClient(
-        api_url=OPENSANDBOX_API,
-        api_key_file=OPENSANDBOX_API_KEY_FILE,
-    ) as client:
-        scheduler = OpenSandboxScheduler(client, tokens)
-        app["opensandbox_client"] = client
-        app["scheduler"] = scheduler
+    app["nc"] = nc
+    app["js"] = nc.jetstream()
+    await ensure_stream(app["js"])
     yield
-        await scheduler.close()
+    await nc.drain()
 
 
 def create_app() -> web.Application:
     app = web.Application(client_max_size=1024 * 1024)
-    app.cleanup_ctx.append(opensandbox_context)
+    app.cleanup_ctx.append(nats_context)
     app.router.add_post("/webhook", webhook)
-    app.router.add_get("/token/{nonce}", registration_token)
     app.router.add_get("/healthz", health)
     return app
 
 
 def main() -> None:
     logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"))
-    web.run_app(
-        create_app(),
-        host=os.environ.get("LISTEN", "0.0.0.0"),
-        port=int(os.environ.get("PORT", "8787")),
-    )
+    web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787")))
 
 
 if __name__ == "__main__":

src/gitea_dynamic_runner/opensandbox.py

@@ -1,79 +0,0 @@
-"""Minimal client for the OpenSandbox lifecycle API."""
-
-from __future__ import annotations
-
-from pathlib import Path
-from urllib.parse import quote
-
-from aiohttp import ClientResponseError, ClientSession
-
-
-class OpenSandboxClient:
-    def __init__(
-        self,
-        *,
-        api_url: str,
-        api_key_file: Path | None = None,
-    ) -> None:
-        self.api_url = api_url.rstrip("/")
-        self.api_key_file = api_key_file
-        self.session: ClientSession | None = None
-
-    async def __aenter__(self) -> OpenSandboxClient:
-        headers = {}
-        if self.api_key_file is not None:
-            headers["OPEN-SANDBOX-API-KEY"] = self.api_key_file.read_text().strip()
-        self.session = ClientSession(headers=headers, raise_for_status=True)
-        return self
-
-    async def __aexit__(self, *_: object) -> None:
-        if self.session is not None:
-            await self.session.close()
-
-    async def create(
-        self,
-        *,
-        pool: str,
-        timeout: int,
-        entrypoint: list[str],
-        env: dict[str, str],
-        metadata: dict[str, str],
-    ) -> dict[str, object]:
-        response = await self._request(
-            "POST",
-            "/v1/sandboxes",
-            json={
-                "timeout": timeout,
-                "entrypoint": entrypoint,
-                "env": env,
-                "metadata": metadata,
-                "extensions": {"poolRef": pool},
-            },
-        )
-        return await response.json()
-
-    async def get(self, sandbox_id: str) -> dict[str, object] | None:
-        try:
-            response = await self._request(
-                "GET", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
-            )
-        except ClientResponseError as error:
-            if error.status == 404:
-                return None
-            raise
-        return await response.json()
-
-    async def delete(self, sandbox_id: str) -> None:
-        try:
-            response = await self._request(
-                "DELETE", f"/v1/sandboxes/{quote(sandbox_id, safe='')}"
-            )
-            response.release()
-        except ClientResponseError as error:
-            if error.status != 404:
-                raise
-
-    async def _request(self, method: str, path: str, **kwargs: object):
-        if self.session is None:
-            raise RuntimeError("OpenSandboxClient is not open")
-        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)

src/gitea_dynamic_runner/opensandbox_identity.py

@@ -1,183 +0,0 @@
-"""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()

src/gitea_dynamic_runner/opensandbox_worker.py

@@ -1,163 +0,0 @@
-"""Direct OpenSandbox lifecycle scheduler used by the webhook controller."""
-
-from __future__ import annotations
-
-import asyncio
-import logging
-import os
-import secrets
-import uuid
-from collections.abc import Callable
-
-from .models import RunnerRequest
-from .opensandbox import OpenSandboxClient
-from .pod_worker import identity_path
-
-
-LOG = logging.getLogger(__name__)
-SANDBOX_TIMEOUT = int(os.environ.get("RUNNER_SANDBOX_TIMEOUT", str(4 * 60 * 60)))
-SPIFFE_TRUST_DOMAIN = os.environ.get("SPIFFE_TRUST_DOMAIN", "ddupan.top")
-TOKEN_BASE_URL = os.environ.get(
-    "RUNNER_TOKEN_BASE_URL", "http://192.168.10.127:8787/token"
-).rstrip("/")
-
-
-def sandbox_request(
-    request: RunnerRequest,
-    runner_name: str,
-    registration_token_url: str,
-) -> dict[str, object]:
-    path = identity_path(request.repository, request.job_name)
-    return {
-        "pool": f"ci-{request.backend}",
-        "timeout": SANDBOX_TIMEOUT,
-        "entrypoint": ["/usr/local/libexec/gitea-opensandbox-runner"],
-        "env": {
-            "GITEA_RUNNER_NAME": runner_name,
-            "GITEA_RUNNER_LABELS": f"self-hosted:host,{request.backend}:host",
-            "GITEA_RUNNER_REGISTRATION_TOKEN_URL": registration_token_url,
-            "GITEA_RUNNER_EPHEMERAL": "1",
-            "GITEA_RUNNER_ONCE": "1",
-            "CONFIG_FILE": "/etc/gitea-runner/config.yaml",
-            "CI_SPIFFE_ID": f"spiffe://{SPIFFE_TRUST_DOMAIN}/ci/{path}",
-            "SPIFFE_ENDPOINT_SOCKET": (
-                "unix:///run/spire/agent-sockets/spire-agent.sock"
-            ),
-        },
-        "metadata": {
-            "ci.ddupan.top/runner": "true",
-            "ci.ddupan.top/job-id": str(request.job_id),
-            "ci.ddupan.top/run-id": str(request.run_id),
-            "ci.ddupan.top/runner-name": runner_name,
-        },
-    }
-
-
-class RegistrationTokens:
-    """Single-use registration-token URLs; values never enter Sandbox CRs."""
-
-    def __init__(self, token: bytes) -> None:
-        self._token = token.strip()
-        self._values: dict[str, bytes] = {}
-        self._lock = asyncio.Lock()
-
-    async def issue(self) -> tuple[str, str]:
-        nonce = secrets.token_urlsafe(32)
-        async with self._lock:
-            self._values[nonce] = self._token
-        return nonce, f"{TOKEN_BASE_URL}/{nonce}"
-
-    async def consume(self, nonce: str) -> bytes | None:
-        async with self._lock:
-            return self._values.pop(nonce, None)
-
-    async def revoke(self, nonce: str) -> None:
-        async with self._lock:
-            self._values.pop(nonce, None)
-
-
-class OpenSandboxScheduler:
-    def __init__(
-        self,
-        client: OpenSandboxClient,
-        tokens: RegistrationTokens,
-        *,
-        on_finished: Callable[[int], None] | None = None,
-    ) -> None:
-        self.client = client
-        self.tokens = tokens
-        self.on_finished = on_finished
-        self.active: dict[int, asyncio.Task[None]] = {}
-
-    async def create(self, request: RunnerRequest) -> str:
-        if request.job_id in self.active:
-            raise ValueError(f"job {request.job_id} already has an active sandbox")
-        runner_name = f"gitea-{request.backend}-{uuid.uuid4().hex[:12]}"
-        nonce, token_url = await self.tokens.issue()
-        try:
-            response = await self.client.create(
-                **sandbox_request(request, runner_name, token_url)
-            )
-            sandbox_id = response.get("id")
-            if not isinstance(sandbox_id, str) or not sandbox_id:
-                raise RuntimeError("OpenSandbox create response has no id")
-        except Exception:
-            await self.tokens.revoke(nonce)
-            raise
-
-        task = asyncio.create_task(
-            self._monitor(request, sandbox_id, nonce),
-            name=f"opensandbox-{sandbox_id}",
-        )
-        task.add_done_callback(self._report)
-        self.active[request.job_id] = task
-        LOG.info(
-            "OpenSandbox runner created sandbox=%s runner=%s job_id=%d "
-            "repository=%s job_name=%r pool=ci-%s",
-            sandbox_id,
-            runner_name,
-            request.job_id,
-            request.repository,
-            request.job_name,
-            request.backend,
-        )
-        return sandbox_id
-
-    async def _monitor(
-        self, request: RunnerRequest, sandbox_id: str, nonce: str
-    ) -> None:
-        try:
-            deadline = asyncio.get_running_loop().time() + SANDBOX_TIMEOUT
-            while asyncio.get_running_loop().time() < deadline:
-                sandbox = await self.client.get(sandbox_id)
-                if sandbox is None:
-                    return
-                status = sandbox.get("status")
-                state = status.get("state") if isinstance(status, dict) else None
-                if state in {"Terminated", "Failed"}:
-                    return
-                await asyncio.sleep(2)
-            raise asyncio.TimeoutError(f"OpenSandbox {sandbox_id} timed out")
-        finally:
-            await self.tokens.revoke(nonce)
-            try:
-                await self.client.delete(sandbox_id)
-            finally:
-                self.active.pop(request.job_id, None)
-                if self.on_finished is not None:
-                    self.on_finished(request.job_id)
-
-    @staticmethod
-    def _report(task: asyncio.Task[None]) -> None:
-        if not task.cancelled() and (error := task.exception()) is not None:
-            LOG.error(
-                "OpenSandbox lifecycle task failed",
-                exc_info=(type(error), error, error.__traceback__),
-            )
-
-    async def close(self) -> None:
-        tasks = list(self.active.values())
-        for task in tasks:
-            task.cancel()
-        if tasks:
-            await asyncio.gather(*tasks, return_exceptions=True)

src/gitea_dynamic_runner/sandbox_kubernetes.py

@@ -1,148 +0,0 @@
-"""Kubernetes resources that bind an OpenSandbox Kata guest to SPIRE."""
-
-from __future__ import annotations
-
-import json
-from pathlib import Path
-import ssl
-from urllib.parse import quote
-
-from aiohttp import ClientResponseError, ClientSession, TCPConnector
-
-
-class SandboxKubernetesClient:
-    def __init__(
-        self,
-        *,
-        api_url: str,
-        token_file: Path,
-        ca_file: Path,
-        namespace: str,
-    ) -> None:
-        self.api_url = api_url.rstrip("/")
-        self.token_file = token_file
-        self.ca_file = ca_file
-        self.namespace = namespace
-        self.session: ClientSession | None = None
-
-    async def __aenter__(self) -> SandboxKubernetesClient:
-        context = ssl.create_default_context(cafile=self.ca_file)
-        self.session = ClientSession(
-            connector=TCPConnector(ssl=context),
-            headers={
-                "Authorization": f"Bearer {self.token_file.read_text().strip()}"
-            },
-            raise_for_status=True,
-        )
-        return self
-
-    async def __aexit__(self, *_: object) -> None:
-        if self.session is not None:
-            await self.session.close()
-
-    async def get_batchsandbox(self, sandbox_id: str) -> dict[str, object] | None:
-        try:
-            response = await self._request(
-                "GET", f"{self._batchsandboxes_path()}/{quote(sandbox_id, safe='')}"
-            )
-        except ClientResponseError as error:
-            if error.status == 404:
-                return None
-            raise
-        return await response.json()
-
-    async def list_batchsandboxes(self) -> list[dict[str, object]]:
-        response = await self._request("GET", self._batchsandboxes_path())
-        document = await response.json()
-        items = document.get("items") if isinstance(document, dict) else None
-        return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
-
-    async def get_pod(self, name: str) -> dict[str, object] | None:
-        try:
-            response = await self._request(
-                "GET", f"{self._pods_path()}/{quote(name, safe='')}"
-            )
-        except ClientResponseError as error:
-            if error.status == 404:
-                return None
-            raise
-        return await response.json()
-
-    async def create_entry(self, manifest: dict[str, object]) -> dict[str, object]:
-        response = await self._request(
-            "POST", self._entries_path(), json=manifest
-        )
-        return await response.json()
-
-    async def list_entries(self) -> list[dict[str, object]]:
-        response = await self._request(
-            "GET",
-            f"{self._entries_path()}?labelSelector="
-            "app.kubernetes.io%2Fcomponent%3Dopensandbox-identity",
-        )
-        document = await response.json()
-        items = document.get("items") if isinstance(document, dict) else None
-        return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
-
-    async def get_entry(self, name: str) -> dict[str, object] | None:
-        try:
-            response = await self._request(
-                "GET", f"{self._entries_path()}/{quote(name, safe='')}"
-            )
-        except ClientResponseError as error:
-            if error.status == 404:
-                return None
-            raise
-        return await response.json()
-
-    async def delete_entry(self, name: str) -> None:
-        try:
-            response = await self._request(
-                "DELETE",
-                f"{self._entries_path()}/{quote(name, safe='')}",
-                json={"propagationPolicy": "Background"},
-            )
-            response.release()
-        except ClientResponseError as error:
-            if error.status != 404:
-                raise
-
-    async def _request(self, method: str, path: str, **kwargs: object):
-        if self.session is None:
-            raise RuntimeError("SandboxKubernetesClient is not open")
-        return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
-
-    def _batchsandboxes_path(self) -> str:
-        return (
-            "/apis/sandbox.opensandbox.io/v1alpha1/namespaces/"
-            f"{quote(self.namespace, safe='')}/batchsandboxes"
-        )
-
-    def _pods_path(self) -> str:
-        return f"/api/v1/namespaces/{quote(self.namespace, safe='')}/pods"
-
-    @staticmethod
-    def _entries_path() -> str:
-        return "/apis/spire.spiffe.io/v1alpha1/clusterstaticentries"
-
-
-def allocated_pod_name(batchsandbox: dict[str, object]) -> str | None:
-    metadata = batchsandbox.get("metadata")
-    if not isinstance(metadata, dict):
-        return None
-    annotations = metadata.get("annotations")
-    if not isinstance(annotations, dict):
-        return None
-    raw = annotations.get("sandbox.opensandbox.io/alloc-status")
-    if not isinstance(raw, str):
-        return None
-    try:
-        allocation = json.loads(raw)
-    except json.JSONDecodeError:
-        return None
-    if not isinstance(allocation, dict):
-        return None
-    pods = allocation.get("pods")
-    if not isinstance(pods, list) or len(pods) != 1 or not isinstance(pods[0], str):
-        return None
-    return pods[0]

tests/test_opensandbox_identity.py

@@ -1,53 +0,0 @@
-import json
-
-from gitea_dynamic_runner.opensandbox_identity import (
-    identity_entry,
-    is_runner_sandbox,
-    task_environment,
-)
-from gitea_dynamic_runner.sandbox_kubernetes import allocated_pod_name
-
-
-def batchsandbox():
-    return {
-        "metadata": {
-            "name": "sandbox-1",
-            "labels": {"ci.ddupan.top/runner": "true"},
-            "annotations": {
-                "sandbox.opensandbox.io/alloc-status": json.dumps(
-                    {"pods": ["ci-vm-pod-1"]}
-                )
-            },
-        },
-        "spec": {
-            "taskTemplate": {
-                "spec": {
-                    "process": {
-                        "env": [
-                            {
-                                "name": "CI_SPIFFE_ID",
-                                "value": "spiffe://ddupan.top/ci/org/repo/test",
-                            }
-                        ]
-                    }
-                }
-            }
-        },
-    }
-
-
-def test_extracts_allocation_and_identity_environment():
-    document = batchsandbox()
-    assert is_runner_sandbox(document)
-    assert allocated_pod_name(document) == "ci-vm-pod-1"
-    assert task_environment(document)["CI_SPIFFE_ID"].endswith("/org/repo/test")
-
-
-def test_entry_binds_exact_pod_agent_and_runner_uid():
-    entry = identity_entry(
-        sandbox_id="sandbox-1",
-        pod_uid="pod-uid",
-        spiffe_id="spiffe://ddupan.top/ci/org/repo/test",
-    )
-    assert entry["spec"]["parentID"].endswith("/sandbox-kata/pod/pod-uid")
-    assert entry["spec"]["selectors"] == ["unix:uid:2000"]

tests/test_opensandbox_worker.py

@@ -1,88 +0,0 @@
-import asyncio
-
-import pytest
-
-from gitea_dynamic_runner.models import RunnerRequest
-from gitea_dynamic_runner.opensandbox_worker import (
-    OpenSandboxScheduler,
-    RegistrationTokens,
-    sandbox_request,
-)
-
-
-def request(backend: str = "vm") -> RunnerRequest:
-    return RunnerRequest(
-        job_id=42,
-        run_id=7,
-        backend=backend,
-        repository="panxiao81/example",
-        job_name="publish image",
-        labels=("self-hosted", backend),
-    )
-
-
-def test_sandbox_request_uses_pool_identity_and_one_time_token_url():
-    document = sandbox_request(
-        request(), "gitea-vm-abcd", "http://scheduler/token/nonce"
-    )
-    assert document["pool"] == "ci-vm"
-    assert document["entrypoint"] == [
-        "/usr/local/libexec/gitea-opensandbox-runner"
-    ]
-    assert document["env"]["GITEA_RUNNER_LABELS"] == "self-hosted:host,vm:host"
-    assert document["env"]["GITEA_RUNNER_REGISTRATION_TOKEN_URL"].endswith(
-        "/nonce"
-    )
-    assert document["env"]["CI_SPIFFE_ID"] == (
-        "spiffe://ddupan.top/ci/panxiao81/example/publish-image-3e72cdc4a97e"
-    )
-    assert document["metadata"]["ci.ddupan.top/runner"] == "true"
-
-
-async def test_registration_token_is_single_use(monkeypatch):
-    monkeypatch.setattr(
-        "gitea_dynamic_runner.opensandbox_worker.TOKEN_BASE_URL",
-        "http://scheduler/token",
-    )
-    tokens = RegistrationTokens(b"secret\n")
-    nonce, url = await tokens.issue()
-    assert url == f"http://scheduler/token/{nonce}"
-    assert await tokens.consume(nonce) == b"secret"
-    assert await tokens.consume(nonce) is None
-
-
-class FakeOpenSandbox:
-    def __init__(self):
-        self.created = None
-        self.deleted = []
-
-    async def create(self, **document):
-        self.created = document
-        return {"id": "sandbox-1"}
-
-    async def get(self, sandbox_id):
-        return {"status": {"state": "Terminated"}}
-
-    async def delete(self, sandbox_id):
-        self.deleted.append(sandbox_id)
-
-
-async def test_scheduler_creates_monitors_and_deletes(monkeypatch):
-    monkeypatch.setattr(
-        "gitea_dynamic_runner.opensandbox_worker.TOKEN_BASE_URL",
-        "http://scheduler/token",
-    )
-    client = FakeOpenSandbox()
-    scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
-    assert await scheduler.create(request()) == "sandbox-1"
-    await scheduler.active[42]
-    assert client.created["pool"] == "ci-vm"
-    assert client.deleted == ["sandbox-1"]
-
-
-async def test_scheduler_rejects_duplicate_active_job():
-    client = FakeOpenSandbox()
-    scheduler = OpenSandboxScheduler(client, RegistrationTokens(b"secret"))
-    scheduler.active[42] = asyncio.Future()
-    with pytest.raises(ValueError, match="already"):
-        await scheduler.create(request())