fix: preserve split NATS permissions

README.md

@@ -60,7 +60,8 @@ credential 都从挂载文件读取，不接受明文环境变量：
 
 - `GITEA_RUNNER_UUID_FILE`、`GITEA_RUNNER_TOKEN_FILE`：scheduler 的常驻 RunnerService
   registration；该 credential 不下发给 executor。
-- `NATS_PASSWORD_FILE`：assignment stream 的连接密码。
+- `NATS_PRODUCER_PASSWORD_FILE`、`NATS_WORKER_PASSWORD_FILE`：分别使用现有最小权限的
+  `ci-producer` publish 连接和 `ci-worker` pull/ACK 连接，controller 不合并权限。
 - `RUNNER_FACADE_CAPABILITY_KEY_FILE`：至少 32 字节的 controller HMAC key。
 - `OPENSANDBOX_API_KEY_FILE`：仅启用 `vm-worker` 时读取。
 

cmd/gitea-dynamic-runner/controller.go

@@ -36,7 +36,8 @@ type controllerConfig struct {
 	Components                                                          controller.Selection
 	TrustDomain, WorkloadAPIAddr                                        string
 	GiteaURL, GiteaUUID, GiteaToken                                     string
-	NATSURL, NATSUser, NATSPassword, NATSCA, Stream, SubjectBase        string
+	NATSURL, NATSProducerUser, NATSProducerPassword                     string
+	NATSWorkerUser, NATSWorkerPassword, NATSCA, Stream, SubjectBase     string
 	FacadeListen, FacadeURL, FacadeSPIFFEID                             string
 	CapabilityKey                                                       []byte
 	PodNamespace, PodImage, PodServiceAccount, SPIRECluster, SPIREClass string
@@ -57,25 +58,27 @@ func runController(ctx context.Context) error {
 		return errors.New("scheduler requires at least one local backend worker")
 	}
 
-	natsOptions := []nats.Option{nats.Name("gitea-dynamic-runner")}
+	producerConnection, err := connectNATS(config.NATSURL, config.NATSProducerUser, config.NATSProducerPassword, config.NATSCA, "gitea-dynamic-runner-producer")
-	if config.NATSUser != "" || config.NATSPassword != "" {
-		natsOptions = append(natsOptions, nats.UserInfo(config.NATSUser, config.NATSPassword))
-	}
-	if config.NATSCA != "" {
-		natsOptions = append(natsOptions, nats.RootCAs(config.NATSCA))
-	}
-	natsConnection, err := nats.Connect(config.NATSURL, natsOptions...)
 	if err != nil {
-		return fmt.Errorf("connect NATS: %w", err)
+		return fmt.Errorf("connect NATS producer: %w", err)
 	}
-	defer natsConnection.Close()
+	defer producerConnection.Close()
-	js, err := jetstream.New(natsConnection)
+	producerJS, err := jetstream.New(producerConnection)
 	if err != nil {
-		return fmt.Errorf("open JetStream: %w", err)
+		return fmt.Errorf("open producer JetStream: %w", err)
 	}
-	if _, err := js.Stream(ctx, config.Stream); err != nil {
+	if _, err := producerJS.Stream(ctx, config.Stream); err != nil {
 		return fmt.Errorf("open assignment stream %s: %w", config.Stream, err)
 	}
+	workerConnection, err := connectNATS(config.NATSURL, config.NATSWorkerUser, config.NATSWorkerPassword, config.NATSCA, "gitea-dynamic-runner-worker")
+	if err != nil {
+		return fmt.Errorf("connect NATS worker: %w", err)
+	}
+	defer workerConnection.Close()
+	workerJS, err := jetstream.New(workerConnection)
+	if err != nil {
+		return fmt.Errorf("open worker JetStream: %w", err)
+	}
 
 	capabilities, err := runnerfacade.NewCapabilities(config.CapabilityKey)
 	if err != nil {
@@ -98,7 +101,7 @@ func runController(ctx context.Context) error {
 	poller := taskscheduler.Poller{
 		Client: giteaClient,
 		Scheduler: &taskscheduler.Scheduler{TrustDomain: config.TrustDomain, Dispatcher: assignmentqueue.Publisher{
-			JetStream: js, SubjectBase: config.SubjectBase,
+			JetStream: producerJS, SubjectBase: config.SubjectBase,
 		}},
 		Config:  taskscheduler.PollerConfig{Version: "gitea-dynamic-runner/0.4", Labels: labels},
 		OnError: func(err error) { log.Printf("scheduler: %v", err) },
@@ -126,7 +129,7 @@ func runController(ctx context.Context) error {
 			ExecutorArgs: []string{"executor"}, TrustDomain: config.TrustDomain,
 			SPIRECluster: config.SPIRECluster, SPIREClass: config.SPIREClass, ExecutorUID: config.PodExecutorUID,
 		}}
-		component, err := workerComponent(ctx, js, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendPod, config.PodCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
 		if err != nil {
 			return err
 		}
@@ -139,7 +142,7 @@ func runController(ctx context.Context) error {
 			Entrypoint: []string{"/usr/local/bin/gitea-dynamic-runner", "executor"},
 			Env:        map[string]string{"SPIFFE_ENDPOINT_SOCKET": config.WorkloadAPIAddr},
 		}}
-		component, err := workerComponent(ctx, js, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
+		component, err := workerComponent(ctx, workerJS, config, taskassignment.BackendVM, config.VMCapacity, taskworker.Worker{Backend: backend, Bootstrap: bootstrap}, registry)
 		if err != nil {
 			return err
 		}
@@ -148,6 +151,14 @@ func runController(ctx context.Context) error {
 	return controller.Run(ctx, config.Components, components)
 }
 
+func connectNATS(server, user, password, caFile, clientName string) (*nats.Conn, error) {
+	options := []nats.Option{nats.Name(clientName), nats.UserInfo(user, password)}
+	if caFile != "" {
+		options = append(options, nats.RootCAs(caFile))
+	}
+	return nats.Connect(server, options...)
+}
+
 func workerComponent(ctx context.Context, js jetstream.JetStream, config controllerConfig, backend taskassignment.Backend, capacity int, accepter assignmentqueue.Accepter, claims assignmentqueue.Claims) (controller.Component, error) {
 	consumer, err := assignmentqueue.OpenConsumer(ctx, js, config.Stream, config.SubjectBase, backend, capacity)
 	if err != nil {
@@ -184,7 +195,11 @@ func loadControllerConfig() (controllerConfig, error) {
 	if err != nil {
 		return controllerConfig{}, err
 	}
-	natsPassword, err := read("NATS_PASSWORD_FILE")
+	producerPassword, err := read("NATS_PRODUCER_PASSWORD_FILE")
+	if err != nil {
+		return controllerConfig{}, err
+	}
+	workerPassword, err := read("NATS_WORKER_PASSWORD_FILE")
 	if err != nil {
 		return controllerConfig{}, err
 	}
@@ -195,7 +210,9 @@ func loadControllerConfig() (controllerConfig, error) {
 	config := controllerConfig{
 		Components: selection, TrustDomain: env("TRUST_DOMAIN", "ddupan.top"), WorkloadAPIAddr: os.Getenv("SPIFFE_ENDPOINT_SOCKET"),
 		GiteaURL: env("GITEA_INSTANCE_URL", "https://git.ddupan.top"), GiteaUUID: uuid, GiteaToken: token,
-		NATSURL: env("NATS_URL", "tls://nats.ad.ddupan.top:4222"), NATSUser: env("NATS_USER", "ci-worker"), NATSPassword: natsPassword,
+		NATSURL:          env("NATS_URL", "tls://nats.ad.ddupan.top:4222"),
+		NATSProducerUser: env("NATS_PRODUCER_USER", "ci-producer"), NATSProducerPassword: producerPassword,
+		NATSWorkerUser: env("NATS_WORKER_USER", "ci-worker"), NATSWorkerPassword: workerPassword,
 		NATSCA: os.Getenv("NATS_CA_FILE"), Stream: env("NATS_STREAM", "CI_RUNNER"), SubjectBase: env("NATS_SUBJECT_BASE", "ci.runner"),
 		FacadeListen: env("RUNNER_FACADE_LISTEN", ":8443"), FacadeURL: os.Getenv("RUNNER_FACADE_URL"), FacadeSPIFFEID: os.Getenv("RUNNER_FACADE_SPIFFE_ID"), CapabilityKey: []byte(capabilityKey),
 		PodNamespace: env("POD_NAMESPACE", "gitea-actions"), PodImage: os.Getenv("POD_EXECUTOR_IMAGE"), PodServiceAccount: env("POD_SERVICE_ACCOUNT", "gitea-task-executor"),

cmd/gitea-dynamic-runner/controller_test.go

@@ -21,7 +21,8 @@ func TestLoadControllerConfigUsesFileSecrets(t *testing.T) {
 	t.Setenv("COMPONENTS", "scheduler,pod-worker")
 	t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "scheduler-uuid"))
 	t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "scheduler-token"))
-	t.Setenv("NATS_PASSWORD_FILE", secretFile(t, "nats", "nats-password"))
+	t.Setenv("NATS_PRODUCER_PASSWORD_FILE", secretFile(t, "nats-producer", "producer-password"))
+	t.Setenv("NATS_WORKER_PASSWORD_FILE", secretFile(t, "nats-worker", "worker-password"))
 	t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef"))
 	t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock")
 	t.Setenv("RUNNER_FACADE_URL", "https://gitea-runner-facade.gitea-actions.svc:8443")
@@ -35,7 +36,7 @@ func TestLoadControllerConfigUsesFileSecrets(t *testing.T) {
 	if len(config.Components) != 2 || config.Components[0] != controller.Scheduler || config.Components[1] != controller.PodWorker {
 		t.Fatalf("components = %#v", config.Components)
 	}
-	if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSPassword != "nats-password" {
+	if config.GiteaUUID != "scheduler-uuid" || config.GiteaToken != "scheduler-token" || config.NATSProducerPassword != "producer-password" || config.NATSWorkerPassword != "worker-password" {
 		t.Fatal("file secrets were not loaded")
 	}
 	if string(config.CapabilityKey) != "0123456789abcdef0123456789abcdef" || config.PodExecutorUID != 2000 {
@@ -47,7 +48,8 @@ func TestLoadControllerConfigRequiresOpenSandboxSecretOnlyForVM(t *testing.T) {
 	t.Setenv("COMPONENTS", "scheduler,vm-worker")
 	t.Setenv("GITEA_RUNNER_UUID_FILE", secretFile(t, "uuid", "uuid"))
 	t.Setenv("GITEA_RUNNER_TOKEN_FILE", secretFile(t, "token", "token"))
-	t.Setenv("NATS_PASSWORD_FILE", secretFile(t, "nats", "password"))
+	t.Setenv("NATS_PRODUCER_PASSWORD_FILE", secretFile(t, "nats-producer", "producer-password"))
+	t.Setenv("NATS_WORKER_PASSWORD_FILE", secretFile(t, "nats-worker", "worker-password"))
 	t.Setenv("RUNNER_FACADE_CAPABILITY_KEY_FILE", secretFile(t, "capability", "0123456789abcdef0123456789abcdef"))
 	t.Setenv("SPIFFE_ENDPOINT_SOCKET", "unix:///run/spire/agent-sockets/spire-agent.sock")
 	t.Setenv("RUNNER_FACADE_URL", "https://facade:8443")