diff --git a/docs/design-principles.md b/docs/design-principles.md index 32f9040..3f87fe8 100644 --- a/docs/design-principles.md +++ b/docs/design-principles.md @@ -94,4 +94,5 @@ OpenBao 或其他资源的特殊权限。资源所有者在资源端按照有意 基础设施协议优先使用上游维护的成熟客户端，不在 controller 内重复实现认证、连接、 资源编码或错误语义。Kubernetes 使用 `client-go`，NATS JetStream 使用 `nats.go`， Gitea RunnerService 使用 `actionslib`，OpenSandbox Lifecycle API 使用官方 Go SDK； -自定义代码只保留领域模型、reconcile 规则及上游客户端未覆盖的最小适配层。 +SPIFFE Workload API 与 mTLS 使用 `go-spiffe`。自定义代码只保留领域模型、reconcile +规则及上游客户端未覆盖的最小适配层。 diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 52f0899..f748ddf 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -28,6 +28,13 @@ controller 暴露兼容 RunnerService 的 facade：`FetchTask` 只返回已分 ID，以及由 controller 密钥确定性生成的 assignment HMAC capability；该 capability 只绑定执行实例，不参与 Zot/OpenBao 等业务授权。 +executor 为官方 runner 生成与 v3.5.0 schema 一致的一次性 `.runner` 文件，并以 +`daemon --once` 启动。runner 只访问 executor 内的 loopback HTTP proxy；proxy 使用 +`go-spiffe` 从 Workload API 持续取得和轮换 X509-SVID，再以 mTLS 连接 controller +facade，并严格校验 facade 的 SPIFFE ID。这样无需修改 runner 或把静态客户端证书写入 +镜像。assignment capability 会进入一次性 executor 环境，但不会进入 label、annotation +或 OpenSandbox metadata；它只对该 assignment 有效，并且不能绕过 SPIFFE 身份校验。 + 这与“收到 webhook 后临时注册另一个 act_runner”不同。`FetchTask` 已经完成任务分配， 不能再期待 Gitea 把同一个 task 分配给随后启动的 runner。协议调度器必须让 executor 执行已经领取的 task，并继续完成日志、状态、心跳、取消和最终结果上报。 @@ -67,6 +74,8 @@ ID，以及由 controller 密钥确定性生成的 assignment HMAC capability； - consumer 在 executor 使用上述 facade 成功 claim task 后确认 assignment；无需把完整 task 写入 Pod annotation、OpenSandbox metadata 或环境变量。 - Pod 与 VM 共享 task/executor 协议，只有环境创建和销毁实现不同。 +- 两种 backend 都注入同一份 runner bootstrap 环境；Pod 仍由 homelab Kubernetes 原生 + 创建，只有 VM 经 OpenSandbox 创建，bootstrap 机制不改变 backend 边界。 ## 实现顺序 diff --git a/go.mod b/go.mod index f236b3b..aa2b32b 100644 --- a/go.mod +++ b/go.mod @@ -7,6 +7,8 @@ require ( gitea.dev/actionslib v1.0.0 github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5 github.com/nats-io/nats.go v1.54.0 + github.com/spiffe/go-spiffe/v2 v2.8.2 + golang.org/x/sync v0.23.0 google.golang.org/protobuf v1.36.12 k8s.io/api v0.37.0 k8s.io/apimachinery v0.37.0 @@ -14,9 +16,11 @@ require ( ) require ( + github.com/Microsoft/go-winio v0.6.2 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.13.0 // indirect github.com/fxamacker/cbor/v2 v2.9.1 // indirect + github.com/go-jose/go-jose/v4 v4.1.5 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-openapi/jsonpointer v1.0.0 // indirect github.com/go-openapi/jsonreference v1.0.0 // indirect @@ -49,11 +53,12 @@ require ( golang.org/x/crypto v0.57.0 // indirect golang.org/x/net v0.58.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.23.0 // indirect golang.org/x/sys v0.48.0 // indirect golang.org/x/term v0.46.0 // indirect golang.org/x/text v0.42.0 // indirect golang.org/x/time v0.15.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect + google.golang.org/grpc v1.79.3 // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect k8s.io/klog/v2 v2.140.0 // indirect diff --git a/go.sum b/go.sum index 0f54ddc..62546f1 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,12 @@ connectrpc.com/connect v1.20.0 h1:6TNDAB+WeNd2uolWNlYczB5E0KNNaVMNUEx8JEUsPmQ= connectrpc.com/connect v1.20.0/go.mod h1:A2ygJrukXwWy32vkCAAHNVguZrqZ+jeZ9rGRnGR4dN4= gitea.dev/actionslib v1.0.0 h1:l0oFJP+P4Ds1rlCI5zk618dYkuBc2mU7Gz5wPeG0lZY= gitea.dev/actionslib v1.0.0/go.mod h1:6O8YHkqVTKSR0LL2e5VhIDePYzGTZCbfmSVqJWEhk9g= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5 h1:7mZNkBh4VaI+i4VOTJIknLCBkbGizTkdHYqgcMqeDBM= github.com/alibaba/OpenSandbox/sdks/sandbox/go v1.0.5/go.mod h1:w0nIMCTL1L3oSS67ABFOdYxrezYQZhk01UtmKSi2UhA= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= @@ -12,8 +16,12 @@ github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bF github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ= github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/go-jose/go-jose/v4 v4.1.5 h1:RjgjO2LOtWOJKUC5wpwY9LR3B3vwVAz6JS2YHfYU6eA= +github.com/go-jose/go-jose/v4 v4.1.5/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s= github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y= github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkrAdlAVWnpY= @@ -48,6 +56,8 @@ github.com/go-openapi/testify/enable/yaml/v2 v2.6.0 h1:gGHwAJ0R/5jU8BEGDbfRNR3hL github.com/go-openapi/testify/enable/yaml/v2 v2.6.0/go.mod h1:tY+St1SGq4NFl0QIqdTY4aEdbChAHxhyB77XQi9iJCo= github.com/go-openapi/testify/v2 v2.6.0 h1:5PKH2HE7YJ/LuRPQGvSxBRlFXNQhSetBLlGAgUEu3ug= github.com/go-openapi/testify/v2 v2.6.0/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo= github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= @@ -80,6 +90,8 @@ github.com/sirupsen/logrus v1.10.2 h1:G2SED73/qrAu6YwbdxOD6peLkCBI3z7L+ykJFTXJBB github.com/sirupsen/logrus v1.10.2/go.mod h1:SLEg8TqYulVKKfIGHldVp2K2aYz2DKSVBq4g/H5bR7Q= github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spiffe/go-spiffe/v2 v2.8.2 h1:jUEsvCMD6fH25J8K/w3q/XnIx8W1lb8+YLaEEHIjHmc= +github.com/spiffe/go-spiffe/v2 v2.8.2/go.mod h1:w2CLWKLMTX/PPYUEUPv3ltH0RXsw5S8suwNF46w9/Aw= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= @@ -88,6 +100,18 @@ github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWD github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= +go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8= +go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0= +go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs= +go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= +go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= +go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= +go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= +go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= +go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw= @@ -110,6 +134,12 @@ golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= +google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE= +google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo= diff --git a/internal/opensandboxbackend/backend.go b/internal/opensandboxbackend/backend.go index 3b098c5..2cad2de 100644 --- a/internal/opensandboxbackend/backend.go +++ b/internal/opensandboxbackend/backend.go @@ -61,7 +61,7 @@ func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Exe return executor(result.Items[0]), nil } -func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, metadata taskworker.Metadata) (*taskworker.Executor, error) { +func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, launch taskworker.LaunchSpec) (*taskworker.Executor, error) { if err := b.validate(); err != nil { return nil, err } @@ -69,10 +69,11 @@ func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignmen return nil, fmt.Errorf("OpenSandbox backend cannot create %q assignment", assignment.Backend) } environment := clone(b.Config.Env) - environment["CI_ASSIGNMENT_ID"] = assignment.ID - environment["CI_SPIFFE_ID"] = assignment.Identity.SPIFFEID - sandboxMetadata := clone(metadata.Annotations) - for key, value := range metadata.Labels { + for key, value := range launch.Environment { + environment[key] = value + } + sandboxMetadata := clone(launch.Metadata.Annotations) + for key, value := range launch.Metadata.Labels { sandboxMetadata[key] = value } request := opensandbox.CreateSandboxRequest{ diff --git a/internal/opensandboxbackend/backend_test.go b/internal/opensandboxbackend/backend_test.go index cc879f5..07ee979 100644 --- a/internal/opensandboxbackend/backend_test.go +++ b/internal/opensandboxbackend/backend_test.go @@ -62,13 +62,22 @@ func TestFindRecoversSandboxByMetadata(t *testing.T) { func TestCreateUsesPoolAndPersistsRecoveryMetadata(t *testing.T) { lifecycle := &fakeLifecycle{} metadata := taskworker.BackendMetadata(assignment()) - executor, err := backend(lifecycle).Create(context.Background(), assignment(), metadata) + executor, err := backend(lifecycle).Create(context.Background(), assignment(), taskworker.LaunchSpec{ + Metadata: metadata, + Environment: map[string]string{ + "CI_SPIFFE_ID": assignment().Identity.SPIFFEID, + "CI_RUNNER_CAPABILITY": "capability", + }, + }) if err != nil { t.Fatal(err) } if lifecycle.created.Extensions["poolRef"] != "ci-vm" || lifecycle.created.Metadata[assignmentMetadata] != assignment().ID || lifecycle.created.Env["CI_SPIFFE_ID"] != assignment().Identity.SPIFFEID || executor.Name != "sandbox-42" { t.Fatalf("request=%#v executor=%#v", lifecycle.created, executor) } + if lifecycle.created.Env["CI_RUNNER_CAPABILITY"] != "capability" { + t.Fatalf("environment = %#v", lifecycle.created.Env) + } } func TestBindIdentityVerifiesPersistedMetadata(t *testing.T) { diff --git a/internal/podbackend/backend.go b/internal/podbackend/backend.go index 4ad0489..98e8f01 100644 --- a/internal/podbackend/backend.go +++ b/internal/podbackend/backend.go @@ -35,6 +35,7 @@ type PodManifest struct { Image string ServiceAccount string Args []string + Environment map[string]string } // IdentityEntry is a ClusterStaticEntry pinned to one concrete Pod UID. @@ -89,24 +90,25 @@ func (b Backend) Find(ctx context.Context, assignmentID string) (*taskworker.Exe return executor(pods[0]), nil } -func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, metadata taskworker.Metadata) (*taskworker.Executor, error) { +func (b Backend) Create(ctx context.Context, assignment taskassignment.Assignment, launch taskworker.LaunchSpec) (*taskworker.Executor, error) { if err := b.validate(); err != nil { return nil, err } if assignment.Backend != taskassignment.BackendPod { return nil, fmt.Errorf("Pod backend cannot create %q assignment", assignment.Backend) } - labels := clone(metadata.Labels) + labels := clone(launch.Metadata.Labels) labels["app.kubernetes.io/name"] = "gitea-dynamic-runner" labels["app.kubernetes.io/component"] = "executor" pod, err := b.API.CreatePod(ctx, PodManifest{ Name: assignment.ID, Namespace: b.Config.Namespace, Labels: labels, - Annotations: clone(metadata.Annotations), + Annotations: clone(launch.Metadata.Annotations), Image: b.Config.Image, ServiceAccount: b.Config.ServiceAccount, Args: append(append([]string{}, b.Config.ExecutorArgs...), assignment.ID), + Environment: clone(launch.Environment), }) if err != nil { return nil, fmt.Errorf("create assignment Pod: %w", err) diff --git a/internal/podbackend/backend_test.go b/internal/podbackend/backend_test.go index 6ae80d0..201f90f 100644 --- a/internal/podbackend/backend_test.go +++ b/internal/podbackend/backend_test.go @@ -75,7 +75,9 @@ func TestFindRecoversPodByAssignmentLabel(t *testing.T) { func TestCreateUsesDeterministicNameAndRecoveryMetadata(t *testing.T) { api := &fakeAPI{} metadata := taskworker.BackendMetadata(assignment()) - executor, err := backend(api).Create(context.Background(), assignment(), metadata) + executor, err := backend(api).Create(context.Background(), assignment(), taskworker.LaunchSpec{ + Metadata: metadata, Environment: map[string]string{"CI_RUNNER_CAPABILITY": "capability"}, + }) if err != nil { t.Fatal(err) } @@ -85,6 +87,9 @@ func TestCreateUsesDeterministicNameAndRecoveryMetadata(t *testing.T) { if api.created.Annotations["ci.ddupan.top/spiffe-id"] != assignment().Identity.SPIFFEID { t.Fatalf("annotations = %#v", api.created.Annotations) } + if api.created.Environment["CI_RUNNER_CAPABILITY"] != "capability" { + t.Fatalf("environment = %#v", api.created.Environment) + } if len(api.created.Args) != 2 || api.created.Args[1] != "gitea-task-42" || executor.IdentityTarget != "pod-uid" { t.Fatalf("args=%v executor=%#v", api.created.Args, executor) } diff --git a/internal/podbackend/client.go b/internal/podbackend/client.go index 36f0fef..fe1d46b 100644 --- a/internal/podbackend/client.go +++ b/internal/podbackend/client.go @@ -59,6 +59,10 @@ func (c *Client) ListPods(ctx context.Context, namespace, selector string) ([]Po } func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, error) { + environment := make([]corev1.EnvVar, 0, len(manifest.Environment)) + for name, value := range manifest.Environment { + environment = append(environment, corev1.EnvVar{Name: name, Value: value}) + } document := &corev1.Pod{ ObjectMeta: metav1.ObjectMeta{ Name: manifest.Name, Namespace: manifest.Namespace, @@ -68,7 +72,7 @@ func (c *Client) CreatePod(ctx context.Context, manifest PodManifest) (Pod, erro ServiceAccountName: manifest.ServiceAccount, RestartPolicy: corev1.RestartPolicyNever, Containers: []corev1.Container{{ - Name: "executor", Image: manifest.Image, Args: manifest.Args, + Name: "executor", Image: manifest.Image, Args: manifest.Args, Env: environment, SecurityContext: &corev1.SecurityContext{Privileged: boolPointer(true)}, VolumeMounts: []corev1.VolumeMount{{ Name: "spire-agent-socket", MountPath: "/run/spire/agent-sockets", ReadOnly: true, diff --git a/internal/podbackend/client_test.go b/internal/podbackend/client_test.go index fc07259..36ea21d 100644 --- a/internal/podbackend/client_test.go +++ b/internal/podbackend/client_test.go @@ -25,7 +25,7 @@ func TestClientPodLifecycleUsesTypedClient(t *testing.T) { Name: "gitea-task-42", Namespace: "gitea-actions", Labels: map[string]string{assignmentLabel: "gitea-task-42"}, Image: "zot/ci-executor:main", ServiceAccount: "gitea-task-executor", - Args: []string{"execute", "gitea-task-42"}, + Args: []string{"execute", "gitea-task-42"}, Environment: map[string]string{"CI_RUNNER_CAPABILITY": "capability"}, }) if err != nil { t.Fatal(err) @@ -34,6 +34,9 @@ func TestClientPodLifecycleUsesTypedClient(t *testing.T) { if err != nil { t.Fatal(err) } + if got := pod.Spec.Containers[0].Env; len(got) != 1 || got[0].Name != "CI_RUNNER_CAPABILITY" || got[0].Value != "capability" { + t.Fatalf("environment = %#v", got) + } pod.UID = types.UID("pod-uid") pod.Status.Phase = corev1.PodRunning if _, err := client.Kubernetes.CoreV1().Pods("gitea-actions").Update(context.Background(), pod, metav1.UpdateOptions{}); err != nil { diff --git a/internal/runnerbootstrap/bootstrap.go b/internal/runnerbootstrap/bootstrap.go new file mode 100644 index 0000000..59c64f8 --- /dev/null +++ b/internal/runnerbootstrap/bootstrap.go @@ -0,0 +1,89 @@ +// Package runnerbootstrap configures an unmodified, one-shot Gitea Runner to +// consume exactly the task assigned by the controller facade. +package runnerbootstrap + +import ( + "encoding/json" + "errors" + "fmt" + "net/url" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" +) + +const ( + EnvAssignmentID = "CI_ASSIGNMENT_ID" + EnvCapability = "CI_RUNNER_CAPABILITY" + EnvFacadeURL = "CI_RUNNER_FACADE_URL" + EnvFacadeID = "CI_RUNNER_FACADE_SPIFFE_ID" + EnvSPIFFEID = "CI_SPIFFE_ID" + EnvBackend = "CI_RUNNER_BACKEND" +) + +// Bootstrap emits assignment-scoped launch configuration. FacadeURL is the +// controller endpoint reached by the local SPIFFE proxy, not by Runner itself. +type Bootstrap struct { + Capabilities runnerfacade.Capabilities + FacadeURL string + FacadeSPIFFEID string +} + +func (b Bootstrap) Environment(assignment taskassignment.Assignment) (map[string]string, error) { + if assignment.ID == "" || assignment.Identity.SPIFFEID == "" || b.FacadeSPIFFEID == "" { + return nil, errors.New("assignment ID and SPIFFE ID are required") + } + parsed, err := url.Parse(b.FacadeURL) + if err != nil || parsed.Scheme != "https" || parsed.Host == "" { + return nil, fmt.Errorf("runner facade URL must be an absolute https URL") + } + capability := b.Capabilities.Issue(assignment.ID) + if capability == "" { + return nil, errors.New("runner capability issuer is not configured") + } + return map[string]string{ + EnvAssignmentID: assignment.ID, + EnvCapability: capability, + EnvFacadeURL: b.FacadeURL, + EnvFacadeID: b.FacadeSPIFFEID, + EnvSPIFFEID: assignment.Identity.SPIFFEID, + EnvBackend: string(assignment.Backend), + }, nil +} + +// Registration mirrors Gitea Runner v3.5.0's registration file schema. ID is +// intentionally zero: the facade authenticates UUID and token and never uses +// the server-issued runner database ID. +type Registration struct { + Warning string `json:"WARNING"` + ID int64 `json:"id"` + UUID string `json:"uuid"` + Name string `json:"name"` + Token string `json:"token"` + Address string `json:"address"` + Labels []string `json:"labels"` + Ephemeral bool `json:"ephemeral"` +} + +func RegistrationJSON(assignmentID, capability, localProxyURL string, backend taskassignment.Backend) ([]byte, error) { + if assignmentID == "" || capability == "" { + return nil, errors.New("assignment ID and runner capability are required") + } + parsed, err := url.Parse(localProxyURL) + if err != nil || parsed.Scheme != "http" || parsed.Host == "" { + return nil, errors.New("local runner proxy URL must be an absolute http URL") + } + if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM { + return nil, fmt.Errorf("unsupported runner backend %q", backend) + } + registration := Registration{ + Warning: "Generated for one preassigned task by gitea-dynamic-runner.", + UUID: assignmentID, Name: assignmentID, Token: capability, + Address: localProxyURL, Labels: []string{"self-hosted", string(backend)}, Ephemeral: true, + } + data, err := json.MarshalIndent(registration, "", " ") + if err != nil { + return nil, err + } + return append(data, '\n'), nil +} diff --git a/internal/runnerbootstrap/bootstrap_test.go b/internal/runnerbootstrap/bootstrap_test.go new file mode 100644 index 0000000..e3f60c4 --- /dev/null +++ b/internal/runnerbootstrap/bootstrap_test.go @@ -0,0 +1,74 @@ +package runnerbootstrap + +import ( + "encoding/json" + "testing" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/runnerfacade" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" +) + +func testBootstrap(t *testing.T) Bootstrap { + t.Helper() + capabilities, err := runnerfacade.NewCapabilities([]byte("0123456789abcdef0123456789abcdef")) + if err != nil { + t.Fatal(err) + } + return Bootstrap{ + Capabilities: capabilities, + FacadeURL: "https://runner-facade.gitea-actions.svc:8443", + FacadeSPIFFEID: "spiffe://ddupan.top/ns/gitea-actions/sa/gitea-dynamic-runner", + } +} + +func testAssignment() taskassignment.Assignment { + return taskassignment.Assignment{ + ID: "gitea-task-42", Backend: taskassignment.BackendPod, + Identity: taskidentity.Identity{SPIFFEID: "spiffe://ddupan.top/ci/owner/repo/publish"}, + } +} + +func TestEnvironmentIsDeterministicAndAssignmentScoped(t *testing.T) { + bootstrap := testBootstrap(t) + first, err := bootstrap.Environment(testAssignment()) + if err != nil { + t.Fatal(err) + } + second, err := bootstrap.Environment(testAssignment()) + if err != nil { + t.Fatal(err) + } + if first[EnvCapability] == "" || first[EnvCapability] != second[EnvCapability] { + t.Fatalf("capabilities = %q, %q", first[EnvCapability], second[EnvCapability]) + } + if first[EnvAssignmentID] != "gitea-task-42" || first[EnvSPIFFEID] != testAssignment().Identity.SPIFFEID { + t.Fatalf("environment = %#v", first) + } + if first[EnvBackend] != "pod" || first[EnvFacadeID] == "" { + t.Fatalf("environment = %#v", first) + } +} + +func TestRegistrationMatchesOfficialRunnerSchema(t *testing.T) { + data, err := RegistrationJSON("gitea-task-42", "capability", "http://127.0.0.1:8080", taskassignment.BackendVM) + if err != nil { + t.Fatal(err) + } + var registration Registration + if err := json.Unmarshal(data, ®istration); err != nil { + t.Fatal(err) + } + if registration.UUID != "gitea-task-42" || registration.Token != "capability" || registration.Address != "http://127.0.0.1:8080" || !registration.Ephemeral { + t.Fatalf("registration = %#v", registration) + } + if len(registration.Labels) != 2 || registration.Labels[0] != "self-hosted" || registration.Labels[1] != "vm" { + t.Fatalf("labels = %#v", registration.Labels) + } +} + +func TestRegistrationRejectsNonLocalTLSAddress(t *testing.T) { + if _, err := RegistrationJSON("id", "capability", "https://facade.example", taskassignment.BackendPod); err == nil { + t.Fatal("expected local proxy URL validation error") + } +} diff --git a/internal/runnerbootstrap/executor.go b/internal/runnerbootstrap/executor.go new file mode 100644 index 0000000..8ab8f1c --- /dev/null +++ b/internal/runnerbootstrap/executor.go @@ -0,0 +1,116 @@ +package runnerbootstrap + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "time" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment" +) + +type ExecutorConfig struct { + AssignmentID string + Capability string + Backend taskassignment.Backend + FacadeURL string + FacadeSPIFFEID string + WorkloadAPIAddr string + RunnerBinary string + ListenAddress string + WorkDir string + Stdout *os.File + Stderr *os.File +} + +// RunExecutor runs the SPIFFE proxy and one unmodified official Runner process. +// The generated registration file exists only in the executor's temporary +// work directory and the runner exits after its preassigned task. +func RunExecutor(ctx context.Context, config ExecutorConfig) error { + if config.RunnerBinary == "" { + config.RunnerBinary = "gitea-runner" + } + if config.ListenAddress == "" { + config.ListenAddress = "127.0.0.1:0" + } + listener, err := net.Listen("tcp", config.ListenAddress) + if err != nil { + return fmt.Errorf("listen for local runner proxy: %w", err) + } + defer listener.Close() + address, ok := listener.Addr().(*net.TCPAddr) + if !ok || !address.IP.IsLoopback() { + return errors.New("runner proxy must listen on a loopback address") + } + + proxy, err := NewProxy(ctx, config.FacadeURL, config.FacadeSPIFFEID, config.WorkloadAPIAddr) + if err != nil { + return err + } + defer proxy.Close() + + workDir := config.WorkDir + removeWorkDir := false + if workDir == "" { + workDir, err = os.MkdirTemp("", "gitea-task-runner-") + if err != nil { + return fmt.Errorf("create runner work directory: %w", err) + } + removeWorkDir = true + } + if removeWorkDir { + defer os.RemoveAll(workDir) + } + registration, err := RegistrationJSON( + config.AssignmentID, config.Capability, "http://"+listener.Addr().String(), config.Backend, + ) + if err != nil { + return err + } + if err := os.WriteFile(filepath.Join(workDir, ".runner"), registration, 0o600); err != nil { + return fmt.Errorf("write one-shot runner registration: %w", err) + } + + server := &http.Server{Handler: proxy.Handler, ReadHeaderTimeout: 10 * time.Second} + serverErrors := make(chan error, 1) + go func() { serverErrors <- server.Serve(listener) }() + + command := exec.CommandContext(ctx, config.RunnerBinary, "daemon", "--once") + command.Dir = workDir + command.Stdout = config.Stdout + command.Stderr = config.Stderr + runnerErr := command.Run() + shutdownContext, cancel := context.WithTimeout(context.Background(), 10*time.Second) + shutdownErr := server.Shutdown(shutdownContext) + cancel() + serverErr := <-serverErrors + if errors.Is(serverErr, http.ErrServerClosed) { + serverErr = nil + } + return errors.Join(runnerErr, shutdownErr, serverErr) +} + +// ExecutorConfigFromEnvironment reads the non-secret image configuration and +// the assignment-scoped values injected by the backend. The Workload API +// address follows SPIFFE_ENDPOINT_SOCKET through go-spiffe when not set here. +func ExecutorConfigFromEnvironment() (ExecutorConfig, error) { + backend := taskassignment.Backend(os.Getenv(EnvBackend)) + if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM { + return ExecutorConfig{}, fmt.Errorf("invalid %s %q", EnvBackend, backend) + } + config := ExecutorConfig{ + AssignmentID: os.Getenv(EnvAssignmentID), Capability: os.Getenv(EnvCapability), + Backend: backend, FacadeURL: os.Getenv(EnvFacadeURL), FacadeSPIFFEID: os.Getenv(EnvFacadeID), + RunnerBinary: os.Getenv("GITEA_RUNNER_BINARY"), ListenAddress: "127.0.0.1:0", + Stdout: os.Stdout, Stderr: os.Stderr, + } + if config.AssignmentID == "" || config.Capability == "" || config.FacadeURL == "" || config.FacadeSPIFFEID == "" { + return ExecutorConfig{}, errors.New("complete runner assignment and facade environment is required") + } + return config, nil +} diff --git a/internal/runnerbootstrap/proxy.go b/internal/runnerbootstrap/proxy.go new file mode 100644 index 0000000..e3ed475 --- /dev/null +++ b/internal/runnerbootstrap/proxy.go @@ -0,0 +1,56 @@ +package runnerbootstrap + +import ( + "context" + "fmt" + "net/http" + "net/http/httputil" + "net/url" + + "github.com/spiffe/go-spiffe/v2/spiffeid" + "github.com/spiffe/go-spiffe/v2/spiffetls/tlsconfig" + "github.com/spiffe/go-spiffe/v2/workloadapi" +) + +type Proxy struct { + Handler http.Handler + source *workloadapi.X509Source +} + +// NewProxy obtains rotating X509-SVIDs from the Workload API and authorizes +// one exact controller identity. The official Runner talks plain HTTP only to +// this executor-local handler. +func NewProxy(ctx context.Context, facadeURL, facadeSPIFFEID, workloadAPIAddr string) (*Proxy, error) { + target, err := url.Parse(facadeURL) + if err != nil || target.Scheme != "https" || target.Host == "" { + return nil, fmt.Errorf("runner facade URL must be an absolute https URL") + } + serverID, err := spiffeid.FromString(facadeSPIFFEID) + if err != nil { + return nil, fmt.Errorf("parse runner facade SPIFFE ID: %w", err) + } + options := []workloadapi.X509SourceOption{} + if workloadAPIAddr != "" { + options = append(options, workloadapi.WithClientOptions(workloadapi.WithAddr(workloadAPIAddr))) + } + source, err := workloadapi.NewX509Source(ctx, options...) + if err != nil { + return nil, fmt.Errorf("open SPIFFE Workload API X509 source: %w", err) + } + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = tlsconfig.MTLSClientConfig(source, source, tlsconfig.AuthorizeID(serverID)) + return &Proxy{Handler: NewProxyHandler(target, transport), source: source}, nil +} + +func NewProxyHandler(target *url.URL, transport http.RoundTripper) http.Handler { + proxy := httputil.NewSingleHostReverseProxy(target) + proxy.Transport = transport + return proxy +} + +func (p *Proxy) Close() error { + if p == nil || p.source == nil { + return nil + } + return p.source.Close() +} diff --git a/internal/taskworker/worker.go b/internal/taskworker/worker.go index 863564d..69de29b 100644 --- a/internal/taskworker/worker.go +++ b/internal/taskworker/worker.go @@ -34,11 +34,26 @@ type Metadata struct { Annotations map[string]string } +// LaunchSpec contains the durable resource metadata and the short-lived +// executor environment. Environment values configure the one-shot runner but +// are deliberately excluded from labels and annotations. +type LaunchSpec struct { + Metadata Metadata + Environment map[string]string +} + +// Bootstrap produces assignment-scoped executor configuration. Implementations +// must be deterministic so a redelivery after controller restart creates the +// same credentials without storing another lifecycle record. +type Bootstrap interface { + Environment(taskassignment.Assignment) (map[string]string, error) +} + // Backend is implemented by the native Pod and OpenSandbox adapters. // Every method must be idempotent. type Backend interface { Find(context.Context, string) (*Executor, error) - Create(context.Context, taskassignment.Assignment, Metadata) (*Executor, error) + Create(context.Context, taskassignment.Assignment, LaunchSpec) (*Executor, error) BindIdentity(context.Context, *Executor, taskidentity.Identity) error Delete(context.Context, *Executor) error } @@ -53,16 +68,17 @@ type TaskState interface { // Worker has no correctness-critical in-memory state. Handle may be called // again for the same assignment after any operation. type Worker struct { - Backend Backend - Tasks TaskState + Backend Backend + Tasks TaskState + Bootstrap Bootstrap } // Accept completes the durable handoff from JetStream to the backend. Once it // returns true, all recovery information exists in Kubernetes/OpenSandbox and // the assignment message can be acknowledged immediately. func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment) (bool, error) { - if w.Backend == nil || w.Tasks == nil { - return false, errors.New("backend and Gitea task state are required") + if w.Backend == nil || w.Tasks == nil || w.Bootstrap == nil { + return false, errors.New("backend, Gitea task state, and runner bootstrap are required") } if assignment.ID == "" || assignment.Task == nil { return false, errors.New("valid assignment is required") @@ -84,7 +100,11 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment return true, nil } if executor == nil { - executor, err = w.Backend.Create(ctx, assignment, BackendMetadata(assignment)) + launch, launchErr := w.launchSpec(assignment) + if launchErr != nil { + return false, launchErr + } + executor, err = w.Backend.Create(ctx, assignment, launch) if err != nil { return false, err } @@ -101,8 +121,8 @@ func (w Worker) Accept(ctx context.Context, assignment taskassignment.Assignment // Handle performs one reconciliation. Done means the queue message may be // acknowledged. A false result should remain pending and be reconciled again. func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment) (done bool, err error) { - if w.Backend == nil || w.Tasks == nil { - return false, errors.New("backend and Gitea task state are required") + if w.Backend == nil || w.Tasks == nil || w.Bootstrap == nil { + return false, errors.New("backend, Gitea task state, and runner bootstrap are required") } if assignment.ID == "" || assignment.Task == nil { return false, errors.New("valid assignment is required") @@ -126,7 +146,11 @@ func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment } if executor == nil { - executor, err = w.Backend.Create(ctx, assignment, BackendMetadata(assignment)) + launch, launchErr := w.launchSpec(assignment) + if launchErr != nil { + return false, launchErr + } + executor, err = w.Backend.Create(ctx, assignment, launch) if err != nil { return false, err } @@ -154,6 +178,14 @@ func (w Worker) Handle(ctx context.Context, assignment taskassignment.Assignment } } +func (w Worker) launchSpec(assignment taskassignment.Assignment) (LaunchSpec, error) { + environment, err := w.Bootstrap.Environment(assignment) + if err != nil { + return LaunchSpec{}, err + } + return LaunchSpec{Metadata: BackendMetadata(assignment), Environment: environment}, nil +} + // BackendMetadata is the shared metadata contract for Pods and OpenSandbox. func BackendMetadata(assignment taskassignment.Assignment) Metadata { return Metadata{ diff --git a/internal/taskworker/worker_test.go b/internal/taskworker/worker_test.go index 4d1a5fd..c03be14 100644 --- a/internal/taskworker/worker_test.go +++ b/internal/taskworker/worker_test.go @@ -17,8 +17,14 @@ type fakeBackend struct { deleted int } +type fakeBootstrap struct{} + +func (fakeBootstrap) Environment(taskassignment.Assignment) (map[string]string, error) { + return map[string]string{"CI_RUNNER_CAPABILITY": "capability"}, nil +} + func (b *fakeBackend) Find(context.Context, string) (*Executor, error) { return b.executor, nil } -func (b *fakeBackend) Create(_ context.Context, _ taskassignment.Assignment, _ Metadata) (*Executor, error) { +func (b *fakeBackend) Create(_ context.Context, _ taskassignment.Assignment, _ LaunchSpec) (*Executor, error) { b.created++ b.executor = &Executor{Name: "executor", IdentityTarget: "pod-uid", Phase: PhaseRunning} return b.executor, nil @@ -60,7 +66,7 @@ func assignment() taskassignment.Assignment { func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) { backend := &fakeBackend{executor: &Executor{Name: "existing", IdentityTarget: "uid", Phase: PhaseRunning}} - worker := Worker{Backend: backend, Tasks: &fakeTasks{}} + worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}} done, err := worker.Handle(context.Background(), assignment()) if err != nil || done { @@ -73,7 +79,7 @@ func TestHandleRecoversExistingExecutorWithoutCreatingAnother(t *testing.T) { func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) { backend := &fakeBackend{} - worker := Worker{Backend: backend, Tasks: &fakeTasks{}} + worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}} accepted, err := worker.Accept(context.Background(), assignment()) if err != nil || !accepted { @@ -86,7 +92,7 @@ func TestAcceptAcknowledgesAfterBackendAndIdentityAreDurable(t *testing.T) { func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) { backend := &fakeBackend{executor: &Executor{Name: "pending", Phase: PhasePending}} - worker := Worker{Backend: backend, Tasks: &fakeTasks{}} + worker := Worker{Backend: backend, Tasks: &fakeTasks{}, Bootstrap: fakeBootstrap{}} accepted, err := worker.Accept(context.Background(), assignment()) if err != nil || accepted { @@ -100,7 +106,7 @@ func TestAcceptRetriesWhileBackendIdentityTargetIsUnavailable(t *testing.T) { func TestHandleReportsBeforeCleanupAndBecomesRecoverable(t *testing.T) { backend := &fakeBackend{executor: &Executor{Name: "finished", IdentityTarget: "uid", Phase: PhaseSucceeded}} tasks := &fakeTasks{} - worker := Worker{Backend: backend, Tasks: tasks} + worker := Worker{Backend: backend, Tasks: tasks, Bootstrap: fakeBootstrap{}} done, err := worker.Handle(context.Background(), assignment()) if err != nil || !done {