diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index 785a120..cb40e4d 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -6,6 +6,16 @@ on: pull_request: jobs: + go: + runs-on: [self-hosted, pod] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-go@v6 + with: + go-version-file: go.mod + - run: go test ./... + - run: go vet ./... + python: runs-on: self-hosted steps: diff --git a/docs/runner-protocol-roadmap.md b/docs/runner-protocol-roadmap.md index 17d16f5..a185c79 100644 --- a/docs/runner-protocol-roadmap.md +++ b/docs/runner-protocol-roadmap.md @@ -29,6 +29,8 @@ dynamic-runner scheduler - 每个 executor 只执行一个 task,完成后销毁。 - SPIFFE 身份从实际领取的 task 的 repository 和 job name 派生,不需要 queued 与 in-progress webhook 的二阶段关联。 +- scheduler 只做确定性的身份派生与 executor 绑定,不维护业务授权 policy;Zot、 + OpenBao 等资源服务继续是唯一授权决策点。 - scheduler 的 runner registration credential 不进入 executor;executor 只得到执行 当前 task 所需的短期 lease/capability。 - task ACK、心跳和结果必须能够跨 scheduler 重启恢复;NATS 可以继续作为内部 handoff, diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..065521b --- /dev/null +++ b/go.mod @@ -0,0 +1,9 @@ +module git.ddupan.top/panxiao81/gitea-dynamic-runner + +go 1.27 + +require ( + connectrpc.com/connect v1.20.0 + gitea.dev/actionslib v1.0.0 + google.golang.org/protobuf v1.36.12 +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..f6c544f --- /dev/null +++ b/go.sum @@ -0,0 +1,8 @@ +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/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= +google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/internal/giteaactions/client.go b/internal/giteaactions/client.go new file mode 100644 index 0000000..299bdae --- /dev/null +++ b/internal/giteaactions/client.go @@ -0,0 +1,61 @@ +// Package giteaactions provides the authenticated Gitea RunnerService client. +package giteaactions + +import ( + "context" + "net/http" + "strings" + + "connectrpc.com/connect" + "gitea.dev/actionslib/pkg/protocol" + runnerv1 "gitea.dev/actionslib/runner/v1" + "gitea.dev/actionslib/runner/v1/runnerv1connect" +) + +// Client is the subset of RunnerService owned by the scheduler. +type Client struct { + runner runnerv1connect.RunnerServiceClient +} + +// NewClient authenticates every RPC with the persistent scheduler runner. +func NewClient(httpClient connect.HTTPClient, instanceURL, uuid, token string) *Client { + auth := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, request connect.AnyRequest) (connect.AnyResponse, error) { + request.Header().Set("User-Agent", "gitea-dynamic-runner-go/0") + request.Header().Set(protocol.UUIDHeader, uuid) + request.Header().Set(protocol.TokenHeader, token) + return next(ctx, request) + } + }) + baseURL := strings.TrimRight(instanceURL, "/") + "/api/actions" + return &Client{runner: runnerv1connect.NewRunnerServiceClient( + httpClient, + baseURL, + connect.WithInterceptors(auth), + )} +} + +// Declare advertises the scheduler labels before tasks are fetched. +func (c *Client) Declare(ctx context.Context, version string, labels []string) error { + _, err := c.runner.Declare(ctx, connect.NewRequest(&runnerv1.DeclareRequest{ + Version: version, + Labels: labels, + })) + return err +} + +// FetchTask asks Gitea to atomically assign the next matching task. +func (c *Client) FetchTask(ctx context.Context, tasksVersion int64) (*runnerv1.FetchTaskResponse, error) { + response, err := c.runner.FetchTask(ctx, connect.NewRequest(&runnerv1.FetchTaskRequest{ + TasksVersion: tasksVersion, + })) + if err != nil { + return nil, err + } + return response.Msg, nil +} + +// DefaultHTTPClient is suitable for the scheduler's long-lived connection. +func DefaultHTTPClient() *http.Client { + return &http.Client{Transport: http.DefaultTransport} +} diff --git a/internal/giteaactions/client_test.go b/internal/giteaactions/client_test.go new file mode 100644 index 0000000..db6a9fa --- /dev/null +++ b/internal/giteaactions/client_test.go @@ -0,0 +1,78 @@ +package giteaactions + +import ( + "context" + "net/http" + "net/http/httptest" + "reflect" + "testing" + + "connectrpc.com/connect" + "gitea.dev/actionslib/pkg/protocol" + runnerv1 "gitea.dev/actionslib/runner/v1" + "gitea.dev/actionslib/runner/v1/runnerv1connect" +) + +type runnerService struct { + runnerv1connect.UnimplementedRunnerServiceHandler + t *testing.T + declaredLabels []string + fetchedVersion int64 + expectedUUID string + expectedToken string +} + +func (s *runnerService) checkAuth(request connect.AnyRequest) { + s.t.Helper() + if got := request.Header().Get(protocol.UUIDHeader); got != s.expectedUUID { + s.t.Errorf("runner UUID header = %q", got) + } + if got := request.Header().Get(protocol.TokenHeader); got != s.expectedToken { + s.t.Errorf("runner token header = %q", got) + } +} + +func (s *runnerService) Declare(_ context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) { + s.checkAuth(request) + s.declaredLabels = request.Msg.Labels + return connect.NewResponse(&runnerv1.DeclareResponse{}), nil +} + +func (s *runnerService) FetchTask(_ context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) { + s.checkAuth(request) + s.fetchedVersion = request.Msg.TasksVersion + return connect.NewResponse(&runnerv1.FetchTaskResponse{ + Task: &runnerv1.Task{Id: 42}, + TasksVersion: 8, + }), nil +} + +func TestClientUsesOfficialRunnerProtocol(t *testing.T) { + service := &runnerService{ + t: t, + expectedUUID: "runner-uuid", + expectedToken: "runner-token", + } + path, handler := runnerv1connect.NewRunnerServiceHandler(service) + mux := http.NewServeMux() + mux.Handle("/api/actions"+path, http.StripPrefix("/api/actions", handler)) + server := httptest.NewServer(mux) + defer server.Close() + + client := NewClient(server.Client(), server.URL, service.expectedUUID, service.expectedToken) + labels := []string{"self-hosted", "pod", "vm"} + if err := client.Declare(context.Background(), "0.1.0", labels); err != nil { + t.Fatal(err) + } + response, err := client.FetchTask(context.Background(), 7) + if err != nil { + t.Fatal(err) + } + + if !reflect.DeepEqual(service.declaredLabels, labels) { + t.Fatalf("declared labels = %#v", service.declaredLabels) + } + if service.fetchedVersion != 7 || response.GetTasksVersion() != 8 || response.GetTask().GetId() != 42 { + t.Fatalf("unexpected FetchTask exchange: request=%d response=%v", service.fetchedVersion, response) + } +} diff --git a/internal/taskidentity/identity.go b/internal/taskidentity/identity.go new file mode 100644 index 0000000..747633f --- /dev/null +++ b/internal/taskidentity/identity.go @@ -0,0 +1,71 @@ +// Package taskidentity derives workload identities from tasks assigned by Gitea. +package taskidentity + +import ( + "crypto/sha256" + "errors" + "fmt" + "regexp" + "strings" + + runnerv1 "gitea.dev/actionslib/runner/v1" +) + +var safeSegment = regexp.MustCompile(`^[A-Za-z0-9._-]+$`) + +// Identity is the trusted identity context extracted from a fetched task. +type Identity struct { + Repository string + Task string + SPIFFEID string +} + +// FromTask derives the repository/task SPIFFE ID from Gitea's trusted task +// context. Workflow input never supplies or overrides the resulting ID. +func FromTask(task *runnerv1.Task, trustDomain string) (Identity, error) { + if task == nil || task.Context == nil { + return Identity{}, errors.New("task context is required") + } + + repository := strings.TrimSpace(task.Context.GetFields()["repository"].GetStringValue()) + taskName := strings.TrimSpace(task.Context.GetFields()["job"].GetStringValue()) + parts := strings.Split(repository, "/") + if len(parts) != 2 || parts[0] == "" || parts[1] == "" { + return Identity{}, fmt.Errorf("invalid task repository %q", repository) + } + if taskName == "" { + return Identity{}, errors.New("task job is required") + } + trustDomain = strings.TrimSpace(trustDomain) + if trustDomain == "" || strings.ContainsAny(trustDomain, "/ ") { + return Identity{}, fmt.Errorf("invalid trust domain %q", trustDomain) + } + + path := strings.Join([]string{ + sanitize(parts[0]), + sanitize(parts[1]), + sanitize(taskName), + }, "/") + return Identity{ + Repository: repository, + Task: taskName, + SPIFFEID: "spiffe://" + trustDomain + "/ci/" + path, + }, nil +} + +// sanitize intentionally matches the bootstrap Python implementation so +// existing Zot and OpenBao policies keep their current identity names. +func sanitize(value string) string { + if safeSegment.MatchString(value) { + return value + } + slug := strings.Trim(regexp.MustCompile(`[^A-Za-z0-9._-]+`).ReplaceAllString(value, "-"), "-._") + if len(slug) > 48 { + slug = strings.TrimRight(slug[:48], "-._") + } + if slug == "" { + slug = "segment" + } + digest := fmt.Sprintf("%x", sha256.Sum256([]byte(value)))[:12] + return slug + "-" + digest +} diff --git a/internal/taskidentity/identity_test.go b/internal/taskidentity/identity_test.go new file mode 100644 index 0000000..c15c226 --- /dev/null +++ b/internal/taskidentity/identity_test.go @@ -0,0 +1,64 @@ +package taskidentity + +import ( + "testing" + + runnerv1 "gitea.dev/actionslib/runner/v1" + "google.golang.org/protobuf/types/known/structpb" +) + +func TestFromTaskUsesFetchedContext(t *testing.T) { + ctx, err := structpb.NewStruct(map[string]any{ + "repository": "panxiao81/gitea-dynamic-runner", + "job": "publish-images", + }) + if err != nil { + t.Fatal(err) + } + + got, err := FromTask(&runnerv1.Task{Id: 900, Context: ctx}, "ddupan.top") + if err != nil { + t.Fatal(err) + } + if got.Repository != "panxiao81/gitea-dynamic-runner" || got.Task != "publish-images" { + t.Fatalf("unexpected task identity context: %#v", got) + } + want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/publish-images" + if got.SPIFFEID != want { + t.Fatalf("SPIFFE ID = %q, want %q", got.SPIFFEID, want) + } +} + +func TestFromTaskMatchesBootstrapNormalization(t *testing.T) { + ctx, err := structpb.NewStruct(map[string]any{ + "repository": "panxiao81/postgresql-tenant-operator", + "job": "Run on Ubuntu", + }) + if err != nil { + t.Fatal(err) + } + + got, err := FromTask(&runnerv1.Task{Context: ctx}, "ddupan.top") + if err != nil { + t.Fatal(err) + } + want := "spiffe://ddupan.top/ci/panxiao81/postgresql-tenant-operator/Run-on-Ubuntu-8b7cd4c244fb" + if got.SPIFFEID != want { + t.Fatalf("SPIFFE ID = %q, want %q", got.SPIFFEID, want) + } +} + +func TestFromTaskRejectsIncompleteServerContext(t *testing.T) { + for _, fields := range []map[string]any{ + {"repository": "invalid", "job": "test"}, + {"repository": "owner/repo", "job": ""}, + } { + ctx, err := structpb.NewStruct(fields) + if err != nil { + t.Fatal(err) + } + if _, err := FromTask(&runnerv1.Task{Context: ctx}, "ddupan.top"); err == nil { + t.Fatalf("expected invalid task context to fail: %#v", fields) + } + } +} diff --git a/internal/taskscheduler/scheduler.go b/internal/taskscheduler/scheduler.go new file mode 100644 index 0000000..c879400 --- /dev/null +++ b/internal/taskscheduler/scheduler.go @@ -0,0 +1,42 @@ +// Package taskscheduler owns tasks fetched through Gitea's RunnerService and +// dispatches them to an executor only after their identity is known. +package taskscheduler + +import ( + "context" + "errors" + + runnerv1 "gitea.dev/actionslib/runner/v1" + + "git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskidentity" +) + +// Assignment is the immutable input handed to a Pod or VM provisioner. +type Assignment struct { + Task *runnerv1.Task + Identity taskidentity.Identity +} + +// Dispatcher creates exactly one executor for an already assigned Gitea task. +// It must not register another runner or ask Gitea for a task. +type Dispatcher interface { + Dispatch(context.Context, Assignment) error +} + +// Scheduler implements the TaskRunner boundary used by Gitea Runner's poller. +type Scheduler struct { + TrustDomain string + Dispatcher Dispatcher +} + +// Run derives identity from the fetched task before provisioning its executor. +func (s *Scheduler) Run(ctx context.Context, task *runnerv1.Task) error { + if s.Dispatcher == nil { + return errors.New("executor dispatcher is required") + } + identity, err := taskidentity.FromTask(task, s.TrustDomain) + if err != nil { + return err + } + return s.Dispatcher.Dispatch(ctx, Assignment{Task: task, Identity: identity}) +} diff --git a/internal/taskscheduler/scheduler_test.go b/internal/taskscheduler/scheduler_test.go new file mode 100644 index 0000000..6dd1f62 --- /dev/null +++ b/internal/taskscheduler/scheduler_test.go @@ -0,0 +1,49 @@ +package taskscheduler + +import ( + "context" + "testing" + + runnerv1 "gitea.dev/actionslib/runner/v1" + "google.golang.org/protobuf/types/known/structpb" +) + +type recordingDispatcher struct { + assignment Assignment +} + +func (d *recordingDispatcher) Dispatch(_ context.Context, assignment Assignment) error { + d.assignment = assignment + return nil +} + +func TestRunDerivesIdentityBeforeDispatch(t *testing.T) { + taskContext, err := structpb.NewStruct(map[string]any{ + "repository": "panxiao81/gitea-dynamic-runner", + "job": "test", + }) + if err != nil { + t.Fatal(err) + } + task := &runnerv1.Task{Id: 42, Context: taskContext} + dispatcher := &recordingDispatcher{} + scheduler := Scheduler{TrustDomain: "ddupan.top", Dispatcher: dispatcher} + + if err := scheduler.Run(context.Background(), task); err != nil { + t.Fatal(err) + } + if dispatcher.assignment.Task != task { + t.Fatal("dispatcher did not receive the fetched task") + } + want := "spiffe://ddupan.top/ci/panxiao81/gitea-dynamic-runner/test" + if dispatcher.assignment.Identity.SPIFFEID != want { + t.Fatalf("SPIFFE ID = %q, want %q", dispatcher.assignment.Identity.SPIFFEID, want) + } +} + +func TestRunRequiresDispatcher(t *testing.T) { + scheduler := Scheduler{TrustDomain: "ddupan.top"} + if err := scheduler.Run(context.Background(), &runnerv1.Task{}); err == nil { + t.Fatal("expected missing dispatcher to fail") + } +}