Controller: store, certificate authority and Join

.github/workflows/ci.yml

@@ -11,6 +11,17 @@ permissions:
 jobs:
   test:
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_PASSWORD: postgres
+        ports: ["5432:5432"]
+        options: >-
+          --health-cmd pg_isready --health-interval 5s --health-timeout 5s --health-retries 10
+    env:
+      OPIFICA_TEST_DATABASE_URL: postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable
+      OPIFICA_TEST_REQUIRE_DB: "1"
     steps:
       - uses: actions/checkout@v7
       - uses: actions/setup-go@v7
@@ -21,6 +32,6 @@ jobs:
       - name: Generated code is up to date
         run: |
           make generate
-          git diff --exit-code -- gen/ || (echo "Run 'make generate' and commit the result." && exit 1)
+          git diff --exit-code -- gen/ internal/controller/store/db/ || (echo "Run 'make generate' and commit the result." && exit 1)
       - run: make test
       - run: make build

Makefile

@@ -1,6 +1,7 @@
 BUF_VERSION            := v1.75.0
 PROTOC_GEN_GO_VERSION  := v1.36.12
 CONNECT_GO_VERSION     := v1.21.1
+SQLC_VERSION           := v1.31.1
 
 GOBIN   := $(shell go env GOPATH)/bin
 VERSION ?= $(shell git describe --tags --always --dirty 2>/dev/null || echo dev)
@@ -21,14 +22,16 @@ lint: ## gofmt, go vet and buf lint
 	go vet ./...
 	$(GOBIN)/buf lint
 
-generate: ## Regenerate Go code from proto/
+generate: ## Regenerate Go code from proto/ and the SQL queries
 	rm -rf gen
 	PATH="$(GOBIN):$$PATH" $(GOBIN)/buf generate
+	$(GOBIN)/sqlc generate
 
-tools: ## Install buf and the protoc plugins
+tools: ## Install buf, the protoc plugins and sqlc
 	go install github.com/bufbuild/buf/cmd/buf@$(BUF_VERSION)
 	go install google.golang.org/protobuf/cmd/protoc-gen-go@$(PROTOC_GEN_GO_VERSION)
 	go install connectrpc.com/connect/cmd/protoc-gen-connect-go@$(CONNECT_GO_VERSION)
+	go install github.com/sqlc-dev/sqlc/cmd/sqlc@$(SQLC_VERSION)
 
 clean:
 	rm -rf bin

cmd/controller/main.go

@@ -1,47 +1,254 @@
 // Command controller runs the Opifica controller: the API, Git provider
 // integration, pipeline engine, scheduler and secrets broker.
 //
-// This is a skeleton: it serves the worker API with every method returning
-// "unimplemented", so the wiring can be tested end to end.
+// Implemented so far: the database, the built-in certificate authority, join
+// tokens and Join. Every other RPC returns "unimplemented".
 package main
 
 import (
 	"context"
+	"crypto/tls"
+	"errors"
 	"flag"
+	"fmt"
+	"io"
 	"log/slog"
 	"net/http"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
+	"time"
 
+	"connectrpc.com/connect"
+
+	opificav1 "github.com/opifica/opifica/gen/opifica/v1"
 	"github.com/opifica/opifica/gen/opifica/v1/opificav1connect"
+	"github.com/opifica/opifica/internal/controller/ca"
+	"github.com/opifica/opifica/internal/controller/store"
+	"github.com/opifica/opifica/internal/controller/workers"
 	"github.com/opifica/opifica/internal/server"
 	"github.com/opifica/opifica/internal/version"
 )
 
