From 70f0380a3da03420c521745e61b3be3678920a38 Mon Sep 17 00:00:00 2001 From: panxiao81 Date: Wed, 16 Sep 2026 14:22:09 +0000 Subject: [PATCH] =?UTF-8?q?=E5=BB=BA=E7=AB=8B=20Pod=20=E4=B8=8E=20VM=20?= =?UTF-8?q?=E8=B0=83=E5=BA=A6=E6=B6=88=E6=81=AF=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/design-principles.md | 15 ++- scripts/gitea-microvm-guest-runner | 4 +- scripts/microvm-runner-launch | 4 +- src/gitea_microvm_runner/controller.py | 48 ++++--- src/gitea_microvm_runner/models.py | 176 +++++++++++++++++++++++++ src/gitea_microvm_runner/worker.py | 31 ++++- tests/test_controller.py | 86 ++++++++++-- 7 files changed, 322 insertions(+), 42 deletions(-) create mode 100644 src/gitea_microvm_runner/models.py diff --git a/docs/design-principles.md b/docs/design-principles.md index cb5a143..61746a2 100644 --- a/docs/design-principles.md +++ b/docs/design-principles.md @@ -28,6 +28,12 @@ ephemeral runner，只执行一个 job；任务结束后注销 runner，并删 `job_id` 仅用于消息去重、状态追踪、实例关联和失败清理，不进入 workload 身份，也 不参与资源授权。 +Gitea 不保证由某次 `queued` webhook 创建的 runner 一定领取该 webhook 对应的 job。 +因此创建环境时只赋予无业务权限的启动身份。runner 实际领取任务后，controller 根据 +`in_progress` webhook 返回的 `runner_name` 和真实 job 名称绑定业务身份；环境中的 +job-start hook 必须等目标 SVID 可用后才放行 workflow 的第一步。不能依据 queued +事件提前赋予任务权限。 + ## 环境只提供运行边界 基础镜像只提供启动 runner 和执行 workflow 所需的最小环境。Docker、BuildKit、 @@ -48,14 +54,15 @@ kind 等工具由 pipeline 按需安装和启动，而不是由 controller 预 SPIFFE ID 由具有业务意义且稳定的 workflow 上下文派生： ```text -spiffe://ddupan.top/ci//// +spiffe://ddupan.top/ci/// ``` 同一种任务在不同运行中使用相同的逻辑 SPIFFE ID；每次运行取得独立、短期的 SVID。 Pod 与 VM 是可替换的执行实现，因此默认不写入 SPIFFE ID。 -workflow 和 job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界，并在 -发生冲突时拒绝创建环境，不能静默地让两个任务共享身份。 +job 名称必须经过确定性的路径规范化。规范化结果必须保留仓库边界，并在发生冲突时 +拒绝创建环境，不能静默地让两个任务共享身份。同一仓库内需要不同权限的任务应使用 +不同的 job 名称；workflow 文件只是编排载体，不进入权限身份。 ## Self-service 与授权边界 @@ -64,7 +71,7 @@ allowlist。能够修改仓库 CI 的主体本来就能修改该仓库已有任 重复审批不能形成额外的安全边界，只会破坏 self-service。 身份不等于权限。新任务可以立即取得自己的 SPIFFE ID，但默认不会因此获得 Zot、 -OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意义的 workflow/job 身份 +OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意义的 job 身份 配置授权策略。 ## 非目标设计 diff --git a/scripts/gitea-microvm-guest-runner b/scripts/gitea-microvm-guest-runner index 4405c11..7d8d7ba 100755 --- a/scripts/gitea-microvm-guest-runner +++ b/scripts/gitea-microvm-guest-runner @@ -4,7 +4,7 @@ set -eu token_url=${1:?token URL is required} instance=${2:?Gitea instance is required} runner_name=${3:?runner name is required} -runner_label=${4:?runner label is required} +runner_labels=${4:?runner labels are required} token_file=/run/gitea-runner-registration-token cleanup() { @@ -22,7 +22,7 @@ gitea-runner register \ --ephemeral \ --instance "$instance" \ --name "$runner_name" \ - --labels "$runner_label:host" \ + --labels "$runner_labels" \ --token-file "$token_file" rm -f -- "$token_file" gitea-runner daemon diff --git a/scripts/microvm-runner-launch b/scripts/microvm-runner-launch index 7f02cb9..b84d0ce 100755 --- a/scripts/microvm-runner-launch +++ b/scripts/microvm-runner-launch @@ -21,7 +21,7 @@ vm_timeout=${RUNNER_VM_TIMEOUT:-3h} cpus=${RUNNER_VM_CPUS:-4} memory=${RUNNER_VM_MEMORY:-3G} gitea_instance=${GITEA_INSTANCE:-https://git.ddupan.top} -runner_label=${RUNNER_LABEL:-kind-microvm} +runner_labels=${RUNNER_LABELS:-self-hosted:host,vm:host} token_url=${RUNNER_TOKEN_URL:-http://172.30.0.1:8787} vm_dir="$state_root/instances/$instance_id" @@ -48,7 +48,7 @@ EOF cat >"$vm_dir/user-data" < tuple[bool, str | None]: - if not isinstance(payload, dict) or payload.get("action") != "queued": - return False, None - job = payload.get("workflow_job") - if not isinstance(job, dict) or LABEL not in job.get("labels", []): - return False, None - job_id = job.get("id") - if not isinstance(job_id, int) or isinstance(job_id, bool): - return False, None - return True, str(job_id) + """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: @@ -43,7 +39,7 @@ def valid_signature(body: bytes, signature: str) -> bool: async def ensure_stream(js: object) -> None: config = StreamConfig( name=STREAM, - subjects=["ci.runner.*"], + subjects=[f"{SUBJECT_PREFIX}.>"], retention=RetentionPolicy.WORK_QUEUE, storage=StorageType.FILE, discard=DiscardPolicy.OLD, @@ -68,15 +64,25 @@ 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 - accepted, job_id = accepts(payload) - if not accepted: - return web.Response(status=204) - await request.app["js"].publish( - SUBJECT, - body, - headers={"Nats-Msg-Id": f"gitea-workflow-job-{job_id}"}, - ) - return web.Response(status=202, text="queued\n") + 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") + + 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") + + return web.Response(status=204) async def health(request: web.Request) -> web.Response: diff --git a/src/gitea_microvm_runner/models.py b/src/gitea_microvm_runner/models.py new file mode 100644 index 0000000..9b0178b --- /dev/null +++ b/src/gitea_microvm_runner/models.py @@ -0,0 +1,176 @@ +"""Validated messages shared by the controller and runner backends.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +import json + + +BACKENDS = frozenset({"pod", "vm"}) +REQUIRED_LABEL = "self-hosted" + + +@dataclass(frozen=True, slots=True) +class RunnerRequest: + """A queued Gitea job that needs one disposable runner environment.""" + + job_id: int + run_id: int + backend: str + repository: str + job_name: str + labels: tuple[str, ...] + schema_version: int = 1 + + @classmethod + def from_webhook(cls, payload: object) -> RunnerRequest | None: + if not isinstance(payload, dict) or payload.get("action") != "queued": + return None + + job = payload.get("workflow_job") + repository = payload.get("repository") + if not isinstance(job, dict) or not isinstance(repository, dict): + return None + + labels_value = job.get("labels") + if not isinstance(labels_value, list) or not all( + isinstance(label, str) for label in labels_value + ): + return None + labels = tuple(dict.fromkeys(labels_value)) + selected = BACKENDS.intersection(labels) + if REQUIRED_LABEL not in labels or len(selected) != 1: + return None + + job_id = job.get("id") + run_id = job.get("run_id") + job_name = job.get("name") + full_name = repository.get("full_name") + if not _positive_int(job_id) or not _positive_int(run_id): + return None + if not all(_nonempty(value) for value in (job_name, full_name)): + return None + + return cls( + job_id=job_id, + run_id=run_id, + backend=next(iter(selected)), + repository=full_name.strip(), + job_name=job_name.strip(), + labels=labels, + ) + + def to_json(self) -> bytes: + document = asdict(self) + document["labels"] = list(self.labels) + return json.dumps( + document, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode() + + @classmethod + def from_json(cls, body: bytes) -> RunnerRequest: + document = json.loads(body) + if not isinstance(document, dict) or document.get("schema_version") != 1: + raise ValueError("unsupported runner request") + labels = document.get("labels") + if not isinstance(labels, list) or not all( + isinstance(label, str) for label in labels + ): + raise ValueError("invalid runner labels") + try: + request = cls( + job_id=document["job_id"], + run_id=document["run_id"], + backend=document["backend"], + repository=document["repository"], + job_name=document["job_name"], + labels=tuple(labels), + ) + except KeyError as error: + raise ValueError(f"missing runner request field: {error.args[0]}") from error + if ( + not _positive_int(request.job_id) + or not _positive_int(request.run_id) + or request.backend not in BACKENDS + or not _nonempty(request.repository) + or not _nonempty(request.job_name) + or REQUIRED_LABEL not in request.labels + or request.backend not in request.labels + ): + raise ValueError("invalid runner request") + return request + + +@dataclass(frozen=True, slots=True) +class IdentityBinding: + """The actual task claimed by an ephemeral runner.""" + + job_id: int + run_id: int + backend: str + runner_name: str + repository: str + job_name: str + schema_version: int = 1 + + @classmethod + def from_webhook(cls, payload: object) -> IdentityBinding | None: + if not isinstance(payload, dict) or payload.get("action") != "in_progress": + return None + job = payload.get("workflow_job") + repository = payload.get("repository") + if not isinstance(job, dict) or not isinstance(repository, dict): + return None + + labels = job.get("labels") + if not isinstance(labels, list) or not all( + isinstance(label, str) for label in labels + ): + return None + selected = BACKENDS.intersection(labels) + if REQUIRED_LABEL not in labels or len(selected) != 1: + return None + backend = next(iter(selected)) + + job_id = job.get("id") + run_id = job.get("run_id") + runner_name = job.get("runner_name") + job_name = job.get("name") + full_name = repository.get("full_name") + if not _positive_int(job_id) or not _positive_int(run_id): + return None + if not all( + _nonempty(value) + for value in (runner_name, job_name, full_name) + ): + return None + if not runner_name.startswith(f"gitea-{backend}-"): + return None + + return cls( + job_id=job_id, + run_id=run_id, + backend=backend, + runner_name=runner_name.strip(), + repository=full_name.strip(), + job_name=job_name.strip(), + ) + + def to_json(self) -> bytes: + return json.dumps( + asdict(self), + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode() + + +def _positive_int(value: object) -> bool: + return isinstance(value, int) and not isinstance(value, bool) and value > 0 + + +def _nonempty(value: object) -> bool: + return isinstance(value, str) and bool(value.strip()) diff --git a/src/gitea_microvm_runner/worker.py b/src/gitea_microvm_runner/worker.py index 4d3543a..b25d402 100644 --- a/src/gitea_microvm_runner/worker.py +++ b/src/gitea_microvm_runner/worker.py @@ -2,6 +2,7 @@ """Capacity-bounded JetStream consumer that launches one ephemeral VM per job.""" import asyncio +import json import logging import os import secrets @@ -14,11 +15,13 @@ from aiohttp import web from nats.errors import TimeoutError from nats.js.api import AckPolicy, ConsumerConfig +from .models import RunnerRequest + LOG = logging.getLogger(__name__) CAPACITY = int(os.environ.get("RUNNER_CAPACITY", "1")) -SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.kind-microvm") +SUBJECT = os.environ.get("NATS_SUBJECT", "ci.runner.vm") STREAM = os.environ.get("NATS_STREAM", "CI_RUNNER") -DURABLE = os.environ.get("NATS_DURABLE", "kind-microvm") +DURABLE = os.environ.get("NATS_DURABLE", "vm") MAX_INFLIGHT = int(os.environ.get("RUNNER_MAX_INFLIGHT", "64")) NATS_URL = os.environ.get("NATS_URL", "tls://nats.ad.ddupan.top:4222") NATS_USER = os.environ.get("NATS_USER", "ci-worker") @@ -52,6 +55,17 @@ async def heartbeat(message: object, stop: asyncio.Event) -> None: async def run_one(message: object) -> None: + try: + request = RunnerRequest.from_json(message.data) + except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as error: + LOG.error("discarding invalid runner request: %s", error) + await message.ack() + return + if request.backend != "vm": + LOG.error("discarding %s request received by VM worker", request.backend) + await message.ack() + return + instance_id = str(uuid.uuid4()) nonce = secrets.token_urlsafe(32) async with token_lock: @@ -59,7 +73,18 @@ async def run_one(message: object) -> None: stop = asyncio.Event() pulse = asyncio.create_task(heartbeat(message, stop)) try: - process = await asyncio.create_subprocess_exec(LAUNCHER, instance_id, nonce) + process = await asyncio.create_subprocess_exec( + LAUNCHER, + instance_id, + nonce, + env={ + **os.environ, + "RUNNER_JOB_ID": str(request.job_id), + "RUNNER_RUN_ID": str(request.run_id), + "RUNNER_REPOSITORY": request.repository, + "RUNNER_JOB_NAME": request.job_name, + }, + ) return_code = await process.wait() if return_code == 0: await message.ack() diff --git a/tests/test_controller.py b/tests/test_controller.py index 9f6cf96..9419598 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -1,22 +1,88 @@ import hashlib import hmac +import json + +import pytest from gitea_microvm_runner import controller +from gitea_microvm_runner.models import IdentityBinding, RunnerRequest -def test_accepts_matching_queued_job(monkeypatch): - monkeypatch.setattr(controller, "LABEL", "kind-microvm") - assert controller.accepts({ +def queued_job(**overrides): + job = { + "id": 47, + "run_id": 12, + "name": "publish-image", + "labels": ["self-hosted", "pod"], + } + job.update(overrides) + return { "action": "queued", - "workflow_job": {"id": 47, "labels": ["linux", "kind-microvm"]}, - }) == (True, "47") + "workflow_job": job, + "repository": {"full_name": "panxiao81/example"}, + } -def test_rejects_other_actions_labels_and_boolean_id(monkeypatch): - monkeypatch.setattr(controller, "LABEL", "kind-microvm") - assert controller.accepts({"action": "completed", "workflow_job": {"id": 1, "labels": ["kind-microvm"]}}) == (False, None) - assert controller.accepts({"action": "queued", "workflow_job": {"id": 1, "labels": ["host"]}}) == (False, None) - assert controller.accepts({"action": "queued", "workflow_job": {"id": True, "labels": ["kind-microvm"]}}) == (False, None) +def test_accepts_matching_queued_job(): + assert controller.accepts(queued_job()) == (True, "47") + + +def test_rejects_other_actions_labels_and_boolean_id(): + completed = queued_job() + completed["action"] = "completed" + assert controller.accepts(completed) == (False, None) + assert controller.accepts(queued_job(labels=["self-hosted", "other"])) == (False, None) + assert controller.accepts(queued_job(labels=["self-hosted", "pod", "vm"])) == (False, None) + assert controller.accepts(queued_job(id=True)) == (False, None) + + +def test_runner_request_contains_stable_identity_context(): + request = RunnerRequest.from_webhook(queued_job()) + assert request is not None + assert request.backend == "pod" + assert request.repository == "panxiao81/example" + assert request.job_name == "publish-image" + assert request.job_id == 47 + assert json.loads(request.to_json()) == { + "backend": "pod", + "job_id": 47, + "job_name": "publish-image", + "labels": ["self-hosted", "pod"], + "repository": "panxiao81/example", + "run_id": 12, + "schema_version": 1, + } + + +def test_runner_request_requires_complete_identity_context(): + assert RunnerRequest.from_webhook(queued_job(run_id=None)) is None + assert RunnerRequest.from_webhook(queued_job(name=" ")) is None + + +def test_runner_request_json_round_trip_and_validation(): + request = RunnerRequest.from_webhook(queued_job()) + assert request is not None + assert RunnerRequest.from_json(request.to_json()) == request + with pytest.raises(ValueError, match="unsupported"): + RunnerRequest.from_json(b'{"schema_version":2}') + + +def test_identity_binding_uses_actual_runner_assignment(): + payload = queued_job(runner_name="gitea-pod-6c47d03d") + payload["action"] = "in_progress" + binding = IdentityBinding.from_webhook(payload) + assert binding is not None + assert binding.backend == "pod" + assert binding.runner_name == "gitea-pod-6c47d03d" + assert binding.repository == "panxiao81/example" + assert binding.job_name == "publish-image" + assert json.loads(binding.to_json())["job_id"] == 47 + + +def test_identity_binding_rejects_runner_from_another_pool(): + payload = queued_job(runner_name="gitea-vm-6c47d03d") + payload["action"] = "in_progress" + assert IdentityBinding.from_webhook(payload) is None def test_signature_accepts_gitea_and_prefixed_forms(tmp_path, monkeypatch):