gitea-dynamic-runner/internal/runnerfacade/facade.go

// Package runnerfacade presents pre-assigned tasks to unmodified Gitea Runner binaries.
package runnerfacade

import (
	"context"
	"errors"
	"net/http"

	"connectrpc.com/connect"
	"gitea.dev/actionslib/pkg/protocol"
	runnerv1 "gitea.dev/actionslib/runner/v1"
	"gitea.dev/actionslib/runner/v1/runnerv1connect"

	"git.ddupan.top/panxiao81/gitea-dynamic-runner/internal/taskassignment"
)

type identityKey struct{}

func WithSPIFFEID(ctx context.Context, id string) context.Context {
	return context.WithValue(ctx, identityKey{}, id)
}

// SPIFFEMiddleware extracts the authenticated workload identity from the mTLS
// peer certificate. TLS verification itself is configured by the server with
// go-spiffe; this layer only passes the verified ID into Connect handlers.
func SPIFFEMiddleware(next http.Handler) http.Handler {
	return http.HandlerFunc(func(response http.ResponseWriter, request *http.Request) {
		if request.TLS == nil || len(request.TLS.PeerCertificates) == 0 {
			http.Error(response, "client SPIFFE identity required", http.StatusUnauthorized)
			return
		}
		var spiffeID string
		for _, uri := range request.TLS.PeerCertificates[0].URIs {
			if uri.Scheme == "spiffe" {
				if spiffeID != "" {
					http.Error(response, "multiple client SPIFFE identities", http.StatusUnauthorized)
					return
				}
				spiffeID = uri.String()
			}
		}
		if spiffeID == "" {
			http.Error(response, "client SPIFFE identity required", http.StatusUnauthorized)
			return
		}
		next.ServeHTTP(response, request.WithContext(WithSPIFFEID(request.Context(), spiffeID)))
	})
}

type Upstream interface {
	UpdateTask(context.Context, *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error)
	UpdateLog(context.Context, *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error)
}

type Facade struct {
	runnerv1connect.UnimplementedRunnerServiceHandler
	Registry     *Registry
	Capabilities Capabilities
	Upstream     Upstream
	OnTerminal   func(context.Context, taskassignment.Assignment) error
}

func (f *Facade) Handler() (string, http.Handler) {
	return runnerv1connect.NewRunnerServiceHandler(f)
}

func (f *Facade) Register(context.Context, *connect.Request[runnerv1.RegisterRequest]) (*connect.Response[runnerv1.RegisterResponse], error) {
	return nil, connect.NewError(connect.CodePermissionDenied, errors.New("executor registration is disabled"))
}

func (f *Facade) Declare(ctx context.Context, request *connect.Request[runnerv1.DeclareRequest]) (*connect.Response[runnerv1.DeclareResponse], error) {
	assignmentID, _, err := f.authenticate(ctx, request)
	if err != nil {
		return nil, err
	}
	return connect.NewResponse(&runnerv1.DeclareResponse{Runner: &runnerv1.Runner{
		Uuid: assignmentID, Name: assignmentID, Status: runnerv1.RunnerStatus_RUNNER_STATUS_IDLE,
		Version: request.Msg.GetVersion(), Labels: append([]string(nil), request.Msg.GetLabels()...), Ephemeral: true,
	}}), nil
}

func (f *Facade) FetchTask(ctx context.Context, request *connect.Request[runnerv1.FetchTaskRequest]) (*connect.Response[runnerv1.FetchTaskResponse], error) {
	assignmentID, spiffeID, err := f.authenticate(ctx, request)
	if err != nil {
		return nil, err
	}
	assignment, err := f.Registry.Claim(assignmentID, spiffeID)
	if err != nil {
		return nil, connect.NewError(connect.CodeFailedPrecondition, err)
	}
	return connect.NewResponse(&runnerv1.FetchTaskResponse{Task: assignment.Task}), nil
}

func (f *Facade) UpdateTask(ctx context.Context, request *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) {
	assignment, err := f.authorizeClaimed(ctx, request)
	if err != nil {
		return nil, err
	}
	if request.Msg.GetState().GetId() != assignment.Task.GetId() {
		return nil, connect.NewError(connect.CodePermissionDenied, errors.New("task update does not match assignment"))
	}
	response, err := f.Upstream.UpdateTask(ctx, connect.NewRequest(request.Msg))
	if err == nil && request.Msg.GetState().GetResult() != runnerv1.Result_RESULT_UNSPECIFIED && f.OnTerminal != nil {
		if terminalErr := f.OnTerminal(ctx, assignment); terminalErr != nil {
			return nil, connect.NewError(connect.CodeUnavailable, terminalErr)
		}
	}
	return response, err
}

func (f *Facade) UpdateLog(ctx context.Context, request *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) {
	assignment, err := f.authorizeClaimed(ctx, request)
	if err != nil {
		return nil, err
	}
	if request.Msg.GetTaskId() != assignment.Task.GetId() {
		return nil, connect.NewError(connect.CodePermissionDenied, errors.New("log update does not match assignment"))
	}
	return f.Upstream.UpdateLog(ctx, connect.NewRequest(request.Msg))
}

func (f *Facade) authorizeClaimed(ctx context.Context, request connect.AnyRequest) (taskassignment.Assignment, error) {
	assignmentID, spiffeID, err := f.authenticate(ctx, request)
	if err != nil {
		return taskassignment.Assignment{}, err
	}
	assignment, err := f.Registry.Resolve(assignmentID, spiffeID)
	if err != nil {
		return taskassignment.Assignment{}, connect.NewError(connect.CodePermissionDenied, err)
	}
	return assignment, nil
}

func (f *Facade) authenticate(ctx context.Context, request connect.AnyRequest) (string, string, error) {
	if f.Registry == nil || f.Upstream == nil {
		return "", "", connect.NewError(connect.CodeInternal, errors.New("runner facade is not configured"))
	}
	assignmentID := request.Header().Get(protocol.UUIDHeader)
	token := request.Header().Get(protocol.TokenHeader)
	spiffeID, _ := ctx.Value(identityKey{}).(string)
	if assignmentID == "" || spiffeID == "" || !f.Capabilities.Verify(assignmentID, token) {
		return "", "", connect.NewError(connect.CodeUnauthenticated, errors.New("invalid executor identity or capability"))
	}
	return assignmentID, spiffeID, nil
}