From fe01fc94fb4ddfe239d6271069d8bded35de5fed Mon Sep 17 00:00:00 2001 From: "K.Milchev" <49519924+KristiforMilchev@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:51:14 +0200 Subject: [PATCH] Controller: store, certificate authority and Join - internal/controller/store: PostgreSQL with embedded migrations (applied under an advisory lock) and sqlc queries for workers and join tokens - internal/controller/ca: built-in CA for worker client certificates and the controller's server certificate - internal/controller/workers: single-use join tokens (only a hash is stored) and Join, which consumes the token, records the worker and signs its CSR in one transaction; the pool comes from the token, never the worker - pkg/worker: PinnedCAConfig, so a joining worker verifies the controller against the CA fingerprint printed with the token - controller serve (TLS) and controller token create - pgtest: per-test databases; CI runs them against PostgreSQL and fails instead of skipping when no database is available - docs/ROADMAP.md Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 13 +- Makefile | 7 +- cmd/controller/main.go | 231 +++++++++++++- cmd/controller/main_test.go | 132 +++++++- docs/ROADMAP.md | 109 +++++++ go.mod | 8 +- go.sum | 30 +- internal/controller/ca/ca.go | 299 ++++++++++++++++++ internal/controller/ca/ca_test.go | 235 ++++++++++++++ internal/controller/store/db/db.go | 32 ++ internal/controller/store/db/models.go | 40 +++ internal/controller/store/db/workers.sql.go | 277 ++++++++++++++++ .../store/migrations/0001_workers.sql | 37 +++ internal/controller/store/queries/workers.sql | 37 +++ internal/controller/store/store.go | 156 +++++++++ internal/controller/store/store_test.go | 140 ++++++++ internal/controller/workers/join.go | 271 ++++++++++++++++ internal/controller/workers/join_test.go | 239 ++++++++++++++ internal/server/server.go | 42 ++- internal/testutil/pgtest/pgtest.go | 170 ++++++++++ pkg/worker/tls.go | 64 ++++ pkg/worker/tls_test.go | 97 ++++++ sqlc.yaml | 29 ++ 23 files changed, 2657 insertions(+), 38 deletions(-) create mode 100644 docs/ROADMAP.md create mode 100644 internal/controller/ca/ca.go create mode 100644 internal/controller/ca/ca_test.go create mode 100644 internal/controller/store/db/db.go create mode 100644 internal/controller/store/db/models.go create mode 100644 internal/controller/store/db/workers.sql.go create mode 100644 internal/controller/store/migrations/0001_workers.sql create mode 100644 internal/controller/store/queries/workers.sql create mode 100644 internal/controller/store/store.go create mode 100644 internal/controller/store/store_test.go create mode 100644 internal/controller/workers/join.go create mode 100644 internal/controller/workers/join_test.go create mode 100644 internal/testutil/pgtest/pgtest.go create mode 100644 pkg/worker/tls.go create mode 100644 pkg/worker/tls_test.go create mode 100644 sqlc.yaml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ef2c542..5bd4cd8 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/Makefile b/Makefile index 4857fcf..748b556 100644 --- a/Makefile +++ b/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 diff --git a/cmd/controller/main.go b/cmd/controller/main.go index 2704998..b11d822 100644 --- a/cmd/controller/main.go +++ b/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 [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 } diff --git a/cmd/controller/main_test.go b/cmd/controller/main_test.go index 502643f..9399a12 100644 --- a/cmd/controller/main_test.go +++ b/cmd/controller/main_test.go @@ -1,37 +1,141 @@ package main import ( + "bytes" "context" - "errors" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "encoding/pem" "io" "log/slog" + "net" "net/http" - "net/http/httptest" + "os" + "path/filepath" + "regexp" "testing" + "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/server" + "github.com/opifica/opifica/internal/testutil/pgtest" + "github.com/opifica/opifica/pkg/worker" ) -func TestHandler(t *testing.T) { - srv := httptest.NewServer(newHandler(slog.New(slog.NewTextHandler(io.Discard, nil)))) - defer srv.Close() +func TestMain(m *testing.M) { os.Exit(pgtest.Run(m)) } - resp, err := http.Get(srv.URL + "/healthz") +// End to end through the binary's own code paths: create a token with +// "token create", serve over TLS, and join as a worker that pins the CA. +func TestTokenCreateAndJoinOverTLS(t *testing.T) { + dbURL := pgtest.New(t) + dataDir := t.TempDir() + + var out, errOut bytes.Buffer + code := run([]string{"token", "create", "-database-url", dbURL, "-data-dir", dataDir, + "-pool", "trusted", "-label", "region=eu"}, &out, &errOut) + if code != 0 { + t.Fatalf("token create exited %d: %s", code, errOut.String()) + } + token := regexp.MustCompile(`opj_\S+`).FindString(out.String()) + fingerprint := regexp.MustCompile(`sha256:[0-9a-f]{64}`).FindString(out.String()) + if token == "" || fingerprint == "" { + t.Fatalf("token create output lacks token or fingerprint:\n%s", out.String()) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c := common{databaseURL: dbURL, dataDir: dataDir} + svc, closeFn, err := c.open(ctx) + if err != nil { + t.Fatal(err) + } + defer closeFn() + svc.Log = slog.New(slog.NewTextHandler(io.Discard, nil)) + tlsCfg, err := tlsConfig(svc.CA, []string{"127.0.0.1"}) if err != nil { t.Fatal(err) } - resp.Body.Close() - if resp.StatusCode != http.StatusOK { - t.Errorf("healthz status = %d", resp.StatusCode) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) } + go server.ServeTLS(ctx, ln, newHandler(svc), tlsCfg, svc.Log) - client := opificav1connect.NewWorkerServiceClient(srv.Client(), srv.URL) - _, err = client.Join(context.Background(), connect.NewRequest(&opificav1.JoinRequest{JoinToken: "x"})) - var cerr *connect.Error - if !errors.As(err, &cerr) || cerr.Code() != connect.CodeUnimplemented { - t.Errorf("Join: err = %v, want unimplemented", err) + // The worker has no CA yet: it accepts the controller only if the + // certificate chains to a CA with the pinned fingerprint. + pinned, err := worker.PinnedCAConfig(fingerprint, "127.0.0.1") + if err != nil { + t.Fatal(err) + } + client := &http.Client{Transport: &http.Transport{ForceAttemptHTTP2: true, TLSClientConfig: pinned}} + base := "https://" + ln.Addr().String() + waitHealthy(t, client, base) + + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + der, _ := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{}, key) + wc := opificav1connect.NewWorkerServiceClient(client, base) + resp, err := wc.Join(ctx, connect.NewRequest(&opificav1.JoinRequest{ + JoinToken: token, + CsrPem: pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: der}), + Info: &opificav1.WorkerInfo{ + Hostname: "build-1", Os: opificav1.OperatingSystem_OPERATING_SYSTEM_LINUX, + Arch: opificav1.Architecture_ARCHITECTURE_AMD64, MaxConcurrentJobs: 4, + }, + })) + if err != nil { + t.Fatalf("Join: %v", err) + } + if resp.Msg.WorkerId == "" || len(resp.Msg.CertificatePem) == 0 { + t.Errorf("Join response = %+v", resp.Msg) + } + + // Still unimplemented: the stream comes next. + stream := wc.Connect(ctx) + stream.Send(&opificav1.ConnectRequest{}) + if _, err := stream.Receive(); connect.CodeOf(err) != connect.CodeUnimplemented { + t.Errorf("Connect: err = %v, want unimplemented", err) + } + stream.CloseRequest() + stream.CloseResponse() + + // The CA survived on disk: a second start must not create a new one. + if _, err := os.Stat(filepath.Join(dataDir, "ca", "ca.key")); err != nil { + t.Errorf("CA key not persisted: %v", err) + } +} + +func waitHealthy(t *testing.T, c *http.Client, base string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + resp, err := c.Get(base + "/healthz") + if err == nil { + resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return + } + } + if time.Now().After(deadline) { + t.Fatalf("controller not healthy: %v", err) + } + time.Sleep(20 * time.Millisecond) + } +} + +func TestUsage(t *testing.T) { + var out, errOut bytes.Buffer + if code := run(nil, &out, &errOut); code != 2 { + t.Errorf("no args: exit %d", code) + } + if code := run([]string{"bogus"}, &out, &errOut); code != 2 { + t.Errorf("unknown command: exit %d", code) + } + if code := run([]string{"token", "create", "-database-url", "x"}, &out, &errOut); code != 2 { + t.Errorf("token create without -pool: exit %d", code) } } diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md new file mode 100644 index 0000000..2c63900 --- /dev/null +++ b/docs/ROADMAP.md @@ -0,0 +1,109 @@ +# Roadmap + +Where Opifica is and what comes next. Each milestone ends in something that can be run and tested. +Tick items off as they land. + +**Defaults chosen for now** (can be revisited): +first Git provider is **Forgejo/Gitea**, **Linux** VMs come before Windows and macOS, and logs are stored on the +**controller's local disk** until S3 storage is added. + +## M0 · Foundations ✅ + +- [x] Organisation, repositories, EUPL-1.2 licence, architecture page +- [x] Pipeline language: parser, validation, CEL conditions, matrix, planning (`internal/pipeline`) +- [x] JSON Schema for editors, tested against the parser (`schema/`) +- [x] API definitions: worker ↔ controller, worker ↔ agent (`proto/`) +- [x] Worker kit: `Hypervisor` interface, warm VM pool, image verification, fake hypervisor (`pkg/worker`) +- [x] CLI: `validate`, `plan`, `schema` +- [x] CI for the core repository + +## M1 · First job end to end, without VMs + +**Goal:** a run created on the controller is scheduled to a worker, executed by a real agent, and its logs and +result are stored. Tested in ordinary CI, using the fake hypervisor with a real agent process behind it. + +### Controller + +- [x] **Store:** PostgreSQL migrations and sqlc queries, for workers and join tokens so far +- [ ] **Store:** runs and jobs +- [x] **Certificate authority:** built-in CA that issues worker client certificates and the server certificate +- [x] **Join:** admins create one-time join tokens (`controller token create`); `Join` checks the token and signs + the worker's CSR; first contact is verified against the CA fingerprint printed with the token +- [ ] **Connect stream:** Hello, heartbeats, worker registry, disconnect and reconnect handling +- [ ] **Leases:** job assignment with a lease renewed by heartbeats; expired leases put the job back in the queue +- [ ] **Scheduler:** match planned jobs to workers by OS, architecture, image, labels and pool; respect `needs` +- [ ] **Runs:** create a run from a pipeline file and an event; track job and run state +- [ ] **Logs:** receive log chunks, drop duplicates after reconnects, store them on disk +- [ ] **Secrets:** encrypted store; requests checked against the job token and policy +- [ ] **Admin config:** static file for pools, routing (event → pool) and secret policy +- [x] **TLS** on the worker API, with worker client certificates accepted +- [ ] **mTLS required** for `Connect`, and certificate renewal +- [ ] **Admin API + CLI:** `opifica worker token create`, `opifica runs list`, `opifica logs ` + +### Worker kit and agent + +- [ ] Controller client: join, connect, reconnect with backoff, heartbeats from pool stats +- [ ] Job runner: lease a VM → checkout → run steps through the agent → stream masked logs → collect reports + and artifacts → destroy the VM +- [ ] Local Git mirror per repository +- [ ] Agent: `Ready`, `Checkout`, `Exec`, `Collect` + +### Testing + +- [ ] Integration test: controller + PostgreSQL + worker (fake hypervisor) + real agent, in CI +- [ ] Failure tests: worker killed mid-job, lease expiry, reconnect, denied secret + +## M2 · Triggered by Git + +**Goal:** a push to a Forgejo repository starts a run, and the commit shows its status. + +- [ ] Provider interface and Forgejo/Gitea provider: webhooks with signature checks, `FetchFile` at a commit, + short-lived clone credentials, commit status +- [ ] Routing events to pools from the admin config (pull requests → `untrusted`, `main` → `trusted`, + tags → `release`) +- [ ] Integration test against a Forgejo container +- [ ] `docker compose` dev environment: PostgreSQL, Forgejo, controller, worker + +## M3 · Real Linux VMs + +**Goal:** every job runs in a fresh Firecracker microVM. + +- [ ] `worker-linux`: Firecracker driver with copy-on-write disks, vsock to the agent, jailer +- [ ] VM networking with egress allowlists (nftables) +- [ ] `images`: Linux golden image (kernel + root filesystem with the agent), built reproducibly +- [ ] E2E test on a machine with `/dev/kvm` +- [ ] Opifica starts building itself (dogfooding) + +## M4 · Releases and project automation + +Can run alongside M2 and M3. + +- [ ] Versioned releases of core (`v0.x` tags) so worker repositories can depend on it safely +- [ ] Release pipeline: cross-compiled binaries, cosign signatures, SBOM, provenance +- [ ] CI for every repository; branch protection on `main` +- [ ] Automated dependency updates + +## M5 · Supply-chain security + +- [ ] Signing service: release artifacts signed with keys in an HSM or KMS; keys never reach a VM +- [ ] Signed provenance for every job (SLSA) +- [ ] Optional rebuild-and-compare for releases on two different workers +- [ ] Cache rules: untrusted jobs can never write caches that trusted or release jobs read + +## M6 · Windows and macOS + +Needs self-hosted hardware: an Apple silicon Mac and a Windows machine with Hyper-V. + +- [ ] `worker-windows`: Hyper-V driver, differencing disks, warm pool +- [ ] `worker-macos`: Virtualization.framework driver, at most 2 VMs per Mac +- [ ] Image recipes for Windows (Packer) and macOS +- [ ] E2E tests on self-hosted machines + +## M7 · Everything else + +- [ ] Web UI (Angular): runs, live logs, test results, workers, administration +- [ ] GitHub and GitLab providers; plain Git with polling +- [ ] GitOps sync of the admin config from a Git repository +- [ ] S3-compatible storage for logs and artifacts +- [ ] OIDC login; OpenBao/Vault as a secrets backend +- [ ] OIDC tokens for jobs (short-lived cloud credentials instead of stored keys) diff --git a/go.mod b/go.mod index e73c730..b6c1407 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,8 @@ go 1.27.0 require ( cel.dev/cel-go v0.33.2 connectrpc.com/connect v1.21.1 + github.com/google/uuid v1.6.0 + github.com/jackc/pgx/v5 v5.12.0 github.com/santhosh-tekuri/jsonschema/v6 v6.0.3 go.yaml.in/yaml/v3 v3.0.6 google.golang.org/protobuf v1.36.12 @@ -13,8 +15,12 @@ require ( require ( cel.dev/expr v0.25.1 // indirect github.com/antlr4-go/antlr/v4 v4.13.1 // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.3 // indirect golang.org/x/exp v0.0.0-20240823005443-9b4947da3948 // indirect - golang.org/x/text v0.22.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/text v0.41.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20240826202546-f6391c0de4c7 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240826202546-f6391c0de4c7 // indirect ) diff --git a/go.sum b/go.sum index 8b19a29..440ad71 100644 --- a/go.sum +++ b/go.sum @@ -6,21 +6,47 @@ connectrpc.com/connect v1.21.1 h1:h2kacdlzXpZU2OQFSv98xkjYXNUSr/ptxmGVbS/+CRM= connectrpc.com/connect v1.21.1/go.mod h1:fhMcVJ5w6r/yTYhnkEbW6+sIF7OZ5Ay7KYgmfebh2ek= github.com/antlr4-go/antlr/v4 v4.13.1 h1:SqQKkuVZ+zWkMMNkjy5FZe5mr5WURWnlpmOuzYWrPrQ= github.com/antlr4-go/antlr/v4 v4.13.1/go.mod h1:GKmUxMtwp6ZgGwZSva4eWPC5mS6vUAmOABFgjdkM7Nw= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dlclark/regexp2 v1.11.0 h1:G/nrcoOa7ZXlpoa/91N3X7mM3r8eIlMBBJZvsz/mxKI= github.com/dlclark/regexp2 v1.11.0/go.mod h1:DHkYz0B9wPfa6wondMfaivmHpzrQ3v9q8cnmRbL6yW8= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.12.0 h1:hWwax8YiGT0YiN50U1oIDn5kgR7KC2tG3rmavwpPkxY= +github.com/jackc/pgx/v5 v5.12.0/go.mod h1:u2Fi6no03HCSBYE+ZVdLxbwdu3xyi2vU+K/YNJioSds= +github.com/jackc/puddle/v2 v2.2.3 h1:f8HoREISb3l5eO9coPgK7U3hKLhWGokOFVcThl5jrao= +github.com/jackc/puddle/v2 v2.2.3/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/santhosh-tekuri/jsonschema/v6 v6.0.3 h1:1EYB5IzjZawrrnELUi78f9fPu57HuXjmddZPjrls/28= github.com/santhosh-tekuri/jsonschema/v6 v6.0.3/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= go.yaml.in/yaml/v3 v3.0.6 h1:zcDfvjXEqTQmpjniUBLc91xYJMB2c4QOFxEF/LCEckM= go.yaml.in/yaml/v3 v3.0.6/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg= golang.org/x/exp v0.0.0-20240823005443-9b4947da3948 h1:kx6Ds3MlpiUHKj7syVnbp57++8WpuKPcR5yjLBjvLEA= golang.org/x/exp v0.0.0-20240823005443-9b4947da3948/go.mod h1:akd2r19cwCdwSwWeIdzYQGa/EZZyqcOdwWiwj5L5eKQ= -golang.org/x/text v0.22.0 h1:bofq7m3/HAFvbF51jz3Q9wLg3jkvSPuiZu/pD1XwgtM= -golang.org/x/text v0.22.0/go.mod h1:YRoo4H8PVmsu+E3Ou7cqLVH8oXWIHVoX0jqUWALQhfY= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= google.golang.org/genproto/googleapis/api v0.0.0-20240826202546-f6391c0de4c7 h1:YcyjlL1PRr2Q17/I0dPk2JmYS5CDXfcdb2Z3YRioEbw= google.golang.org/genproto/googleapis/api v0.0.0-20240826202546-f6391c0de4c7/go.mod h1:OCdP9MfskevB/rbYvHTsXTtKC+3bHWajPdoKgjcYkfo= google.golang.org/genproto/googleapis/rpc v0.0.0-20240826202546-f6391c0de4c7 h1:2035KHhUv+EpyB+hWgJnaWKJOdX1E95w2S8Rr4uWKTs= google.golang.org/genproto/googleapis/rpc v0.0.0-20240826202546-f6391c0de4c7/go.mod h1:UqMtugtsSgubUsoxbuAoiCXvqvErP7Gf0so0mK9tHxU= google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc= google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/controller/ca/ca.go b/internal/controller/ca/ca.go new file mode 100644 index 0000000..07cee03 --- /dev/null +++ b/internal/controller/ca/ca.go @@ -0,0 +1,299 @@ +// Package ca is the controller's built-in certificate authority. It issues +// the client certificates workers use for mTLS, and the controller's own +// server certificate. +// +// The CA key is kept in a file readable only by the controller. Moving it to +// an HSM or KMS is planned (see docs/ROADMAP.md, M5). +package ca + +import ( + "crypto" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/x509" + "crypto/x509/pkix" + "encoding/hex" + "encoding/pem" + "errors" + "fmt" + "math/big" + "net" + "net/url" + "os" + "path/filepath" + "time" +) + +const ( + certFile = "ca.crt" + keyFile = "ca.key" + + // WorkerCertLifetime is how long a worker certificate is valid. Workers + // will renew before expiry (not implemented yet). + WorkerCertLifetime = 30 * 24 * time.Hour + // ServerCertLifetime is how long the controller's server certificate is + // valid. It is reissued at every start. + ServerCertLifetime = 90 * 24 * time.Hour + caLifetime = 10 * 365 * 24 * time.Hour +) + +// CA signs certificates. +type CA struct { + cert *x509.Certificate + certPEM []byte + key crypto.Signer + now func() time.Time +} + +// LoadOrCreate loads the CA from dir, creating a new one on first start. +func LoadOrCreate(dir string) (*CA, error) { + certPath, keyPath := filepath.Join(dir, certFile), filepath.Join(dir, keyFile) + certPEM, certErr := os.ReadFile(certPath) + keyPEM, keyErr := os.ReadFile(keyPath) + switch { + case certErr == nil && keyErr == nil: + return parse(certPEM, keyPEM) + case errors.Is(certErr, os.ErrNotExist) && errors.Is(keyErr, os.ErrNotExist): + // First start. + case certErr != nil && !errors.Is(certErr, os.ErrNotExist): + return nil, certErr + case keyErr != nil && !errors.Is(keyErr, os.ErrNotExist): + return nil, keyErr + default: + return nil, fmt.Errorf("%s has only one of %s and %s; restore the missing file or remove both", dir, certFile, keyFile) + } + + c, keyPEM, err := create() + if err != nil { + return nil, err + } + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + // Write the key first and exclusively, so two controllers starting at + // once can't end up with different CAs. + if err := writeExclusive(keyPath, keyPEM, 0o600); err != nil { + return nil, err + } + if err := writeExclusive(certPath, c.certPEM, 0o644); err != nil { + return nil, err + } + return c, nil +} + +func writeExclusive(path string, data []byte, perm os.FileMode) error { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, perm) + if err != nil { + return err + } + if _, err := f.Write(data); err != nil { + f.Close() + return err + } + return f.Close() +} + +func create() (*CA, []byte, error) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, nil, err + } + serial, err := newSerial() + if err != nil { + return nil, nil, err + } + now := time.Now() + tmpl := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: "Opifica CA", Organization: []string{"Opifica"}}, + NotBefore: now.Add(-time.Minute), + NotAfter: now.Add(caLifetime), + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, + BasicConstraintsValid: true, + IsCA: true, + MaxPathLenZero: true, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + return nil, nil, err + } + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + return nil, nil, err + } + certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}) + c, err := parse(certPEM, keyPEM) + return c, keyPEM, err +} + +func parse(certPEM, keyPEM []byte) (*CA, error) { + cb, _ := pem.Decode(certPEM) + if cb == nil || cb.Type != "CERTIFICATE" { + return nil, errors.New("CA certificate: no PEM certificate found") + } + cert, err := x509.ParseCertificate(cb.Bytes) + if err != nil { + return nil, fmt.Errorf("CA certificate: %w", err) + } + kb, _ := pem.Decode(keyPEM) + if kb == nil { + return nil, errors.New("CA key: no PEM block found") + } + k, err := x509.ParsePKCS8PrivateKey(kb.Bytes) + if err != nil { + return nil, fmt.Errorf("CA key: %w", err) + } + signer, ok := k.(crypto.Signer) + if !ok { + return nil, errors.New("CA key cannot sign") + } + if !publicKeysEqual(cert.PublicKey, signer.Public()) { + return nil, errors.New("CA key does not match the CA certificate") + } + return &CA{cert: cert, certPEM: certPEM, key: signer, now: time.Now}, nil +} + +func publicKeysEqual(a, b crypto.PublicKey) bool { + type equaler interface{ Equal(crypto.PublicKey) bool } + e, ok := a.(equaler) + return ok && e.Equal(b) +} + +// CertPEM returns the CA certificate, which workers use to verify the controller. +func (c *CA) CertPEM() []byte { return c.certPEM } + +// Pool returns a certificate pool containing only this CA. +func (c *CA) Pool() *x509.CertPool { + p := x509.NewCertPool() + p.AddCert(c.cert) + return p +} + +// Fingerprint is "sha256:" of the CA's public key. A worker joining for +// the first time checks the controller against it, so a join token can't be +// used against an impostor controller. +func (c *CA) Fingerprint() string { + sum := sha256.Sum256(c.cert.RawSubjectPublicKeyInfo) + return "sha256:" + hex.EncodeToString(sum[:]) +} + +// WorkerURI is the identity placed in a worker certificate's URI SAN. +func WorkerURI(workerID string) string { return "opifica://worker/" + workerID } + +// SignWorkerCSR checks a worker's certificate request and issues a client +// certificate for workerID. Whatever names the CSR asks for are ignored: the +// identity comes only from workerID. +func (c *CA) SignWorkerCSR(csrPEM []byte, workerID string) (certPEM []byte, serial string, err error) { + b, _ := pem.Decode(csrPEM) + if b == nil || b.Type != "CERTIFICATE REQUEST" { + return nil, "", errors.New("no PEM certificate request found") + } + csr, err := x509.ParseCertificateRequest(b.Bytes) + if err != nil { + return nil, "", fmt.Errorf("parsing certificate request: %w", err) + } + if err := csr.CheckSignature(); err != nil { + return nil, "", fmt.Errorf("certificate request signature: %w", err) + } + if err := checkKeyStrength(csr.PublicKey); err != nil { + return nil, "", err + } + uri, err := url.Parse(WorkerURI(workerID)) + if err != nil { + return nil, "", err + } + sn, err := newSerial() + if err != nil { + return nil, "", err + } + now := c.now() + tmpl := &x509.Certificate{ + SerialNumber: sn, + Subject: pkix.Name{CommonName: workerID, Organization: []string{"Opifica worker"}}, + URIs: []*url.URL{uri}, + NotBefore: now.Add(-time.Minute), + NotAfter: now.Add(WorkerCertLifetime), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, c.cert, csr.PublicKey, c.key) + if err != nil { + return nil, "", err + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), sn.Text(16), nil +} + +// IssueServerCert issues a server certificate for the controller with the +// given DNS names and IP addresses. +func (c *CA) IssueServerCert(hosts []string) (certPEM, keyPEM []byte, err error) { + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, nil, err + } + sn, err := newSerial() + if err != nil { + return nil, nil, err + } + now := c.now() + tmpl := &x509.Certificate{ + SerialNumber: sn, + Subject: pkix.Name{CommonName: "Opifica controller"}, + NotBefore: now.Add(-time.Minute), + NotAfter: now.Add(ServerCertLifetime), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + for _, h := range hosts { + if ip := net.ParseIP(h); ip != nil { + tmpl.IPAddresses = append(tmpl.IPAddresses, ip) + } else { + tmpl.DNSNames = append(tmpl.DNSNames, h) + } + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, c.cert, &key.PublicKey, c.key) + if err != nil { + return nil, nil, err + } + keyDER, err := x509.MarshalPKCS8PrivateKey(key) + if err != nil { + return nil, nil, err + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), + pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}), nil +} + +// WorkerID returns the worker identity in a verified client certificate. +func WorkerID(cert *x509.Certificate) (string, error) { + for _, u := range cert.URIs { + if u.Scheme == "opifica" && u.Host == "worker" && len(u.Path) > 1 { + return u.Path[1:], nil + } + } + return "", errors.New("certificate has no Opifica worker identity") +} + +func checkKeyStrength(pub crypto.PublicKey) error { + switch k := pub.(type) { + case *ecdsa.PublicKey: + if k.Curve.Params().BitSize < 256 { + return fmt.Errorf("ECDSA key too small (%d bits)", k.Curve.Params().BitSize) + } + return nil + case interface{ Size() int }: // *rsa.PublicKey + if k.Size()*8 < 2048 { + return fmt.Errorf("RSA key too small (%d bits)", k.Size()*8) + } + return nil + } + if _, ok := pub.(interface{ Equal(crypto.PublicKey) bool }); ok { + return nil // ed25519 and other modern keys + } + return fmt.Errorf("unsupported key type %T", pub) +} + +func newSerial() (*big.Int, error) { + return rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128)) +} diff --git a/internal/controller/ca/ca_test.go b/internal/controller/ca/ca_test.go new file mode 100644 index 0000000..2a82296 --- /dev/null +++ b/internal/controller/ca/ca_test.go @@ -0,0 +1,235 @@ +package ca + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/rsa" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "io" + "net" + "os" + "path/filepath" + "strings" + "testing" +) + +func newCSR(t *testing.T, key any, dnsNames ...string) []byte { + t.Helper() + der, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{ + Subject: pkix.Name{CommonName: "i-am-the-controller"}, + DNSNames: dnsNames, + }, key) + if err != nil { + t.Fatal(err) + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: der}) +} + +func ecKey(t *testing.T) *ecdsa.PrivateKey { + t.Helper() + k, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + return k +} + +func parseCert(t *testing.T, p []byte) *x509.Certificate { + t.Helper() + b, _ := pem.Decode(p) + if b == nil { + t.Fatal("no PEM block") + } + c, err := x509.ParseCertificate(b.Bytes) + if err != nil { + t.Fatal(err) + } + return c +} + +func TestLoadOrCreatePersists(t *testing.T) { + dir := filepath.Join(t.TempDir(), "ca") + c1, err := LoadOrCreate(dir) + if err != nil { + t.Fatal(err) + } + c2, err := LoadOrCreate(dir) + if err != nil { + t.Fatal(err) + } + if c1.Fingerprint() != c2.Fingerprint() { + t.Error("reloading created a different CA") + } + if !strings.HasPrefix(c1.Fingerprint(), "sha256:") || len(c1.Fingerprint()) != 7+64 { + t.Errorf("fingerprint = %q", c1.Fingerprint()) + } + info, err := os.Stat(filepath.Join(dir, keyFile)) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0o600 { + t.Errorf("CA key permissions = %v, want 0600", info.Mode().Perm()) + } +} + +func TestLoadOrCreateRejectsBrokenState(t *testing.T) { + dir := t.TempDir() + if _, err := LoadOrCreate(dir); err != nil { + t.Fatal(err) + } + + // Only one of the two files. + half := t.TempDir() + data, _ := os.ReadFile(filepath.Join(dir, certFile)) + os.WriteFile(filepath.Join(half, certFile), data, 0o644) + if _, err := LoadOrCreate(half); err == nil || !strings.Contains(err.Error(), "only one of") { + t.Errorf("half-present CA: err = %v", err) + } + + // Key from a different CA. + other := t.TempDir() + if _, err := LoadOrCreate(other); err != nil { + t.Fatal(err) + } + key, _ := os.ReadFile(filepath.Join(other, keyFile)) + os.WriteFile(filepath.Join(dir, keyFile), key, 0o600) + if _, err := LoadOrCreate(dir); err == nil || !strings.Contains(err.Error(), "does not match") { + t.Errorf("mismatched key: err = %v", err) + } +} + +func TestSignWorkerCSR(t *testing.T) { + c, err := LoadOrCreate(t.TempDir()) + if err != nil { + t.Fatal(err) + } + // The CSR asks to be "the controller"; it must get only the worker identity. + certPEM, serial, err := c.SignWorkerCSR(newCSR(t, ecKey(t), "controller.example"), "w-123") + if err != nil { + t.Fatal(err) + } + cert := parseCert(t, certPEM) + if cert.SerialNumber.Text(16) != serial { + t.Errorf("serial = %s, returned %s", cert.SerialNumber.Text(16), serial) + } + if len(cert.DNSNames) != 0 || cert.Subject.CommonName != "w-123" { + t.Errorf("certificate took names from the CSR: CN=%q DNS=%v", cert.Subject.CommonName, cert.DNSNames) + } + if id, err := WorkerID(cert); err != nil || id != "w-123" { + t.Errorf("WorkerID = %q, %v", id, err) + } + if _, err := cert.Verify(x509.VerifyOptions{ + Roots: c.Pool(), KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + }); err != nil { + t.Errorf("worker certificate doesn't verify as a client certificate: %v", err) + } + if _, err := cert.Verify(x509.VerifyOptions{ + Roots: c.Pool(), KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + }); err == nil { + t.Error("worker certificate is accepted as a server certificate") + } +} + +func TestSignWorkerCSRRejects(t *testing.T) { + c, err := LoadOrCreate(t.TempDir()) + if err != nil { + t.Fatal(err) + } + weak, err := rsa.GenerateKey(rand.Reader, 1024) + if err != nil { + t.Fatal(err) + } + tampered := newCSR(t, ecKey(t)) + b, _ := pem.Decode(tampered) + b.Bytes[len(b.Bytes)-5] ^= 0xff // corrupt the signature + tampered = pem.EncodeToMemory(b) + + for name, csr := range map[string][]byte{ + "not PEM": []byte("hello"), + "wrong type": pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: []byte{1}}), + "bad signature": tampered, + "weak key": newCSR(t, weak), + } { + if _, _, err := c.SignWorkerCSR(csr, "w"); err == nil { + t.Errorf("%s: accepted", name) + } + } +} + +// A full mTLS handshake: the worker verifies the controller with the CA +// bundle, and the controller verifies the worker's client certificate. +func TestMutualTLS(t *testing.T) { + c, err := LoadOrCreate(t.TempDir()) + if err != nil { + t.Fatal(err) + } + srvCert, srvKey, err := c.IssueServerCert([]string{"controller.local", "127.0.0.1"}) + if err != nil { + t.Fatal(err) + } + serverPair, err := tls.X509KeyPair(srvCert, srvKey) + if err != nil { + t.Fatal(err) + } + + wkey := ecKey(t) + wcert, _, err := c.SignWorkerCSR(newCSR(t, wkey), "w-42") + if err != nil { + t.Fatal(err) + } + wkeyDER, _ := x509.MarshalPKCS8PrivateKey(wkey) + workerPair, err := tls.X509KeyPair(wcert, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: wkeyDER})) + if err != nil { + t.Fatal(err) + } + + ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{ + Certificates: []tls.Certificate{serverPair}, + ClientCAs: c.Pool(), + ClientAuth: tls.RequireAndVerifyClientCert, + MinVersion: tls.VersionTLS13, + }) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + gotID := make(chan string, 1) + go func() { + conn, err := ln.Accept() + if err != nil { + gotID <- "accept: " + err.Error() + return + } + defer conn.Close() + tc := conn.(*tls.Conn) + if err := tc.Handshake(); err != nil { + gotID <- "handshake: " + err.Error() + return + } + id, err := WorkerID(tc.ConnectionState().PeerCertificates[0]) + if err != nil { + gotID <- err.Error() + return + } + gotID <- id + io.Copy(io.Discard, conn) + }() + + _, port, _ := net.SplitHostPort(ln.Addr().String()) + conn, err := tls.Dial("tcp", "127.0.0.1:"+port, &tls.Config{ + Certificates: []tls.Certificate{workerPair}, + RootCAs: c.Pool(), + MinVersion: tls.VersionTLS13, + }) + if err != nil { + t.Fatalf("worker could not connect: %v", err) + } + conn.Close() + if id := <-gotID; id != "w-42" { + t.Errorf("controller saw worker %q, want w-42", id) + } +} diff --git a/internal/controller/store/db/db.go b/internal/controller/store/db/db.go new file mode 100644 index 0000000..468d1fa --- /dev/null +++ b/internal/controller/store/db/db.go @@ -0,0 +1,32 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package db + +import ( + "context" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" +) + +type DBTX interface { + Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error) + Query(context.Context, string, ...interface{}) (pgx.Rows, error) + QueryRow(context.Context, string, ...interface{}) pgx.Row +} + +func New(db DBTX) *Queries { + return &Queries{db: db} +} + +type Queries struct { + db DBTX +} + +func (q *Queries) WithTx(tx pgx.Tx) *Queries { + return &Queries{ + db: tx, + } +} diff --git a/internal/controller/store/db/models.go b/internal/controller/store/db/models.go new file mode 100644 index 0000000..524dbe8 --- /dev/null +++ b/internal/controller/store/db/models.go @@ -0,0 +1,40 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 + +package db + +import ( + "encoding/json" + "time" + + "github.com/google/uuid" +) + +type JoinToken struct { + ID uuid.UUID + TokenHash []byte + Pool string + Labels json.RawMessage + Description string + CreatedAt time.Time + ExpiresAt time.Time + UsedAt *time.Time + UsedBy *uuid.UUID +} + +type Worker struct { + ID uuid.UUID + Hostname string + Os string + Arch string + Pool string + Labels json.RawMessage + MaxConcurrentJobs int32 + Version string + CertSerial string + JoinTokenID uuid.UUID + CreatedAt time.Time + LastSeenAt *time.Time + RevokedAt *time.Time +} diff --git a/internal/controller/store/db/workers.sql.go b/internal/controller/store/db/workers.sql.go new file mode 100644 index 0000000..dfc2409 --- /dev/null +++ b/internal/controller/store/db/workers.sql.go @@ -0,0 +1,277 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: workers.sql + +package db + +import ( + "context" + "encoding/json" + "time" + + "github.com/google/uuid" +) + +const claimJoinToken = `-- name: ClaimJoinToken :one +UPDATE join_tokens +SET used_at = now() +WHERE token_hash = $1 + AND used_at IS NULL + AND expires_at > now() +RETURNING id, token_hash, pool, labels, description, created_at, expires_at, used_at, used_by +` + +// Marks a valid token as used. Returns no row if the token is unknown, +// already used or expired, so each token admits exactly one worker. +func (q *Queries) ClaimJoinToken(ctx context.Context, tokenHash []byte) (JoinToken, error) { + row := q.db.QueryRow(ctx, claimJoinToken, tokenHash) + var i JoinToken + err := row.Scan( + &i.ID, + &i.TokenHash, + &i.Pool, + &i.Labels, + &i.Description, + &i.CreatedAt, + &i.ExpiresAt, + &i.UsedAt, + &i.UsedBy, + ) + return i, err +} + +const createJoinToken = `-- name: CreateJoinToken :one +INSERT INTO join_tokens (id, token_hash, pool, labels, description, expires_at) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING id, token_hash, pool, labels, description, created_at, expires_at, used_at, used_by +` + +type CreateJoinTokenParams struct { + ID uuid.UUID + TokenHash []byte + Pool string + Labels json.RawMessage + Description string + ExpiresAt time.Time +} + +func (q *Queries) CreateJoinToken(ctx context.Context, arg CreateJoinTokenParams) (JoinToken, error) { + row := q.db.QueryRow(ctx, createJoinToken, + arg.ID, + arg.TokenHash, + arg.Pool, + arg.Labels, + arg.Description, + arg.ExpiresAt, + ) + var i JoinToken + err := row.Scan( + &i.ID, + &i.TokenHash, + &i.Pool, + &i.Labels, + &i.Description, + &i.CreatedAt, + &i.ExpiresAt, + &i.UsedAt, + &i.UsedBy, + ) + return i, err +} + +const createWorker = `-- name: CreateWorker :one +INSERT INTO workers (id, hostname, os, arch, pool, labels, max_concurrent_jobs, version, cert_serial, join_token_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) +RETURNING id, hostname, os, arch, pool, labels, max_concurrent_jobs, version, cert_serial, join_token_id, created_at, last_seen_at, revoked_at +` + +type CreateWorkerParams struct { + ID uuid.UUID + Hostname string + Os string + Arch string + Pool string + Labels json.RawMessage + MaxConcurrentJobs int32 + Version string + CertSerial string + JoinTokenID uuid.UUID +} + +func (q *Queries) CreateWorker(ctx context.Context, arg CreateWorkerParams) (Worker, error) { + row := q.db.QueryRow(ctx, createWorker, + arg.ID, + arg.Hostname, + arg.Os, + arg.Arch, + arg.Pool, + arg.Labels, + arg.MaxConcurrentJobs, + arg.Version, + arg.CertSerial, + arg.JoinTokenID, + ) + var i Worker + err := row.Scan( + &i.ID, + &i.Hostname, + &i.Os, + &i.Arch, + &i.Pool, + &i.Labels, + &i.MaxConcurrentJobs, + &i.Version, + &i.CertSerial, + &i.JoinTokenID, + &i.CreatedAt, + &i.LastSeenAt, + &i.RevokedAt, + ) + return i, err +} + +const getWorker = `-- name: GetWorker :one +SELECT id, hostname, os, arch, pool, labels, max_concurrent_jobs, version, cert_serial, join_token_id, created_at, last_seen_at, revoked_at FROM workers WHERE id = $1 +` + +func (q *Queries) GetWorker(ctx context.Context, id uuid.UUID) (Worker, error) { + row := q.db.QueryRow(ctx, getWorker, id) + var i Worker + err := row.Scan( + &i.ID, + &i.Hostname, + &i.Os, + &i.Arch, + &i.Pool, + &i.Labels, + &i.MaxConcurrentJobs, + &i.Version, + &i.CertSerial, + &i.JoinTokenID, + &i.CreatedAt, + &i.LastSeenAt, + &i.RevokedAt, + ) + return i, err +} + +const listJoinTokens = `-- name: ListJoinTokens :many +SELECT id, token_hash, pool, labels, description, created_at, expires_at, used_at, used_by FROM join_tokens ORDER BY created_at DESC +` + +func (q *Queries) ListJoinTokens(ctx context.Context) ([]JoinToken, error) { + rows, err := q.db.Query(ctx, listJoinTokens) + if err != nil { + return nil, err + } + defer rows.Close() + var items []JoinToken + for rows.Next() { + var i JoinToken + if err := rows.Scan( + &i.ID, + &i.TokenHash, + &i.Pool, + &i.Labels, + &i.Description, + &i.CreatedAt, + &i.ExpiresAt, + &i.UsedAt, + &i.UsedBy, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listWorkers = `-- name: ListWorkers :many +SELECT id, hostname, os, arch, pool, labels, max_concurrent_jobs, version, cert_serial, join_token_id, created_at, last_seen_at, revoked_at FROM workers ORDER BY created_at +` + +func (q *Queries) ListWorkers(ctx context.Context) ([]Worker, error) { + rows, err := q.db.Query(ctx, listWorkers) + if err != nil { + return nil, err + } + defer rows.Close() + var items []Worker + for rows.Next() { + var i Worker + if err := rows.Scan( + &i.ID, + &i.Hostname, + &i.Os, + &i.Arch, + &i.Pool, + &i.Labels, + &i.MaxConcurrentJobs, + &i.Version, + &i.CertSerial, + &i.JoinTokenID, + &i.CreatedAt, + &i.LastSeenAt, + &i.RevokedAt, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const revokeWorker = `-- name: RevokeWorker :one +UPDATE workers SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL RETURNING id, hostname, os, arch, pool, labels, max_concurrent_jobs, version, cert_serial, join_token_id, created_at, last_seen_at, revoked_at +` + +func (q *Queries) RevokeWorker(ctx context.Context, id uuid.UUID) (Worker, error) { + row := q.db.QueryRow(ctx, revokeWorker, id) + var i Worker + err := row.Scan( + &i.ID, + &i.Hostname, + &i.Os, + &i.Arch, + &i.Pool, + &i.Labels, + &i.MaxConcurrentJobs, + &i.Version, + &i.CertSerial, + &i.JoinTokenID, + &i.CreatedAt, + &i.LastSeenAt, + &i.RevokedAt, + ) + return i, err +} + +const setJoinTokenWorker = `-- name: SetJoinTokenWorker :exec +UPDATE join_tokens SET used_by = $2 WHERE id = $1 +` + +type SetJoinTokenWorkerParams struct { + ID uuid.UUID + UsedBy *uuid.UUID +} + +func (q *Queries) SetJoinTokenWorker(ctx context.Context, arg SetJoinTokenWorkerParams) error { + _, err := q.db.Exec(ctx, setJoinTokenWorker, arg.ID, arg.UsedBy) + return err +} + +const touchWorker = `-- name: TouchWorker :exec +UPDATE workers SET last_seen_at = now() WHERE id = $1 +` + +func (q *Queries) TouchWorker(ctx context.Context, id uuid.UUID) error { + _, err := q.db.Exec(ctx, touchWorker, id) + return err +} diff --git a/internal/controller/store/migrations/0001_workers.sql b/internal/controller/store/migrations/0001_workers.sql new file mode 100644 index 0000000..48d79c4 --- /dev/null +++ b/internal/controller/store/migrations/0001_workers.sql @@ -0,0 +1,37 @@ +-- One-time tokens an admin creates so a new build machine can join. +-- Only a hash of the token is stored; the token itself is shown once. +CREATE TABLE join_tokens ( + id uuid PRIMARY KEY, + token_hash bytea NOT NULL UNIQUE, + -- The pool is decided by the admin who created the token, never by the + -- worker, so a machine can't place itself in a more trusted pool. + pool text NOT NULL, + labels jsonb NOT NULL DEFAULT '{}', + description text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL, + used_at timestamptz, + used_by uuid +); + +CREATE TABLE workers ( + id uuid PRIMARY KEY, + hostname text NOT NULL, + os text NOT NULL, + arch text NOT NULL, + pool text NOT NULL, + labels jsonb NOT NULL DEFAULT '{}', + max_concurrent_jobs integer NOT NULL CHECK (max_concurrent_jobs > 0), + version text NOT NULL DEFAULT '', + -- Serial of the client certificate issued at join, hex-encoded. + cert_serial text NOT NULL, + join_token_id uuid NOT NULL REFERENCES join_tokens (id), + created_at timestamptz NOT NULL DEFAULT now(), + last_seen_at timestamptz, + revoked_at timestamptz +); + +ALTER TABLE join_tokens + ADD CONSTRAINT join_tokens_used_by_fkey FOREIGN KEY (used_by) REFERENCES workers (id); + +CREATE INDEX workers_pool ON workers (pool) WHERE revoked_at IS NULL; diff --git a/internal/controller/store/queries/workers.sql b/internal/controller/store/queries/workers.sql new file mode 100644 index 0000000..5419fbf --- /dev/null +++ b/internal/controller/store/queries/workers.sql @@ -0,0 +1,37 @@ +-- name: CreateJoinToken :one +INSERT INTO join_tokens (id, token_hash, pool, labels, description, expires_at) +VALUES ($1, $2, $3, $4, $5, $6) +RETURNING *; + +-- name: ClaimJoinToken :one +-- Marks a valid token as used. Returns no row if the token is unknown, +-- already used or expired, so each token admits exactly one worker. +UPDATE join_tokens +SET used_at = now() +WHERE token_hash = $1 + AND used_at IS NULL + AND expires_at > now() +RETURNING *; + +-- name: SetJoinTokenWorker :exec +UPDATE join_tokens SET used_by = $2 WHERE id = $1; + +-- name: ListJoinTokens :many +SELECT * FROM join_tokens ORDER BY created_at DESC; + +-- name: CreateWorker :one +INSERT INTO workers (id, hostname, os, arch, pool, labels, max_concurrent_jobs, version, cert_serial, join_token_id) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) +RETURNING *; + +-- name: GetWorker :one +SELECT * FROM workers WHERE id = $1; + +-- name: ListWorkers :many +SELECT * FROM workers ORDER BY created_at; + +-- name: TouchWorker :exec +UPDATE workers SET last_seen_at = now() WHERE id = $1; + +-- name: RevokeWorker :one +UPDATE workers SET revoked_at = now() WHERE id = $1 AND revoked_at IS NULL RETURNING *; diff --git a/internal/controller/store/store.go b/internal/controller/store/store.go new file mode 100644 index 0000000..41a24cd --- /dev/null +++ b/internal/controller/store/store.go @@ -0,0 +1,156 @@ +// Package store is the controller's PostgreSQL storage. Queries are written +// in queries/*.sql and compiled to Go by sqlc (package db); the schema lives +// in migrations/*.sql and is applied at startup. +package store + +import ( + "context" + "embed" + "errors" + "fmt" + "io/fs" + "sort" + "strconv" + "strings" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/opifica/opifica/internal/controller/store/db" +) + +//go:embed migrations/*.sql +var migrationFiles embed.FS + +// Store wraps a connection pool and the generated queries. +type Store struct { + Pool *pgxpool.Pool + *db.Queries +} + +// Open connects to PostgreSQL and applies pending migrations. +func Open(ctx context.Context, url string) (*Store, error) { + pool, err := pgxpool.New(ctx, url) + if err != nil { + return nil, fmt.Errorf("connecting to database: %w", err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, fmt.Errorf("connecting to database: %w", err) + } + if err := Migrate(ctx, pool); err != nil { + pool.Close() + return nil, err + } + return &Store{Pool: pool, Queries: db.New(pool)}, nil +} + +// Close closes the connection pool. +func (s *Store) Close() { s.Pool.Close() } + +// Tx runs fn in a transaction, committing if it returns nil. +func (s *Store) Tx(ctx context.Context, fn func(q *db.Queries) error) error { + return pgx.BeginFunc(ctx, s.Pool, func(tx pgx.Tx) error { + return fn(s.Queries.WithTx(tx)) + }) +} + +// IsNotFound reports whether err means a query returned no row. +func IsNotFound(err error) bool { return errors.Is(err, pgx.ErrNoRows) } + +type migration struct { + version int + name string + sql string +} + +func loadMigrations() ([]migration, error) { + entries, err := fs.ReadDir(migrationFiles, "migrations") + if err != nil { + return nil, err + } + var ms []migration + for _, e := range entries { + num, _, ok := strings.Cut(e.Name(), "_") + v, err := strconv.Atoi(num) + if !ok || err != nil { + return nil, fmt.Errorf("migration %q: name must start with a number, e.g. 0002_runs.sql", e.Name()) + } + body, err := migrationFiles.ReadFile("migrations/" + e.Name()) + if err != nil { + return nil, err + } + ms = append(ms, migration{version: v, name: e.Name(), sql: string(body)}) + } + sort.Slice(ms, func(i, j int) bool { return ms[i].version < ms[j].version }) + for i := 1; i < len(ms); i++ { + if ms[i].version == ms[i-1].version { + return nil, fmt.Errorf("migrations %q and %q have the same number", ms[i-1].name, ms[i].name) + } + } + return ms, nil +} + +// migrationLock is a PostgreSQL advisory lock key, so controllers starting at +// the same time don't apply migrations twice. +const migrationLock = 0x6f70696669636121 // "opifica!" + +// Migrate applies every migration not yet recorded in schema_migrations, each +// in its own transaction. +func Migrate(ctx context.Context, pool *pgxpool.Pool) error { + ms, err := loadMigrations() + if err != nil { + return err + } + conn, err := pool.Acquire(ctx) + if err != nil { + return err + } + defer conn.Release() + + if _, err := conn.Exec(ctx, "SELECT pg_advisory_lock($1)", int64(migrationLock)); err != nil { + return fmt.Errorf("taking migration lock: %w", err) + } + defer conn.Exec(context.WithoutCancel(ctx), "SELECT pg_advisory_unlock($1)", int64(migrationLock)) + + if _, err := conn.Exec(ctx, `CREATE TABLE IF NOT EXISTS schema_migrations ( + version integer PRIMARY KEY, + name text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT now() + )`); err != nil { + return fmt.Errorf("creating schema_migrations: %w", err) + } + + applied := map[int]bool{} + rows, err := conn.Query(ctx, "SELECT version FROM schema_migrations") + if err != nil { + return err + } + for rows.Next() { + var v int + if err := rows.Scan(&v); err != nil { + return err + } + applied[v] = true + } + if err := rows.Err(); err != nil { + return err + } + + for _, m := range ms { + if applied[m.version] { + continue + } + err := pgx.BeginFunc(ctx, conn, func(tx pgx.Tx) error { + if _, err := tx.Exec(ctx, m.sql); err != nil { + return err + } + _, err := tx.Exec(ctx, "INSERT INTO schema_migrations (version, name) VALUES ($1, $2)", m.version, m.name) + return err + }) + if err != nil { + return fmt.Errorf("applying migration %s: %w", m.name, err) + } + } + return nil +} diff --git a/internal/controller/store/store_test.go b/internal/controller/store/store_test.go new file mode 100644 index 0000000..13b3907 --- /dev/null +++ b/internal/controller/store/store_test.go @@ -0,0 +1,140 @@ +package store_test + +import ( + "context" + "encoding/json" + "os" + "sync" + "testing" + "time" + + "github.com/google/uuid" + + "github.com/opifica/opifica/internal/controller/store" + "github.com/opifica/opifica/internal/controller/store/db" + "github.com/opifica/opifica/internal/testutil/pgtest" +) + +func TestMain(m *testing.M) { os.Exit(pgtest.Run(m)) } + +func open(t *testing.T) *store.Store { + t.Helper() + s, err := store.Open(context.Background(), pgtest.New(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + return s +} + +func TestMigrateIsIdempotentAndConcurrent(t *testing.T) { + url := pgtest.New(t) + ctx := context.Background() + var wg sync.WaitGroup + errs := make(chan error, 4) + for range 4 { + wg.Add(1) + go func() { + defer wg.Done() + s, err := store.Open(ctx, url) + if err != nil { + errs <- err + return + } + s.Close() + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Errorf("concurrent Open: %v", err) + } + + s, err := store.Open(ctx, url) + if err != nil { + t.Fatal(err) + } + defer s.Close() + var n int + if err := s.Pool.QueryRow(ctx, "SELECT count(*) FROM schema_migrations").Scan(&n); err != nil { + t.Fatal(err) + } + if n == 0 { + t.Error("no migrations recorded") + } +} + +func newToken(t *testing.T, s *store.Store, hash string, expires time.Time) db.JoinToken { + t.Helper() + tok, err := s.CreateJoinToken(context.Background(), db.CreateJoinTokenParams{ + ID: uuid.New(), TokenHash: []byte(hash), Pool: "trusted", + Labels: json.RawMessage(`{}`), ExpiresAt: expires, + }) + if err != nil { + t.Fatal(err) + } + return tok +} + +func TestJoinTokenSingleUse(t *testing.T) { + s := open(t) + ctx := context.Background() + newToken(t, s, "valid", time.Now().Add(time.Hour)) + newToken(t, s, "expired", time.Now().Add(-time.Minute)) + + if _, err := s.ClaimJoinToken(ctx, []byte("valid")); err != nil { + t.Fatalf("first claim: %v", err) + } + if _, err := s.ClaimJoinToken(ctx, []byte("valid")); !store.IsNotFound(err) { + t.Errorf("second claim of the same token: err = %v, want not found", err) + } + if _, err := s.ClaimJoinToken(ctx, []byte("expired")); !store.IsNotFound(err) { + t.Errorf("claim of expired token: err = %v, want not found", err) + } + if _, err := s.ClaimJoinToken(ctx, []byte("unknown")); !store.IsNotFound(err) { + t.Errorf("claim of unknown token: err = %v, want not found", err) + } +} + +// Many workers racing for one token: exactly one may win. +func TestJoinTokenClaimRace(t *testing.T) { + s := open(t) + newToken(t, s, "contested", time.Now().Add(time.Hour)) + var wg sync.WaitGroup + var mu sync.Mutex + wins := 0 + for range 10 { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := s.ClaimJoinToken(context.Background(), []byte("contested")); err == nil { + mu.Lock() + wins++ + mu.Unlock() + } + }() + } + wg.Wait() + if wins != 1 { + t.Errorf("%d claims succeeded, want exactly 1", wins) + } +} + +func TestTxRollsBack(t *testing.T) { + s := open(t) + ctx := context.Background() + tok := newToken(t, s, "tx", time.Now().Add(time.Hour)) + + err := s.Tx(ctx, func(q *db.Queries) error { + if _, err := q.ClaimJoinToken(ctx, []byte("tx")); err != nil { + return err + } + return context.Canceled // anything non-nil rolls back + }) + if err == nil { + t.Fatal("Tx returned nil") + } + if _, err := s.ClaimJoinToken(ctx, tok.TokenHash); err != nil { + t.Errorf("token was consumed despite rollback: %v", err) + } +} diff --git a/internal/controller/workers/join.go b/internal/controller/workers/join.go new file mode 100644 index 0000000..256b5ed --- /dev/null +++ b/internal/controller/workers/join.go @@ -0,0 +1,271 @@ +// Package workers manages build machines: join tokens, joining, and (later) +// the Connect stream and worker registry. +package workers + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "log/slog" + "maps" + "regexp" + "strings" + "time" + + "connectrpc.com/connect" + "github.com/google/uuid" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "github.com/opifica/opifica/internal/controller/ca" + "github.com/opifica/opifica/internal/controller/store" + "github.com/opifica/opifica/internal/controller/store/db" +) + +const ( + tokenPrefix = "opj_" + + // DefaultTokenTTL is how long a join token is valid if not specified. + DefaultTokenTTL = 24 * time.Hour + maxTokenTTL = 30 * 24 * time.Hour + maxJobsPerHost = 256 +) + +var ( + poolRe = regexp.MustCompile(`^[a-z][a-z0-9-]{0,62}$`) + labelRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._/-]{0,62}$`) +) + +// reservedLabels can't be set by a worker: they come from the admin. +var reservedLabels = []string{"pool"} + +// Service implements joining. It is used by the WorkerService handler and by +// admin tooling that creates tokens. +type Service struct { + Store *store.Store + CA *ca.CA + Log *slog.Logger + Now func() time.Time +} + +func (s *Service) now() time.Time { + if s.Now != nil { + return s.Now() + } + return time.Now() +} + +// TokenRequest describes a join token to create. +type TokenRequest struct { + Pool string + Labels map[string]string + Description string + TTL time.Duration +} + +// CreatedToken is a new join token. Secret is shown once and never stored. +type CreatedToken struct { + ID uuid.UUID + Secret string + ExpiresAt time.Time + // CAFingerprint lets the worker check it is talking to this controller. + CAFingerprint string +} + +// CreateToken creates a one-time join token for a worker in req.Pool. +func (s *Service) CreateToken(ctx context.Context, req TokenRequest) (*CreatedToken, error) { + if !poolRe.MatchString(req.Pool) { + return nil, fmt.Errorf("invalid pool name %q (lowercase letters, digits and -)", req.Pool) + } + if err := checkLabels(req.Labels); err != nil { + return nil, err + } + ttl := req.TTL + if ttl == 0 { + ttl = DefaultTokenTTL + } + if ttl < 0 || ttl > maxTokenTTL { + return nil, fmt.Errorf("token lifetime must be between 0 and %v", maxTokenTTL) + } + labels := req.Labels + if labels == nil { + labels = map[string]string{} + } + labelsJSON, err := json.Marshal(labels) + if err != nil { + return nil, err + } + + raw := make([]byte, 32) + if _, err := rand.Read(raw); err != nil { + return nil, err + } + secret := tokenPrefix + base64.RawURLEncoding.EncodeToString(raw) + tok, err := s.Store.CreateJoinToken(ctx, db.CreateJoinTokenParams{ + ID: uuid.New(), + TokenHash: hashToken(secret), + Pool: req.Pool, + Labels: labelsJSON, + Description: req.Description, + ExpiresAt: s.now().Add(ttl), + }) + if err != nil { + return nil, fmt.Errorf("storing join token: %w", err) + } + return &CreatedToken{ID: tok.ID, Secret: secret, ExpiresAt: tok.ExpiresAt, CAFingerprint: s.CA.Fingerprint()}, nil +} + +func hashToken(secret string) []byte { + h := sha256.Sum256([]byte(secret)) + return h[:] +} + +// errBadToken is deliberately the same for unknown, used and expired tokens, +// so a caller learns nothing about which tokens exist. +var errBadToken = connect.NewError(connect.CodeUnauthenticated, errors.New("invalid or expired join token")) + +// Join admits a new worker: it consumes the join token, records the worker +// and signs its certificate request, all in one transaction. +func (s *Service) Join(ctx context.Context, req *opificav1.JoinRequest) (*opificav1.JoinResponse, error) { + if !strings.HasPrefix(req.GetJoinToken(), tokenPrefix) { + return nil, errBadToken + } + info := req.GetInfo() + osName, arch, err := checkInfo(info) + if err != nil { + return nil, connect.NewError(connect.CodeInvalidArgument, err) + } + if len(req.GetCsrPem()) == 0 { + return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("csr_pem is required")) + } + + workerID := uuid.New() + var certPEM []byte + var pool string + err = s.Store.Tx(ctx, func(q *db.Queries) error { + tok, err := q.ClaimJoinToken(ctx, hashToken(req.GetJoinToken())) + if store.IsNotFound(err) { + return errBadToken + } + if err != nil { + return err + } + pool = tok.Pool + + var tokenLabels map[string]string + if err := json.Unmarshal(tok.Labels, &tokenLabels); err != nil { + return fmt.Errorf("join token labels: %w", err) + } + // Admin-assigned labels win over what the worker reports. + labels := maps.Clone(info.GetLabels()) + if labels == nil { + labels = map[string]string{} + } + maps.Copy(labels, tokenLabels) + labelsJSON, err := json.Marshal(labels) + if err != nil { + return err + } + + cert, serial, err := s.CA.SignWorkerCSR(req.GetCsrPem(), workerID.String()) + if err != nil { + return connect.NewError(connect.CodeInvalidArgument, err) + } + certPEM = cert + + if _, err := q.CreateWorker(ctx, db.CreateWorkerParams{ + ID: workerID, + Hostname: info.GetHostname(), + Os: osName, + Arch: arch, + Pool: tok.Pool, + Labels: labelsJSON, + MaxConcurrentJobs: int32(info.GetMaxConcurrentJobs()), + Version: info.GetVersion(), + CertSerial: serial, + JoinTokenID: tok.ID, + }); err != nil { + return err + } + return q.SetJoinTokenWorker(ctx, db.SetJoinTokenWorkerParams{ID: tok.ID, UsedBy: &workerID}) + }) + if err != nil { + var cerr *connect.Error + if errors.As(err, &cerr) { + return nil, cerr + } + s.log().Error("join failed", "err", err) + return nil, connect.NewError(connect.CodeInternal, errors.New("join failed")) + } + + s.log().Info("worker joined", "worker", workerID, "hostname", info.GetHostname(), "os", osName, "arch", arch, "pool", pool) + return &opificav1.JoinResponse{ + WorkerId: workerID.String(), + CertificatePem: certPEM, + CaBundlePem: s.CA.CertPEM(), + }, nil +} + +func (s *Service) log() *slog.Logger { + if s.Log != nil { + return s.Log + } + return slog.Default() +} + +func checkInfo(info *opificav1.WorkerInfo) (osName, arch string, err error) { + if info == nil { + return "", "", errors.New("info is required") + } + switch info.GetOs() { + case opificav1.OperatingSystem_OPERATING_SYSTEM_LINUX: + osName = "linux" + case opificav1.OperatingSystem_OPERATING_SYSTEM_WINDOWS: + osName = "windows" + case opificav1.OperatingSystem_OPERATING_SYSTEM_MACOS: + osName = "macos" + default: + return "", "", errors.New("info.os is required") + } + switch info.GetArch() { + case opificav1.Architecture_ARCHITECTURE_AMD64: + arch = "amd64" + case opificav1.Architecture_ARCHITECTURE_ARM64: + arch = "arm64" + default: + return "", "", errors.New("info.arch is required") + } + if h := info.GetHostname(); h == "" || len(h) > 253 { + return "", "", errors.New("info.hostname is required (at most 253 characters)") + } + if n := info.GetMaxConcurrentJobs(); n < 1 || n > maxJobsPerHost { + return "", "", fmt.Errorf("info.max_concurrent_jobs must be between 1 and %d", maxJobsPerHost) + } + if err := checkLabels(info.GetLabels()); err != nil { + return "", "", err + } + for _, r := range reservedLabels { + if _, ok := info.GetLabels()[r]; ok { + return "", "", fmt.Errorf("label %q is set by the join token, not by the worker", r) + } + } + return osName, arch, nil +} + +func checkLabels(labels map[string]string) error { + if len(labels) > 64 { + return errors.New("at most 64 labels") + } + for k, v := range labels { + if !labelRe.MatchString(k) { + return fmt.Errorf("invalid label key %q", k) + } + if len(v) > 255 { + return fmt.Errorf("label %q: value longer than 255 characters", k) + } + } + return nil +} diff --git a/internal/controller/workers/join_test.go b/internal/controller/workers/join_test.go new file mode 100644 index 0000000..18cae65 --- /dev/null +++ b/internal/controller/workers/join_test.go @@ -0,0 +1,239 @@ +package workers_test + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "encoding/json" + "encoding/pem" + "errors" + "io" + "log/slog" + "os" + "strings" + "sync" + "testing" + "time" + + "connectrpc.com/connect" + "github.com/google/uuid" + + opificav1 "github.com/opifica/opifica/gen/opifica/v1" + "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/testutil/pgtest" +) + +func TestMain(m *testing.M) { os.Exit(pgtest.Run(m)) } + +func newService(t *testing.T) *workers.Service { + t.Helper() + ctx := context.Background() + st, err := store.Open(ctx, pgtest.New(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(st.Close) + authority, err := ca.LoadOrCreate(t.TempDir()) + if err != nil { + t.Fatal(err) + } + return &workers.Service{Store: st, CA: authority, Log: slog.New(slog.NewTextHandler(io.Discard, nil))} +} + +func csr(t *testing.T) []byte { + t.Helper() + key, _ := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + der, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{}, key) + if err != nil { + t.Fatal(err) + } + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE REQUEST", Bytes: der}) +} + +func joinRequest(t *testing.T, token string) *opificav1.JoinRequest { + return &opificav1.JoinRequest{ + JoinToken: token, + CsrPem: csr(t), + Info: &opificav1.WorkerInfo{ + Hostname: "mac-mini-1", + Os: opificav1.OperatingSystem_OPERATING_SYSTEM_MACOS, + Arch: opificav1.Architecture_ARCHITECTURE_ARM64, + MaxConcurrentJobs: 2, + Labels: map[string]string{"xcode": "16", "region": "worker-says-this"}, + Version: "dev", + }, + } +} + +func codeOf(err error) connect.Code { + var cerr *connect.Error + if errors.As(err, &cerr) { + return cerr.Code() + } + return 0 +} + +func TestJoin(t *testing.T) { + s := newService(t) + ctx := context.Background() + tok, err := s.CreateToken(ctx, workers.TokenRequest{ + Pool: "release", Labels: map[string]string{"region": "eu-west"}, Description: "office Mac", + }) + if err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(tok.Secret, "opj_") || tok.CAFingerprint != s.CA.Fingerprint() { + t.Errorf("token = %+v", tok) + } + + resp, err := s.Join(ctx, joinRequest(t, tok.Secret)) + if err != nil { + t.Fatalf("Join: %v", err) + } + b, _ := pem.Decode(resp.CertificatePem) + cert, err := x509.ParseCertificate(b.Bytes) + if err != nil { + t.Fatal(err) + } + if id, _ := ca.WorkerID(cert); id != resp.WorkerId { + t.Errorf("certificate identity %q != worker id %q", id, resp.WorkerId) + } + if string(resp.CaBundlePem) != string(s.CA.CertPEM()) { + t.Error("CA bundle missing from response") + } + + w, err := s.Store.GetWorker(ctx, uuid.MustParse(resp.WorkerId)) + if err != nil { + t.Fatal(err) + } + if w.Pool != "release" || w.Os != "macos" || w.Arch != "arm64" || w.MaxConcurrentJobs != 2 { + t.Errorf("stored worker = %+v", w) + } + if w.CertSerial != cert.SerialNumber.Text(16) { + t.Error("stored certificate serial doesn't match the issued certificate") + } + var labels map[string]string + json.Unmarshal(w.Labels, &labels) + if labels["xcode"] != "16" || labels["region"] != "eu-west" { + t.Errorf("labels = %v; token labels must win over the worker's", labels) + } + + // The token is spent. + if _, err := s.Join(ctx, joinRequest(t, tok.Secret)); codeOf(err) != connect.CodeUnauthenticated { + t.Errorf("second join with the same token: err = %v", err) + } +} + +func TestJoinRejectsBadTokens(t *testing.T) { + s := newService(t) + ctx := context.Background() + now := time.Now() + s.Now = func() time.Time { return now.Add(-2 * time.Hour) } + expired, err := s.CreateToken(ctx, workers.TokenRequest{Pool: "trusted", TTL: time.Hour}) + if err != nil { + t.Fatal(err) + } + s.Now = nil + + for name, token := range map[string]string{ + "empty": "", + "no prefix": "hunter2", + "unknown": "opj_doesnotexist", + "expired": expired.Secret, + } { + _, err := s.Join(ctx, joinRequest(t, token)) + if codeOf(err) != connect.CodeUnauthenticated { + t.Errorf("%s: err = %v, want unauthenticated", name, err) + } + if err != nil && !strings.Contains(err.Error(), "invalid or expired join token") { + t.Errorf("%s: error should be the generic message, got %v", name, err) + } + } +} + +func TestJoinRejectsBadRequests(t *testing.T) { + s := newService(t) + ctx := context.Background() + tests := map[string]func(*opificav1.JoinRequest){ + "no info": func(r *opificav1.JoinRequest) { r.Info = nil }, + "no os": func(r *opificav1.JoinRequest) { r.Info.Os = 0 }, + "no arch": func(r *opificav1.JoinRequest) { r.Info.Arch = 0 }, + "no hostname": func(r *opificav1.JoinRequest) { r.Info.Hostname = "" }, + "zero jobs": func(r *opificav1.JoinRequest) { r.Info.MaxConcurrentJobs = 0 }, + "too many jobs": func(r *opificav1.JoinRequest) { r.Info.MaxConcurrentJobs = 10000 }, + "chooses own pool": func(r *opificav1.JoinRequest) { r.Info.Labels["pool"] = "release" }, + "bad label": func(r *opificav1.JoinRequest) { r.Info.Labels["Bad Key"] = "x" }, + "no csr": func(r *opificav1.JoinRequest) { r.CsrPem = nil }, + "bad csr": func(r *opificav1.JoinRequest) { r.CsrPem = []byte("nope") }, + } + for name, mutate := range tests { + t.Run(name, func(t *testing.T) { + tok, err := s.CreateToken(ctx, workers.TokenRequest{Pool: "trusted"}) + if err != nil { + t.Fatal(err) + } + req := joinRequest(t, tok.Secret) + mutate(req) + if _, err := s.Join(ctx, req); codeOf(err) != connect.CodeInvalidArgument { + t.Errorf("err = %v, want invalid argument", err) + } + // A rejected request must not burn the token. + if _, err := s.Join(ctx, joinRequest(t, tok.Secret)); err != nil { + t.Errorf("token unusable after a rejected request: %v", err) + } + }) + } +} + +// Ten machines racing with one token: exactly one becomes a worker. +func TestJoinRace(t *testing.T) { + s := newService(t) + ctx := context.Background() + tok, err := s.CreateToken(ctx, workers.TokenRequest{Pool: "trusted"}) + if err != nil { + t.Fatal(err) + } + var wg sync.WaitGroup + var mu sync.Mutex + joined := 0 + for range 10 { + req := joinRequest(t, tok.Secret) + wg.Add(1) + go func() { + defer wg.Done() + if _, err := s.Join(ctx, req); err == nil { + mu.Lock() + joined++ + mu.Unlock() + } + }() + } + wg.Wait() + ws, err := s.Store.ListWorkers(ctx) + if err != nil { + t.Fatal(err) + } + if joined != 1 || len(ws) != 1 { + t.Errorf("joined=%d workers=%d, want 1 and 1", joined, len(ws)) + } +} + +func TestCreateTokenValidation(t *testing.T) { + s := newService(t) + ctx := context.Background() + for name, req := range map[string]workers.TokenRequest{ + "no pool": {}, + "bad pool": {Pool: "Release!"}, + "bad label": {Pool: "trusted", Labels: map[string]string{"": "x"}}, + "ttl too big": {Pool: "trusted", TTL: 365 * 24 * time.Hour}, + "ttl < 0": {Pool: "trusted", TTL: -time.Hour}, + } { + if _, err := s.CreateToken(ctx, req); err == nil { + t.Errorf("%s: accepted", name) + } + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 51e09cb..9a79dba 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -3,6 +3,7 @@ package server import ( "context" + "crypto/tls" "errors" "log/slog" "net" @@ -17,9 +18,8 @@ func Health(w http.ResponseWriter, _ *http.Request) { } // Run serves h on addr over HTTP/1.1 and unencrypted HTTP/2 (needed for -// streaming RPCs) until ctx is cancelled, then shuts down gracefully. -// -// TODO: mTLS. Until then, run services only on trusted networks. +// streaming RPCs) until ctx is cancelled, then shuts down gracefully. Use it +// only where the channel is already private, e.g. the agent inside a VM. func Run(ctx context.Context, addr string, h http.Handler, log *slog.Logger) error { ln, err := net.Listen("tcp", addr) if err != nil { @@ -30,18 +30,48 @@ func Run(ctx context.Context, addr string, h http.Handler, log *slog.Logger) err // Serve is Run on an existing listener. func Serve(ctx context.Context, ln net.Listener, h http.Handler, log *slog.Logger) error { + return serve(ctx, ln, h, nil, log) +} + +// RunTLS serves h on addr over TLS (HTTP/1.1 and HTTP/2) until ctx is +// cancelled. +func RunTLS(ctx context.Context, addr string, h http.Handler, cfg *tls.Config, log *slog.Logger) error { + ln, err := net.Listen("tcp", addr) + if err != nil { + return err + } + return ServeTLS(ctx, ln, h, cfg, log) +} + +// ServeTLS is RunTLS on an existing listener. +func ServeTLS(ctx context.Context, ln net.Listener, h http.Handler, cfg *tls.Config, log *slog.Logger) error { + return serve(ctx, ln, h, cfg, log) +} + +func serve(ctx context.Context, ln net.Listener, h http.Handler, tlsCfg *tls.Config, log *slog.Logger) error { protocols := new(http.Protocols) protocols.SetHTTP1(true) - protocols.SetUnencryptedHTTP2(true) + if tlsCfg != nil { + protocols.SetHTTP2(true) + } else { + protocols.SetUnencryptedHTTP2(true) + } srv := &http.Server{ Handler: h, Protocols: protocols, + TLSConfig: tlsCfg, ReadHeaderTimeout: 10 * time.Second, } errc := make(chan error, 1) - go func() { errc <- srv.Serve(ln) }() - log.Info("listening", "addr", ln.Addr().String()) + go func() { + if tlsCfg != nil { + errc <- srv.ServeTLS(ln, "", "") + } else { + errc <- srv.Serve(ln) + } + }() + log.Info("listening", "addr", ln.Addr().String(), "tls", tlsCfg != nil) select { case err := <-errc: diff --git a/internal/testutil/pgtest/pgtest.go b/internal/testutil/pgtest/pgtest.go new file mode 100644 index 0000000..e36b24b --- /dev/null +++ b/internal/testutil/pgtest/pgtest.go @@ -0,0 +1,170 @@ +// Package pgtest gives tests a fresh PostgreSQL database each. +// +// If OPIFICA_TEST_DATABASE_URL is set (as in CI), databases are created on +// that server. Otherwise a throwaway server is started from a local +// PostgreSQL installation (initdb and pg_ctl on PATH or under +// /usr/lib/postgresql/*/bin). If neither is available, tests are skipped, +// unless OPIFICA_TEST_REQUIRE_DB is set: then they fail, so CI can't pass by +// skipping. +// +// Use it from TestMain: +// +// func TestMain(m *testing.M) { os.Exit(pgtest.Run(m)) } +package pgtest + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "net/url" + "os" + "os/exec" + "path/filepath" + "sort" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" +) + +const ( + envURL = "OPIFICA_TEST_DATABASE_URL" + envRequire = "OPIFICA_TEST_REQUIRE_DB" +) + +var ( + mu sync.Mutex + adminURL string // server to create test databases on + skipReason string + stop func() +) + +// Run starts the server if needed, runs the tests and stops it again. +func Run(m *testing.M) int { + setup() + code := m.Run() + if stop != nil { + stop() + } + return code +} + +func setup() { + if u := os.Getenv(envURL); u != "" { + adminURL = u + return + } + bin, err := findBinaries() + if err != nil { + skipReason = fmt.Sprintf("no database: set %s or install PostgreSQL (%v)", envURL, err) + return + } + u, s, err := startServer(bin) + if err != nil { + skipReason = fmt.Sprintf("starting a temporary PostgreSQL: %v", err) + return + } + adminURL, stop = u, s +} + +// New returns the URL of a new, empty database, dropped when the test ends. +func New(t testing.TB) string { + t.Helper() + mu.Lock() + base, reason := adminURL, skipReason + mu.Unlock() + if base == "" { + if reason == "" { + reason = "pgtest.Run was not called from TestMain" + } + if os.Getenv(envRequire) != "" { + t.Fatal(reason) + } + t.Skip(reason) + } + + b := make([]byte, 6) + rand.Read(b) + name := "opifica_test_" + hex.EncodeToString(b) + + ctx := context.Background() + conn, err := pgx.Connect(ctx, base) + if err != nil { + t.Fatalf("connecting to test server: %v", err) + } + defer conn.Close(ctx) + if _, err := conn.Exec(ctx, "CREATE DATABASE "+name); err != nil { + t.Fatalf("creating test database: %v", err) + } + t.Cleanup(func() { + conn, err := pgx.Connect(ctx, base) + if err != nil { + return + } + defer conn.Close(ctx) + conn.Exec(ctx, "DROP DATABASE IF EXISTS "+name+" WITH (FORCE)") + }) + + u, err := url.Parse(base) + if err != nil { + t.Fatalf("parsing %s: %v", envURL, err) + } + u.Path = "/" + name + return u.String() +} + +func findBinaries() (string, error) { + if p, err := exec.LookPath("initdb"); err == nil { + return filepath.Dir(p), nil + } + dirs, _ := filepath.Glob("/usr/lib/postgresql/*/bin") + sort.Strings(dirs) + for i := len(dirs) - 1; i >= 0; i-- { // newest version first + if _, err := os.Stat(filepath.Join(dirs[i], "initdb")); err == nil { + return dirs[i], nil + } + } + return "", fmt.Errorf("initdb not found") +} + +// startServer runs a private server listening only on a Unix socket in a +// temporary directory. +func startServer(bin string) (string, func(), error) { + dir, err := os.MkdirTemp("", "opifica-pgtest-") + if err != nil { + return "", nil, err + } + data := filepath.Join(dir, "data") + cleanup := func() { os.RemoveAll(dir) } + + out, err := exec.Command(filepath.Join(bin, "initdb"), "-D", data, "-U", "postgres", + "--auth=trust", "--encoding=UTF8", "--no-sync").CombinedOutput() + if err != nil { + cleanup() + return "", nil, fmt.Errorf("initdb: %v: %s", err, out) + } + opts := fmt.Sprintf("-k %s -c listen_addresses='' -c fsync=off -c synchronous_commit=off -c full_page_writes=off", dir) + out, err = exec.Command(filepath.Join(bin, "pg_ctl"), "-D", data, "-o", opts, + "-l", filepath.Join(dir, "log"), "-w", "-t", "30", "start").CombinedOutput() + if err != nil { + cleanup() + return "", nil, fmt.Errorf("pg_ctl start: %v: %s", err, out) + } + stopFn := func() { + exec.Command(filepath.Join(bin, "pg_ctl"), "-D", data, "-m", "immediate", "stop").Run() + cleanup() + } + + u := "postgres://postgres@/postgres?host=" + url.QueryEscape(dir) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + conn, err := pgx.Connect(ctx, u) + if err != nil { + stopFn() + return "", nil, err + } + conn.Close(ctx) + return u, stopFn, nil +} diff --git a/pkg/worker/tls.go b/pkg/worker/tls.go new file mode 100644 index 0000000..fb7b02c --- /dev/null +++ b/pkg/worker/tls.go @@ -0,0 +1,64 @@ +package worker + +import ( + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "errors" + "fmt" + "strings" +) + +// PinnedCAConfig returns a TLS client config for a worker's first contact +// with the controller, before it has the CA certificate. The controller is +// accepted only if it presents a chain ending at a CA whose public key has +// the given fingerprint ("sha256:", printed by `controller token +// create`), and whose leaf certificate is valid for host. +// +// After joining, the worker uses the CA bundle from the Join response +// instead. +func PinnedCAConfig(fingerprint, host string) (*tls.Config, error) { + want, ok := strings.CutPrefix(fingerprint, "sha256:") + if !ok || len(want) != sha256.Size*2 { + return nil, fmt.Errorf("CA fingerprint must be sha256:<64 hex characters>, got %q", fingerprint) + } + want = strings.ToLower(want) + return &tls.Config{ + MinVersion: tls.VersionTLS13, + // Standard verification needs the CA, which the worker doesn't have + // yet. VerifyConnection does the full check against the pinned CA. + InsecureSkipVerify: true, + VerifyConnection: func(cs tls.ConnectionState) error { + return verifyPinned(cs.PeerCertificates, want, host) + }, + }, nil +} + +func verifyPinned(chain []*x509.Certificate, wantHex, host string) error { + if len(chain) < 2 { + return errors.New("controller did not send its CA certificate") + } + leaf := chain[0] + for _, c := range chain[1:] { + sum := sha256.Sum256(c.RawSubjectPublicKeyInfo) + if hex.EncodeToString(sum[:]) != wantHex { + continue + } + if !c.IsCA { + return errors.New("pinned certificate is not a CA") + } + roots := x509.NewCertPool() + roots.AddCert(c) + _, err := leaf.Verify(x509.VerifyOptions{ + Roots: roots, + DNSName: host, + KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + }) + if err != nil { + return fmt.Errorf("controller certificate: %w", err) + } + return nil + } + return errors.New("controller's CA does not match the pinned fingerprint") +} diff --git a/pkg/worker/tls_test.go b/pkg/worker/tls_test.go new file mode 100644 index 0000000..aaae675 --- /dev/null +++ b/pkg/worker/tls_test.go @@ -0,0 +1,97 @@ +package worker_test + +import ( + "crypto/tls" + "io" + "strings" + "testing" + + "github.com/opifica/opifica/internal/controller/ca" + "github.com/opifica/opifica/pkg/worker" +) + +// serve starts a TLS server with a certificate from authority for hosts, +// sending the CA in the chain like the controller does. +func serve(t *testing.T, authority *ca.CA, hosts []string, sendCA bool) string { + t.Helper() + certPEM, keyPEM, err := authority.IssueServerCert(hosts) + if err != nil { + t.Fatal(err) + } + if sendCA { + certPEM = append(certPEM, authority.CertPEM()...) + } + pair, err := tls.X509KeyPair(certPEM, keyPEM) + if err != nil { + t.Fatal(err) + } + ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{pair}, MinVersion: tls.VersionTLS13}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + go func() { + c.(*tls.Conn).Handshake() + io.Copy(io.Discard, c) + c.Close() + }() + } + }() + return ln.Addr().String() +} + +func dial(t *testing.T, addr, fingerprint, host string) error { + t.Helper() + cfg, err := worker.PinnedCAConfig(fingerprint, host) + if err != nil { + return err + } + c, err := tls.Dial("tcp", addr, cfg) + if err == nil { + c.Close() + } + return err +} + +func TestPinnedCA(t *testing.T) { + good, err := ca.LoadOrCreate(t.TempDir()) + if err != nil { + t.Fatal(err) + } + impostor, err := ca.LoadOrCreate(t.TempDir()) + if err != nil { + t.Fatal(err) + } + + addr := serve(t, good, []string{"127.0.0.1", "controller.local"}, true) + if err := dial(t, addr, good.Fingerprint(), "127.0.0.1"); err != nil { + t.Errorf("genuine controller rejected: %v", err) + } + + tests := []struct { + name, addr, fingerprint, host, want string + }{ + {"wrong pin", addr, impostor.Fingerprint(), "127.0.0.1", "does not match the pinned fingerprint"}, + {"wrong host", addr, good.Fingerprint(), "evil.example", "controller certificate"}, + {"impostor controller", serve(t, impostor, []string{"127.0.0.1"}, true), good.Fingerprint(), "127.0.0.1", "does not match"}, + {"no CA in chain", serve(t, good, []string{"127.0.0.1"}, false), good.Fingerprint(), "127.0.0.1", "did not send its CA"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := dial(t, tt.addr, tt.fingerprint, tt.host) + if err == nil || !strings.Contains(err.Error(), tt.want) { + t.Errorf("err = %v, want %q", err, tt.want) + } + }) + } + + if _, err := worker.PinnedCAConfig("md5:abc", "x"); err == nil { + t.Error("malformed fingerprint accepted") + } +} diff --git a/sqlc.yaml b/sqlc.yaml new file mode 100644 index 0000000..aec2312 --- /dev/null +++ b/sqlc.yaml @@ -0,0 +1,29 @@ +version: "2" +sql: + - engine: postgresql + schema: internal/controller/store/migrations + queries: internal/controller/store/queries + gen: + go: + package: db + out: internal/controller/store/db + sql_package: pgx/v5 + emit_pointers_for_null_types: true + overrides: + - db_type: uuid + go_type: github.com/google/uuid.UUID + - db_type: uuid + nullable: true + go_type: + import: github.com/google/uuid + type: UUID + pointer: true + - db_type: timestamptz + go_type: time.Time + - db_type: timestamptz + nullable: true + go_type: + type: time.Time + pointer: true + - db_type: jsonb + go_type: encoding/json.RawMessage