+const usage = `Usage: controller <command> [options]
+
+Commands:
+  serve          Run the controller
+  token create   Create a one-time join token for a new build machine
+  version        Print the version
+
+Common options (or environment variables):
+  -database-url  PostgreSQL URL (OPIFICA_DATABASE_URL)
+  -data-dir      Directory for the CA key and certificate (OPIFICA_DATA_DIR, default ./data)
+`
+
 func main() {
-	addr := flag.String("addr", ":8080", "address to listen on")
-	flag.Parse()
-	log := slog.New(slog.NewTextHandler(os.Stderr, nil)).With("service", "controller")
+	os.Exit(run(os.Args[1:], os.Stdout, os.Stderr))
+}
+
+func run(args []string, stdout, stderr io.Writer) int {
+	if len(args) == 0 {
+		fmt.Fprint(stderr, usage)
+		return 2
+	}
+	switch {
+	case args[0] == "serve":
+		return serve(args[1:], stderr)
+	case args[0] == "token" && len(args) > 1 && args[1] == "create":
+		return tokenCreate(args[2:], stdout, stderr)
+	case args[0] == "version":
+		fmt.Fprintln(stdout, version.Version)
+		return 0
+	case args[0] == "-h" || args[0] == "--help" || args[0] == "help":
+		fmt.Fprint(stdout, usage)
+		return 0
+	}
+	fmt.Fprintf(stderr, "controller: unknown command %q\n\n%s", strings.Join(args, " "), usage)
+	return 2
+}
+
+type common struct {
+	databaseURL string
+	dataDir     string
+}
+
+func (c *common) register(fs *flag.FlagSet) {
+	fs.StringVar(&c.databaseURL, "database-url", os.Getenv("OPIFICA_DATABASE_URL"), "PostgreSQL URL")
+	def := os.Getenv("OPIFICA_DATA_DIR")
+	if def == "" {
+		def = "data"
+	}
+	fs.StringVar(&c.dataDir, "data-dir", def, "directory for the CA key and certificate")
+}
+
+func (c *common) open(ctx context.Context) (*workers.Service, func(), error) {
+	if c.databaseURL == "" {
+		return nil, nil, errors.New("-database-url or OPIFICA_DATABASE_URL is required")
+	}
+	st, err := store.Open(ctx, c.databaseURL)
+	if err != nil {
+		return nil, nil, err
+	}
+	authority, err := ca.LoadOrCreate(c.dataDir + "/ca")
+	if err != nil {
+		st.Close()
+		return nil, nil, fmt.Errorf("loading CA: %w", err)
+	}
+	return &workers.Service{Store: st, CA: authority}, st.Close, nil
+}
+
+func serve(args []string, stderr io.Writer) int {
+	fs := flag.NewFlagSet("serve", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	addr := fs.String("addr", ":8443", "address to listen on")
+	hosts := fs.String("hosts", defaultHosts(), "comma-separated names and IPs workers use to reach the controller")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	log := slog.New(slog.NewTextHandler(stderr, nil)).With("service", "controller")
 	log.Info("starting", "version", version.Version)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stop()
-	if err := server.Run(ctx, *addr, newHandler(log), log); err != nil {
+	svc, closeFn, err := c.open(ctx)
+	if err != nil {
+		log.Error("startup failed", "err", err)
+		return 1
+	}
+	defer closeFn()
+	svc.Log = log
+
+	tlsCfg, err := tlsConfig(svc.CA, splitList(*hosts))
+	if err != nil {
+		log.Error("issuing server certificate", "err", err)
+		return 1
+	}
+	log.Info("certificate authority ready", "fingerprint", svc.CA.Fingerprint())
+	if err := server.RunTLS(ctx, *addr, newHandler(svc), tlsCfg, log); err != nil {
 		log.Error("server stopped", "err", err)
-		os.Exit(1)
+		return 1
 	}
+	return 0
 }
 
-func newHandler(log *slog.Logger) http.Handler {
+// tlsConfig issues a fresh server certificate and accepts, but doesn't yet
+// require, worker client certificates: Join is called before a worker has
+// one. RPCs that need a worker identity check it themselves.
+func tlsConfig(authority *ca.CA, hosts []string) (*tls.Config, error) {
+	certPEM, keyPEM, err := authority.IssueServerCert(hosts)
+	if err != nil {
+		return nil, err
+	}
+	// Send the CA too, so a joining worker can check it against the pinned
+	// fingerprint before it has the CA bundle.
+	certPEM = append(certPEM, authority.CertPEM()...)
+	pair, err := tls.X509KeyPair(certPEM, keyPEM)
+	if err != nil {
+		return nil, err
+	}
+	return &tls.Config{
+		Certificates: []tls.Certificate{pair},
+		ClientCAs:    authority.Pool(),
+		ClientAuth:   tls.VerifyClientCertIfGiven,
+		MinVersion:   tls.VersionTLS13,
+	}, nil
+}
+
+func defaultHosts() string {
+	hosts := []string{"localhost", "127.0.0.1", "::1"}
+	if h, err := os.Hostname(); err == nil && h != "" {
+		hosts = append([]string{h}, hosts...)
+	}
+	return strings.Join(hosts, ",")
+}
+
+func splitList(s string) []string {
+	var out []string
+	for _, p := range strings.Split(s, ",") {
+		if p = strings.TrimSpace(p); p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+func newHandler(svc *workers.Service) http.Handler {
 	mux := http.NewServeMux()
 	mux.HandleFunc("GET /healthz", server.Health)
-	mux.Handle(opificav1connect.NewWorkerServiceHandler(&workerService{log: log}))
+	mux.Handle(opificav1connect.NewWorkerServiceHandler(&workerHandler{svc: svc}))
 	return mux
 }
 
-// workerService implements the API workers connect to.
-type workerService struct {
+// workerHandler implements the API workers connect to.
+type workerHandler struct {
 	opificav1connect.UnimplementedWorkerServiceHandler
-	log *slog.Logger
+	svc *workers.Service
+}
+
+func (h *workerHandler) Join(ctx context.Context, req *connect.Request[opificav1.JoinRequest]) (*connect.Response[opificav1.JoinResponse], error) {
+	resp, err := h.svc.Join(ctx, req.Msg)
+	if err != nil {
+		return nil, err
+	}
+	return connect.NewResponse(resp), nil
+}
+
+type labelFlag map[string]string
+
+func (l labelFlag) String() string { return "" }
+func (l labelFlag) Set(s string) error {
+	k, v, ok := strings.Cut(s, "=")
+	if !ok || k == "" {
+		return fmt.Errorf("label must be key=value, got %q", s)
+	}
+	l[k] = v
+	return nil
+}
+
+func tokenCreate(args []string, stdout, stderr io.Writer) int {
+	fs := flag.NewFlagSet("token create", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	var c common
+	c.register(fs)
+	pool := fs.String("pool", "", "pool the new worker joins (required), e.g. untrusted, trusted or release")
+	ttl := fs.Duration("ttl", workers.DefaultTokenTTL, "how long the token stays valid")
+	desc := fs.String("description", "", "note shown when listing tokens")
+	labels := labelFlag{}
+	fs.Var(labels, "label", "label for the worker, key=value (repeatable)")
+	if err := fs.Parse(args); err != nil {
+		return 2
+	}
+	if *pool == "" {
+		fmt.Fprintln(stderr, "controller: -pool is required")
+		return 2
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	svc, closeFn, err := c.open(ctx)
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	defer closeFn()
+
+	tok, err := svc.CreateToken(ctx, workers.TokenRequest{Pool: *pool, Labels: labels, Description: *desc, TTL: *ttl})
+	if err != nil {
+		fmt.Fprintf(stderr, "controller: %v\n", err)
+		return 1
+	}
+	fmt.Fprintf(stdout, `Join token for pool %q (valid until %s, single use):
+
+  %s
+
+CA fingerprint (the worker checks the controller against it):
+
+  %s
+
+The token is shown only once.
+`, *pool, tok.ExpiresAt.Format(time.RFC3339), tok.Secret, tok.CAFingerprint)
+	return 0
 }