Compare commits

..
Author SHA1 Message Date
panxiao81 ab8fbad310 修正 OpenSandbox SPIRE 父身份格式
test / python (pull_request) Successful in 24s
test / shell (pull_request) Successful in 39s
2026-09-18 18:28:32 +00:00
panxiao81 58b48207fe Merge pull request '重构为直接 OpenSandbox 生命周期调度' (#24) from refactor/direct-opensandbox-scheduler into main
test / python (push) Successful in 11s
test / shell (push) Successful in 16s
publish images / publish-images (push) Failing after 1s
2026-09-18 18:19:38 +00:00
panxiao81 f756a83013 修复 runner 清理脚本检查
test / shell (pull_request) Successful in 18s
test / python (pull_request) Successful in 15s
2026-09-18 18:18:10 +00:00
panxiao81 523ccbae2f 修复 runner token 清理路径
test / python (pull_request) Successful in 15s
test / shell (pull_request) Failing after 17s
2026-09-18 18:13:48 +00:00
panxiao81 f3a199e7ba 重构为直接 OpenSandbox 生命周期调度
test / python (pull_request) Successful in 13s
test / shell (pull_request) Failing after 21s
2026-09-18 18:12:10 +00:00
panxiao81 6a58c95c5c Merge pull request '接入 OpenSandbox Kata Runner 控制面' (#23) from feat/opensandbox-runner into main
test / python (push) Successful in 13s
test / shell (push) Successful in 22s
publish images / publish-images (push) Failing after 0s
2026-09-18 16:22:58 +00:00
panxiao81 2679f81cdd fix: 使用 OpenSandbox chart 服务地址
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 17s
2026-09-18 01:09:37 +00:00
panxiao81 d7ba64d6f3 接入 OpenSandbox Kata Runner 控制面
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 21s
2026-09-18 00:56:10 +00:00
panxiao81 8d70df2128 Merge pull request '在 VM Runner 中构建发布镜像' (#17) from refactor/publish-on-vm into main
test / python (push) Successful in 10s
test / shell (push) Successful in 15s
publish images / publish-images (push) Failing after 0s
Reviewed-on: #17
2026-09-16 18:29:26 +00:00
panxiao81 3e69d102dd 在 VM Runner 中发布镜像
test / python (pull_request) Successful in 14s
test / shell (pull_request) Successful in 22s
2026-09-16 18:28:33 +00:00
panxiao81 0387b2592a Merge pull request '使用 native snapshotter 发布镜像证明' (#16) from fix/publish-native-snapshotter into main
test / python (push) Successful in 10s
test / shell (push) Successful in 16s
publish images / publish-images (push) Canceled after 9m5s
Reviewed-on: #16
2026-09-16 18:20:10 +00:00
panxiao81 057731ae62 使用 native snapshotter 生成镜像证明
test / python (pull_request) Successful in 8s
test / shell (pull_request) Successful in 14s
2026-09-16 18:19:27 +00:00
panxiao81 529daa4c75 Merge pull request '修复 Pod 内嵌套 Docker overlay 构建' (#15) from fix/publish-vfs-builder into main
test / python (push) Successful in 9s
publish images / publish-images (push) Failing after 35s
test / shell (push) Successful in 16s
Reviewed-on: #15
2026-09-16 18:17:24 +00:00
panxiao81 38946e8def 使用 vfs 构建嵌套容器镜像
test / shell (pull_request) Successful in 20s
test / python (pull_request) Successful in 9s
2026-09-16 18:16:41 +00:00
panxiao81 893323a89b Merge pull request '添加 Zot 镜像构建发布流水线' (#14) from feat/image-publish-pipeline into main
test / python (push) Successful in 9s
test / shell (push) Successful in 16s
publish images / publish-images (push) Failing after 57s
Reviewed-on: #14
2026-09-16 18:12:25 +00:00
panxiao81 6f149c85e2 添加 Zot 镜像发布流水线
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 16s
2026-09-16 18:10:02 +00:00
panxiao81 852afbf02e Merge pull request '添加动态 Pod Runner 冒烟测试' (#9) from test/pod-smoke into main
test / python (push) Successful in 10s
test / shell (push) Failing after 17s
Reviewed-on: #9
2026-09-16 18:03:01 +00:00
panxiao81 6a29a10244 Merge pull request '修复 LXC 内 microVM Runner 启动链路' (#11) from fix/microvm-lxc-bootstrap into main
test / shell (push) Failing after 19s
test / python (push) Successful in 9s
Reviewed-on: #11
2026-09-16 18:02:51 +00:00
panxiao81 94a962459c Merge pull request '为 Pod Runner 提供 Ubuntu workflow 环境' (#12) from fix/ubuntu-runner-image into main
test / python (push) Successful in 8s
test / shell (push) Successful in 18s
Reviewed-on: #12
2026-09-16 18:02:16 +00:00
panxiao81 d03927d73c Merge pull request '修复 Runner SPIFFE 身份路径并补充关联日志' (#13) from fix/identity-binding-observability into main
test / shell (push) Successful in 15s
test / python (push) Successful in 8s
Reviewed-on: #13
2026-09-16 18:01:57 +00:00
panxiao81 68bb02b20a 修复 Runner 身份路径并补充关联日志
test / python (pull_request) Successful in 8s
test / shell (pull_request) Successful in 15s
2026-09-16 18:00:50 +00:00
panxiao81 787614667c 为 Pod Runner 提供 Ubuntu workflow 环境
test / python (pull_request) Successful in 11s
test / shell (pull_request) Successful in 17s
2026-09-16 17:21:41 +00:00
panxiao81 bb73a48695 将 Pod smoke 改为手动触发
test / python (pull_request) Successful in 14s
test / shell (pull_request) Successful in 20s
2026-09-16 16:42:49 +00:00
panxiao81 99ef62bfa0 Merge remote-tracking branch 'origin/main' into test/pod-smoke
dynamic Pod smoke test / pod-smoke (pull_request) Successful in 5s
test / python (pull_request) Successful in 10s
test / shell (pull_request) Successful in 18s
2026-09-16 16:06:55 +00:00
panxiao81 21aabe162f 添加动态 Pod Runner 冒烟测试
test / python (pull_request) Successful in 9s
test / shell (pull_request) Successful in 16s
dynamic Pod smoke test / pod-smoke (pull_request) Failing after 1s
2026-09-16 15:32:38 +00:00
18 changed files with 1151 additions and 91 deletions
+19
View File
@@ -0,0 +1,19 @@
name: dynamic Pod smoke test
on:
workflow_dispatch:
jobs:
pod-smoke:
name: pod-smoke
runs-on: [self-hosted, pod]
timeout-minutes: 5
steps:
- name: Verify disposable host environment
run: |
test -S /run/spire/agent-sockets/spire-agent.sock
/opt/spire/bin/spire-agent api fetch jwt \
-audience ci-smoke \
-socketPath /run/spire/agent-sockets/spire-agent.sock \
>/dev/null
test "$(id -u)" = 0
+122
View File
@@ -0,0 +1,122 @@
---
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
+5
View File
@@ -0,0 +1,5 @@
---
self-hosted-runner:
labels:
- pod
- vm
+8 -10
View File
@@ -17,25 +17,23 @@ runs-on: [self-hosted, vm]
组件:
- `controller`:接收 Gitea `workflow_job` webhook,将指定 label 的 queued job
发布到 NATS JetStream。
- `worker`:领取任务、限制并发,并通过 Pod 或 microVM backend 创建一次性环境。
- `controller`:接收 Gitea `workflow_job` webhook,直接调用 OpenSandbox Lifecycle
API,从 `ci-pod` 或 `ci-vm` Pool 创建一次性环境。
- `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 身份。
消息流使用一个 `WorkQueuePolicy` stream。相同 runner label 的所有 worker 共享同一
durable consumer;扩容只需要增加 worker 或提高单机 capacity。
当前 webhook → NATS 流程是用于尽快验证 Pod/VM 生命周期的 bootstrap 实现,不是
长期调度接口。长期目标是让 controller 作为兼容 Gitea Runner 协议的调度器直接注册、
声明 labels、领取 task,并把已领取 task 交给 Pod/VM executor;路线与迁移边界见
OpenSandbox 路径不使用 NATS。旧 Pod/microVM worker 只作为迁移期代码保留,不应重新
部署。长期 RunnerService 协议路线见
[`docs/runner-protocol-roadmap.md`](docs/runner-protocol-roadmap.md)。
## 开发
@@ -49,7 +47,7 @@ pytest
## 安全边界
- NATS 密码、webhook secret 和 Gitea registration token 只从文件读取。
- OpenSandbox API key、webhook secret 和 Gitea registration token 只从文件读取。
- registration token 不写入 seed image;worker 通过单次 nonce endpoint 交给 guest。
- guest 启动时从仅监听 microVM bridge 的 worker endpoint 获取固定版本 Runner 和配置
资产;基础镜像无需为 Runner 发布而重做。
+16 -3
View File
@@ -1,8 +1,21 @@
FROM ghcr.io/spiffe/spire-agent:1.15.3@sha256:41b0dcd8b258a69db9e2768292a060766fb76fd866e4bc925849981ea1b825ff AS spire
FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242
FROM docker.io/gitea/runner:3.5.0@sha256:66b7da94dc7dcadb2e076bec6928221336a9a637196399281c4b766fe1288242 AS runner
# The runner daemon image is intentionally minimal and does not contain the
# Node.js runtime required by JavaScript actions such as actions/checkout.
# Run the daemon in Gitea's Ubuntu workflow image so host-mode jobs and their
# actions share a GitHub Actions-compatible userspace.
FROM docker.io/gitea/runner-images:ubuntu-latest@sha256:fd911d7417bfbf0f454530e447da95b58001e1df41bbc5e1a8dd35d432575aae
USER root
COPY --from=runner /usr/local/bin/gitea-runner /usr/local/bin/gitea-runner
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 scripts/gitea-job-started /usr/local/libexec/gitea-job-started
RUN chmod 0755 /usr/local/libexec/gitea-job-started
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 /
ENTRYPOINT ["/usr/local/bin/run.sh"]
+52
View File
@@ -0,0 +1,52 @@
# 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-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 恢复主动监控,但不得为此重新引入消息
队列。
+1
View File
@@ -16,6 +16,7 @@ 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]
+1 -1
View File
@@ -7,7 +7,7 @@ runner_name=${3:?runner name is required}
runner_labels=${4:?runner labels are required}
token_file=/run/gitea-runner-registration-token
config_file=${GITEA_RUNNER_CONFIG_FILE:-/etc/gitea-runner/config-vm-bootstrap.yaml}
export HOME=${HOME:-/root}
export HOME="${HOME:-/root}"
cleanup() {
rm -f -- "$token_file"
+30
View File
@@ -0,0 +1,30 @@
#!/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
+91 -70
View File
@@ -1,61 +1,54 @@
#!/usr/bin/env python3
"""Gitea workflow_job webhook to NATS JetStream producer."""
"""Gitea workflow_job webhook to the OpenSandbox Lifecycle API."""
from __future__ import annotations
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 IdentityBinding, RunnerRequest
from .models import RunnerRequest
from .opensandbox import OpenSandboxClient
from .opensandbox_worker import OpenSandboxScheduler, RegistrationTokens
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"))
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",
)
)
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", "")):
@@ -64,58 +57,86 @@ async def webhook(request: web.Request) -> web.Response:
payload = json.loads(body)
except json.JSONDecodeError as error:
raise web.HTTPBadRequest(text="invalid JSON\n") from error
context = _webhook_context(payload)
LOG.info("workflow_job webhook received %s", context)
runner_request = RunnerRequest.from_webhook(payload)
if runner_request is not None:
await request.app["js"].publish(
f"{SUBJECT_PREFIX}.{runner_request.backend}",
runner_request.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{runner_request.job_id}-queued"},
)
return web.Response(status=202, text="queued\n")
if runner_request is None:
LOG.info("workflow_job webhook ignored %s", context)
return web.Response(status=204)
binding = IdentityBinding.from_webhook(payload)
if binding is not None:
await request.app["js"].publish(
f"{SUBJECT_PREFIX}.{binding.backend}.binding",
binding.to_json(),
headers={"Nats-Msg-Id": f"gitea-workflow-job-{binding.job_id}-in-progress"},
)
return web.Response(status=202, text="binding queued\n")
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")
return web.Response(status=204)
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):
return f"payload_type={type(payload).__name__}"
job = payload.get("workflow_job")
repository = payload.get("repository")
job = job if isinstance(job, dict) else {}
repository = repository if isinstance(repository, dict) else {}
return (
f"action={payload.get('action')!r} job_id={job.get('id')!r} "
f"run_id={job.get('run_id')!r} runner_name={job.get('runner_name')!r} "
f"repository={repository.get('full_name')!r} job_name={job.get('name')!r} "
f"labels={job.get('labels')!r}"
)
async def health(request: web.Request) -> web.Response:
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",
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,
)
app["nc"] = nc
app["js"] = nc.jetstream()
await ensure_stream(app["js"])
yield
await nc.drain()
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
yield
await scheduler.close()
def create_app() -> web.Application:
app = web.Application(client_max_size=1024 * 1024)
app.cleanup_ctx.append(nats_context)
app.cleanup_ctx.append(opensandbox_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:
web.run_app(create_app(), host=os.environ.get("LISTEN", "0.0.0.0"), port=int(os.environ.get("PORT", "8787")))
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")),
)
if __name__ == "__main__":
+79
View File
@@ -0,0 +1,79 @@
"""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)
@@ -0,0 +1,183 @@
"""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_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()
@@ -0,0 +1,163 @@
"""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)
+76 -5
View File
@@ -4,12 +4,13 @@
from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import os
from pathlib import Path
import re
import ssl
from urllib.parse import quote
import uuid
import nats
@@ -49,16 +50,26 @@ RUNNER_TOKEN_SECRET = os.environ.get("RUNNER_TOKEN_SECRET", "gitea-dynamic-runne
GITEA_INSTANCE = os.environ.get("GITEA_INSTANCE", "https://git.ddupan.top")
CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "4"))
POD_TIMEOUT = int(os.environ.get("RUNNER_POD_TIMEOUT", str(4 * 60 * 60)))
SPIFFE_PATH_SEGMENT = re.compile(r"^[A-Za-z0-9._-]+$")
def identity_path(repository: str, job_name: str) -> str:
parts = repository.split("/")
if len(parts) != 2 or not all(parts):
raise ValueError("repository must be owner/name")
encoded = [quote(part, safe="-._~") for part in (*parts, job_name)]
if not job_name.strip() or any(not part for part in encoded):
if not job_name.strip():
raise ValueError("identity components must not be empty")
return "/".join(encoded)
return "/".join(safe_identity_segment(part) for part in (*parts, job_name))
def safe_identity_segment(value: str) -> str:
"""Map arbitrary Gitea names to stable SPIFFE Operator path segments."""
if SPIFFE_PATH_SEGMENT.fullmatch(value):
return value
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-._")
slug = slug[:48].rstrip("-._") or "segment"
digest = hashlib.sha256(value.encode()).hexdigest()[:12]
return f"{slug}-{digest}"
def pod_manifest(request: RunnerRequest, pod_name: str) -> dict[str, object]:
@@ -174,36 +185,90 @@ async def run_request(message: object, client: KubernetesClient) -> None:
return
pod_name = f"gitea-pod-{uuid.uuid4().hex[:12]}"
LOG.info(
"runner request received job_id=%d run_id=%d repository=%s job_name=%r "
"backend=%s pod=%s",
request.job_id,
request.run_id,
request.repository,
request.job_name,
request.backend,
pod_name,
)
stop = asyncio.Event()
pulse = asyncio.create_task(heartbeat(message, stop))
try:
await client.create_pod(pod_manifest(request, pod_name))
LOG.info(
"runner Pod created pod=%s job_id=%d run_id=%d image=%s",
pod_name,
request.job_id,
request.run_id,
RUNNER_IMAGE,
)
if await wait_for_pod(client, pod_name):
await message.ack()
LOG.info(
"runner request acknowledged pod=%s job_id=%d result=succeeded",
pod_name,
request.job_id,
)
else:
LOG.error("runner Pod %s failed", pod_name)
LOG.error(
"runner Pod failed pod=%s job_id=%d; request will be retried",
pod_name,
request.job_id,
)
await message.nak(delay=30)
except Exception:
LOG.exception(
"runner request failed pod=%s job_id=%d; request will be retried",
pod_name,
request.job_id,
)
await message.nak(delay=30)
raise
finally:
stop.set()
await pulse
LOG.info("deleting runner Pod pod=%s job_id=%d", pod_name, request.job_id)
await client.delete_pod(pod_name)
async def bind_request(message: object, client: KubernetesClient) -> None:
binding: IdentityBinding | None = None
try:
document = json.loads(message.data)
binding = IdentityBinding(**document)
if binding.backend != "pod" or not binding.runner_name.startswith("gitea-pod-"):
raise ValueError("invalid Pod identity binding")
path = identity_path(binding.repository, binding.job_name)
LOG.info(
"identity binding received job_id=%d run_id=%d runner_name=%s "
"repository=%s job_name=%r identity_path=%s",
binding.job_id,
binding.run_id,
binding.runner_name,
binding.repository,
binding.job_name,
path,
)
await client.bind_identity(binding.runner_name, path)
except ClientResponseError as error:
if error.status == 404:
LOG.warning(
"identity binding Pod not found runner_name=%s job_id=%s; binding will be retried",
binding.runner_name if binding else None,
binding.job_id if binding else None,
)
await message.nak(delay=2)
return
LOG.exception(
"identity binding Kubernetes request failed runner_name=%s job_id=%s status=%d",
binding.runner_name if binding else None,
binding.job_id if binding else None,
error.status,
)
await message.nak(delay=30)
raise
except (json.JSONDecodeError, TypeError, ValueError) as error:
@@ -211,6 +276,12 @@ async def bind_request(message: object, client: KubernetesClient) -> None:
await message.ack()
return
await message.ack()
LOG.info(
"identity binding applied and acknowledged runner_name=%s job_id=%d identity_path=%s",
binding.runner_name,
binding.job_id,
path,
)
async def consume_requests(subscription: object, client: KubernetesClient) -> None:
@@ -0,0 +1,148 @@
"""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]
+53
View File
@@ -0,0 +1,53 @@
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-uid")
assert entry["spec"]["selectors"] == ["unix:uid:2000"]
+88
View File
@@ -0,0 +1,88 @@
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())
+16 -2
View File
@@ -16,8 +16,22 @@ def request() -> RunnerRequest:
def test_identity_path_is_meaningful_and_uri_safe():
assert pod_worker.identity_path("panxiao81/example", "publish image") == (
"panxiao81/example/publish%20image"
path = pod_worker.identity_path("panxiao81/example", "publish image")
assert path.startswith("panxiao81/example/publish-image-")
assert "%" not in path
assert all(pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part) for part in path.split("/"))
assert pod_worker.identity_path("panxiao81/example", "lint") == "panxiao81/example/lint"
assert path == pod_worker.identity_path("panxiao81/example", "publish image")
assert path != pod_worker.identity_path("panxiao81/example", "publish-image")
real_path = pod_worker.identity_path(
"panxiao81/postgresql-tenant-operator", "Run on Ubuntu"
)
assert real_path.startswith(
"panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-"
)
assert all(
pod_worker.SPIFFE_PATH_SEGMENT.fullmatch(part)
for part in real_path.split("/")
)
with pytest.raises(ValueError):
pod_worker.identity_path("invalid", "test")