diff --git a/README.md b/README.md index 4bcb1a67..85bdab0e 100644 --- a/README.md +++ b/README.md @@ -175,7 +175,7 @@ Supported deployment shapes include virtual machines, dedicated physical machine | Build a compatible project contract | [Project CI standard](docs/PROJECT-STANDARD.md) | | Verify project compliance | [Compliance checklist](docs/COMPLIANCE-CHECKLIST.md) | | Configure upgrades, cleanup, draining, and rebooting | [Host maintenance](docs/HOST-MAINTENANCE.md) | -| Configure host health and external missed-heartbeat detection | [Fleet health monitoring](docs/HEALTH-MONITORING.md) | +| Configure host health and authenticated outbound observation | [Fleet health monitoring](docs/HEALTH-MONITORING.md) and [status reporting](docs/STATUS-REPORTING.md) | | Understand secret storage and injection | [Secrets model](docs/SECRETS.md) | | Use private fleet workers for a public project | [Public projects and private delivery](docs/PUBLIC-PRIVATE-CONFIGURATION.md) | | See planned work | [Roadmap](docs/ROADMAP.md) | diff --git a/controller/Dockerfile b/controller/Dockerfile index 39634c75..e7e943ac 100644 --- a/controller/Dockerfile +++ b/controller/Dockerfile @@ -12,7 +12,9 @@ FROM debian:13.6-slim ARG CI_FLEET_COMMIT=unknown LABEL org.opencontainers.image.revision="${CI_FLEET_COMMIT}" \ io.randomdevelopment.ci-fleet.managed="true" -RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates && rm -rf /var/lib/apt/lists/* +RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates \ + && install -d -o 65532 -g 65532 /run/ci-fleet \ + && rm -rf /var/lib/apt/lists/* COPY --from=build /out/ci-fleet-controller /usr/local/bin/ci-fleet-controller USER 65532:65532 ENTRYPOINT ["/usr/local/bin/ci-fleet-controller"] diff --git a/controller/config.go b/controller/config.go index 2515894c..9e5cefca 100644 --- a/controller/config.go +++ b/controller/config.go @@ -27,6 +27,7 @@ type Config struct { RunnerMemory int64 DockerGID string RunnerTTL time.Duration + StatusFile string } func configFromEnv() (Config, error) { @@ -85,6 +86,7 @@ func configFromEnv() (Config, error) { RunnerMemory: runnerMemoryMiB * 1024 * 1024, DockerGID: os.Getenv("CI_FLEET_DOCKER_GID"), RunnerTTL: runnerTTL, + StatusFile: getenv("CI_FLEET_STATUS_FILE", "/run/ci-fleet/status.json"), } return cfg, cfg.Validate() } diff --git a/controller/main.go b/controller/main.go index 0d93d1d4..8919eb18 100644 --- a/controller/main.go +++ b/controller/main.go @@ -10,6 +10,7 @@ import ( "os" "os/signal" "syscall" + "time" "github.com/actions/scaleset" "github.com/actions/scaleset/listener" @@ -74,6 +75,8 @@ func run(ctx context.Context) error { scaler := &Scaler{runners: newRunnerState(), dockerClient: docker, scalesetClient: client, logger: logger, config: cfg, scaleSetID: set.ID} if err := scaler.recoverStale(ctx); err != nil { return err } + scaler.writeStatus() + go scaler.publishStatus(ctx, time.Minute) defer scaler.shutdown(context.WithoutCancel(ctx)) hostname, err := os.Hostname() if err != nil { return fmt.Errorf("get hostname: %w", err) } diff --git a/controller/scaler.go b/controller/scaler.go index e5320e51..53707904 100644 --- a/controller/scaler.go +++ b/controller/scaler.go @@ -15,6 +15,7 @@ import ( "github.com/docker/docker/api/types/container" "github.com/docker/docker/api/types/filters" dockerclient "github.com/docker/docker/client" + "github.com/docker/docker/errdefs" "github.com/google/uuid" ) @@ -30,6 +31,7 @@ type Scaler struct { } func (s *Scaler) HandleDesiredRunnerCount(ctx context.Context, count int) (int, error) { + defer s.writeStatus() current := s.runners.count() target := min(s.config.MaxRunners, s.config.MinRunners+count) for i := current; i < target; i++ { @@ -44,16 +46,19 @@ func (s *Scaler) HandleJobStarted(_ context.Context, job *scaleset.JobStarted) e if !s.runners.markBusy(job.RunnerName) { return fmt.Errorf("job started for unknown runner %q", job.RunnerName) } + s.writeStatus() s.logger.Info("job started", "runner", job.RunnerName, "jobID", job.JobID) return nil } func (s *Scaler) HandleJobCompleted(ctx context.Context, job *scaleset.JobCompleted) error { - id, ok := s.runners.markDone(job.RunnerName) + id, cleanup, ok := s.runners.markDone(job.RunnerName) if !ok { return fmt.Errorf("job completed for unknown runner %q", job.RunnerName) } + s.writeStatus() s.logger.Info("job completed", "runner", job.RunnerName, "jobID", job.JobID) + if !cleanup { return nil } return s.logAndRemove(ctx, job.RunnerName, id) } @@ -92,10 +97,50 @@ func (s *Scaler) startRunner(ctx context.Context) (string, error) { return "", fmt.Errorf("start runner container: %w", err) } s.runners.addIdle(name, created.ID) + go s.watchRunner(context.WithoutCancel(ctx), name, created.ID) s.logger.Info("runner started", "runner", name, "containerID", created.ID) return name, nil } +func (s *Scaler) watchRunner(ctx context.Context, name, id string) { +waitLoop: + for { + stopped, errors := s.dockerClient.ContainerWait(ctx, id, container.WaitConditionNotRunning) + select { + case err, ok := <-errors: + if !s.runners.contains(name, id) { return } + if ok && errdefs.IsNotFound(err) { break waitLoop } + if ok { s.logger.Warn("watch runner container", "runner", name, "error", err) } + time.Sleep(time.Minute) + case response, ok := <-stopped: + if !ok { + if !s.runners.contains(name, id) { return } + time.Sleep(time.Minute) + continue + } + if response.Error != nil { + if !s.runners.contains(name, id) { return } + s.logger.Warn("watch runner container", "runner", name, "error", response.Error.Message) + time.Sleep(time.Minute) + continue + } + break waitLoop + } + } + if !s.runners.markExited(name, id) { return } + s.writeStatus() + s.logger.Warn("runner container exited before job completion", "runner", name) + for { + if err := s.logAndRemove(context.WithoutCancel(ctx), name, id); err == nil { + time.AfterFunc(10*time.Minute, func() { s.runners.forgetExited(name, id) }) + return + } else { + s.logger.Warn("retry exited runner cleanup", "runner", name, "error", err) + } + time.Sleep(time.Minute) + } +} + func (s *Scaler) recoverStale(ctx context.Context) error { f := filters.NewArgs( filters.Arg("label", labelPrefix+"managed=true"), @@ -122,13 +167,14 @@ func (s *Scaler) logAndRemove(ctx context.Context, name, id string) error { } else { s.logger.Warn("could not collect runner logs", "runner", name, "error", err) } - if err := s.dockerClient.ContainerRemove(ctx, id, container.RemoveOptions{Force: true, RemoveVolumes: true}); err != nil { + if err := s.dockerClient.ContainerRemove(ctx, id, container.RemoveOptions{Force: true, RemoveVolumes: true}); err != nil && !errdefs.IsNotFound(err) { return fmt.Errorf("remove runner %s: %w", name, err) } return nil } func (s *Scaler) shutdown(ctx context.Context) { + defer s.writeStatus() for name, id := range s.runners.drain() { if err := s.logAndRemove(ctx, name, id); err != nil { s.logger.Error("runner shutdown failed", slog.String("runner", name), slog.String("error", err.Error())) diff --git a/controller/state.go b/controller/state.go index 8986fd29..642299c4 100644 --- a/controller/state.go +++ b/controller/state.go @@ -3,19 +3,25 @@ package main import "sync" type runnerState struct { - mu sync.Mutex - idle map[string]string - busy map[string]string + mu sync.Mutex + idle map[string]string + busy map[string]string + exited map[string]string } func newRunnerState() runnerState { - return runnerState{idle: make(map[string]string), busy: make(map[string]string)} + return runnerState{idle: make(map[string]string), busy: make(map[string]string), exited: make(map[string]string)} } func (r *runnerState) count() int { + current, _ := r.counts() + return current +} + +func (r *runnerState) counts() (int, int) { r.mu.Lock() defer r.mu.Unlock() - return len(r.idle) + len(r.busy) + return len(r.idle) + len(r.busy), len(r.busy) } func (r *runnerState) addIdle(name, id string) { @@ -29,34 +35,69 @@ func (r *runnerState) markBusy(name string) bool { defer r.mu.Unlock() id, ok := r.idle[name] if !ok { - return false + _, ok = r.exited[name] + return ok } delete(r.idle, name) r.busy[name] = id return true } -func (r *runnerState) markDone(name string) (string, bool) { +func (r *runnerState) contains(name, expectedID string) bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.idle[name] == expectedID || r.busy[name] == expectedID || r.exited[name] == expectedID +} + +func (r *runnerState) markExited(name, expectedID string) bool { + r.mu.Lock() + defer r.mu.Unlock() + if id, ok := r.busy[name]; ok && id == expectedID { + delete(r.busy, name) + r.exited[name] = id + return true + } + if id, ok := r.idle[name]; ok && id == expectedID { + delete(r.idle, name) + r.exited[name] = id + return true + } + return false +} + +func (r *runnerState) forgetExited(name, expectedID string) { + r.mu.Lock() + defer r.mu.Unlock() + if r.exited[name] == expectedID { delete(r.exited, name) } +} + +func (r *runnerState) markDone(name string) (string, bool, bool) { r.mu.Lock() defer r.mu.Unlock() if id, ok := r.busy[name]; ok { delete(r.busy, name) - return id, true + return id, true, true } if id, ok := r.idle[name]; ok { delete(r.idle, name) - return id, true + return id, true, true + } + if id, ok := r.exited[name]; ok { + delete(r.exited, name) + return id, false, true } - return "", false + return "", false, false } func (r *runnerState) drain() map[string]string { r.mu.Lock() defer r.mu.Unlock() - all := make(map[string]string, len(r.idle)+len(r.busy)) + all := make(map[string]string, len(r.idle)+len(r.busy)+len(r.exited)) for name, id := range r.idle { all[name] = id } for name, id := range r.busy { all[name] = id } + for name, id := range r.exited { all[name] = id } clear(r.idle) clear(r.busy) + clear(r.exited) return all } diff --git a/controller/status.go b/controller/status.go new file mode 100644 index 00000000..d63e7f16 --- /dev/null +++ b/controller/status.go @@ -0,0 +1,67 @@ +package main + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "sync" + "time" +) + +var statusWriteMu sync.Mutex +var encodeControllerStatus = func(file *os.File, value controllerStatus) error { + return json.NewEncoder(file).Encode(value) +} + +type controllerStatus struct { + Controller string `json:"controller"` + SoftwareVersion string `json:"software_version"` + Current int `json:"current"` + Busy int `json:"busy"` + Maximum int `json:"maximum"` + GeneratedAt int64 `json:"generated_at"` +} + +func (s *Scaler) writeStatus() { + statusWriteMu.Lock() + defer statusWriteMu.Unlock() + current, busy := s.runners.counts() + softwareVersion := commitSHA + if softwareVersion == "unknown" { softwareVersion = version } + value := controllerStatus{ + Controller: s.config.FleetInstance, SoftwareVersion: softwareVersion, + Current: current, Busy: busy, Maximum: s.config.MaxRunners, GeneratedAt: time.Now().Unix(), + } + directory := filepath.Dir(s.config.StatusFile) + if err := os.MkdirAll(directory, 0o755); err != nil { + s.logger.Warn("write controller status", "error", err) + return + } + temporary, err := os.CreateTemp(directory, ".status-*") + if err != nil { + s.logger.Warn("write controller status", "error", err) + return + } + name := temporary.Name() + defer os.Remove(name) + if err = temporary.Chmod(0o644); err == nil { + err = encodeControllerStatus(temporary, value) + } + if closeErr := temporary.Close(); err == nil { err = closeErr } + if err == nil { err = os.Rename(name, s.config.StatusFile) } + if err != nil { s.logger.Warn("write controller status", "error", err) } +} + +func (s *Scaler) publishStatus(ctx context.Context, interval time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.writeStatus() + } + } +} diff --git a/controller/status_test.go b/controller/status_test.go new file mode 100644 index 00000000..8ed3bd59 --- /dev/null +++ b/controller/status_test.go @@ -0,0 +1,138 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "log/slog" + "os" + "path/filepath" + "testing" + "time" +) + +func TestRunnerStateMakesExitAndCompletionIdempotent(t *testing.T) { + state := newRunnerState() + state.addIdle("runner", "container-1") + if state.markExited("runner", "container-2") { t.Fatal("removed replacement runner for stale exit") } + if current, _ := state.counts(); current != 1 { t.Fatalf("current=%d, want 1", current) } + if !state.markExited("runner", "container-1") { t.Fatal("matching exited runner was not removed") } + if current, _ := state.counts(); current != 0 { t.Fatalf("current=%d, want 0", current) } + if !state.markBusy("runner") { t.Fatal("late job start rejected exited runner") } + if current, busy := state.counts(); current != 0 || busy != 0 { t.Fatalf("late start restored exited runner: current=%d busy=%d", current, busy) } + if id, cleanup, ok := state.markDone("runner"); !ok || cleanup || id != "container-1" { + t.Fatalf("late completion = id %q cleanup %t ok %t", id, cleanup, ok) + } + + state.addIdle("normal", "container-2") + if !state.contains("normal", "container-2") { t.Fatal("tracked runner was not found") } + if _, cleanup, ok := state.markDone("normal"); !ok || !cleanup { t.Fatal("normal completion skipped cleanup") } + if state.contains("normal", "container-2") { t.Fatal("completed runner remained tracked") } + if state.markExited("normal", "container-2") { t.Fatal("exit won after normal completion") } +} + +func TestWriteStatusReportsRunnerCountsWithoutControllingExecution(t *testing.T) { + path := filepath.Join(t.TempDir(), "status.json") + scaler := &Scaler{ + runners: newRunnerState(), + logger: slog.New(slog.NewTextHandler(os.Stderr, nil)), + config: Config{FleetInstance: "example-ci-01", MaxRunners: 6, StatusFile: path}, + } + scaler.runners.addIdle("idle", "1") + scaler.runners.addIdle("busy", "2") + if !scaler.runners.markBusy("busy") { + t.Fatal("runner did not become busy") + } + scaler.writeStatus() + var got controllerStatus + body, err := os.ReadFile(path) + if err != nil { t.Fatal(err) } + if err := json.Unmarshal(body, &got); err != nil { t.Fatal(err) } + if got.Controller != "example-ci-01" || got.Current != 2 || got.Busy != 1 || got.Maximum != 6 { + t.Fatalf("unexpected status: %+v", got) + } + + // Reporting is advisory: an unwritable destination has no return path into scaling. + scaler.config.StatusFile = filepath.Join(path, "impossible") + scaler.writeStatus() + if current, busy := scaler.runners.counts(); current != 2 || busy != 1 { + t.Fatalf("status failure changed runner state: current=%d busy=%d", current, busy) + } +} + +func TestWriteStatusPreservesPreviousSnapshotOnEncodingFailure(t *testing.T) { + path := filepath.Join(t.TempDir(), "status.json") + previous := []byte(`{"previous":true}`) + if err := os.WriteFile(path, previous, 0o644); err != nil { t.Fatal(err) } + scaler := &Scaler{ + runners: newRunnerState(), + logger: slog.New(slog.NewTextHandler(os.Stderr, nil)), + config: Config{FleetInstance: "example-ci-01", MaxRunners: 1, StatusFile: path}, + } + original := encodeControllerStatus + encodeControllerStatus = func(file *os.File, _ controllerStatus) error { + _, _ = file.WriteString("truncated") + return errors.New("filesystem full") + } + defer func() { encodeControllerStatus = original }() + scaler.writeStatus() + got, err := os.ReadFile(path) + if err != nil { t.Fatal(err) } + if string(got) != string(previous) { t.Fatalf("status replaced after encoding failure: %q", got) } +} + +func TestStatusWritesUsePublicationLock(t *testing.T) { + scaler := &Scaler{ + runners: newRunnerState(), + logger: slog.New(slog.NewTextHandler(os.Stderr, nil)), + config: Config{FleetInstance: "example-ci-01", MaxRunners: 1, StatusFile: filepath.Join(t.TempDir(), "status.json")}, + } + statusWriteMu.Lock() + done := make(chan struct{}) + go func() { scaler.writeStatus(); close(done) }() + select { + case <-done: + statusWriteMu.Unlock() + t.Fatal("status write bypassed publication lock") + case <-time.After(20 * time.Millisecond): + statusWriteMu.Unlock() + } + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("status write did not resume after publication lock") + } +} + +func TestStatusPublisherRefreshesIdleSnapshot(t *testing.T) { + path := filepath.Join(t.TempDir(), "status.json") + scaler := &Scaler{ + runners: newRunnerState(), + logger: slog.New(slog.NewTextHandler(os.Stderr, nil)), + config: Config{FleetInstance: "example-ci-01", MaxRunners: 1, StatusFile: path}, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + go func() { + scaler.publishStatus(ctx, time.Millisecond) + close(done) + }() + deadline := time.After(time.Second) + for { + if _, err := os.Stat(path); err == nil { + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("idle status publisher did not stop") + } + return + } + select { + case <-deadline: + t.Fatal("idle status publisher did not refresh snapshot") + case <-time.After(time.Millisecond): + } + } +} diff --git a/docs/HEALTH-MONITORING.md b/docs/HEALTH-MONITORING.md index b9549dd2..5791750f 100644 --- a/docs/HEALTH-MONITORING.md +++ b/docs/HEALTH-MONITORING.md @@ -21,7 +21,7 @@ The check covers: - cleanup, drift, health, and update services/timers; - failed package state, pending reboot, and clock synchronization; - an optional host-local backup check; -- optional outbound heartbeat delivery. +- optional authenticated outbound status delivery. It reports but never prunes, restarts, or repairs resources. Project source, logs, environment values, tokens, and private keys are never included. @@ -44,25 +44,19 @@ CI_FLEET_HEALTH_LOAD_WARN_PER_CPU=1.0 CI_FLEET_HEALTH_LOAD_CRITICAL_PER_CPU=1.5 CI_FLEET_HEALTH_RESTART_WARN_COUNT=3 CI_FLEET_HEALTH_BACKUP_CHECK=/usr/local/sbin/ci-fleet-backup-check -CI_FLEET_HEALTH_HEARTBEAT_URL=https://monitor.example.invalid/heartbeat -CI_FLEET_HEALTH_HEARTBEAT_TOKEN_FILE=/etc/ci-fleet/secrets/heartbeat-token +CI_FLEET_HEALTH_STATUS_URL=https://status.example.invalid/v1/status +CI_FLEET_HEALTH_STATUS_KEY_FILE=/etc/ci-fleet/secrets/status-reporting.key ``` -The backup hook must be an absolute, executable, root-owned file that is not group- or world-writable. Its output is discarded; only its exit status is reported. The heartbeat URL must use HTTPS. An optional token file must be root-owned and inaccessible to group/other users. The installer never creates, prints, commits, or removes this host-local file or its credentials, so rollback preserves them. +The backup hook must be an absolute, executable, root-owned file that is not group- or world-writable. Its output is discarded; only its exit status is reported. Status delivery requires HTTPS and a unique 32-128 byte root-owned key file with mode `0600`. The installer never creates, prints, commits, or removes host-local monitoring credentials, so rollback preserves them. Delivery failures are warnings and never block runners or reconciliation. -## External missed-heartbeat detection +See [authenticated controller status reporting](STATUS-REPORTING.md) for the v1 schema, request authentication, receiver, retention, API, and threat model. -A receiver accepts the redacted JSON POST and stores the most recent body as `.json`. Receiver implementation, endpoint, credential, address, and alert destination are provider-local. The external monitor evaluates those files against reviewed desired state: +## External missed-report detection -```bash -python3 scripts/health.py heartbeats \ - --config /srv/rd-delivery-config/fleet.json \ - --input-dir /var/lib/ci-fleet-heartbeats \ - --grace-seconds 900 \ - --json -``` +The authenticated receiver stores each controller's `generated_at` and returns it through the read-only API. An external monitor compares the latest report with reviewed desired controller inventory and treats an active controller with no report inside the grace period as unhealthy. Drained or disabled lifecycle state remains a desired-state decision, not something an absent controller can assert. -An active host with no fresh record is unhealthy. A drained host reports maintenance without a false alarm; a disabled host reports retired. A monitoring outage therefore cannot silently turn missing hosts healthy. +The legacy file-based `health.py heartbeats` evaluator remains available for existing integrations, but new deployments should consume the authenticated API described in [STATUS-REPORTING.md](STATUS-REPORTING.md). During upgrade, a configured legacy heartbeat continues until `CI_FLEET_HEALTH_STATUS_URL` is present; provision and verify the authenticated receiver before removing the legacy settings. ## Operations diff --git a/docs/README.md b/docs/README.md index 1f934d46..bf5d1e60 100644 --- a/docs/README.md +++ b/docs/README.md @@ -21,7 +21,7 @@ New operator? Follow the [Quickstart](QUICKSTART.md): what ci-fleet does, instal | Make a project compliant | [Project CI standard](PROJECT-STANDARD.md) and [compliance checklist](COMPLIANCE-CHECKLIST.md) | | Split tests across parallel workers | [Project CI standard](PROJECT-STANDARD.md) and the [parallel workflow example](../examples/workflows/parallel-ci.yml.example) | | Configure automatic updates and cleanup | [Host maintenance](HOST-MAINTENANCE.md) | -| Monitor hosts and detect missed heartbeats | [Fleet health monitoring](HEALTH-MONITORING.md) | +| Monitor hosts and detect missed reports | [Fleet health monitoring](HEALTH-MONITORING.md) and [authenticated status reporting](STATUS-REPORTING.md) | | Handle GitHub App, workflow, or deployment secrets | [Secrets model](SECRETS.md) and [security policy](../SECURITY.md) | | Review accepted implementation scope | [Design decisions](DESIGN-DECISIONS.md) | | Run private CI or deployment for a public project | [Public projects, private delivery, and private configuration](PUBLIC-PRIVATE-CONFIGURATION.md) | @@ -61,6 +61,7 @@ These pages are normative for compatible projects and hosts: - [Compliance checklist](COMPLIANCE-CHECKLIST.md) - [Host maintenance standard](HOST-MAINTENANCE.md) - [Fleet health monitoring](HEALTH-MONITORING.md) +- [Authenticated controller status reporting](STATUS-REPORTING.md) - [Git-authored controller desired state](DESIRED-STATE.md) - [Secrets model](SECRETS.md) - [Security policy](../SECURITY.md) diff --git a/docs/STATUS-REPORTING.md b/docs/STATUS-REPORTING.md new file mode 100644 index 00000000..c37a530c --- /dev/null +++ b/docs/STATUS-REPORTING.md @@ -0,0 +1,123 @@ +# Authenticated controller status reporting + +The status channel extends the local health collector with a redacted, versioned report for routine remote observation. It is backend infrastructure for a future read-only console. It does not expose a controller listener or permit host actions. + +## Flow + +1. The existing root-owned health timer collects local state. It reads exact runner counts from a small status file inside the controller container with `docker exec`; no additional service receives the Docker socket. +2. The reporter serializes `schemas/status-report-v1.json`, signs the request with that controller's independent key, and sends `POST /v1/status` over HTTPS. +3. A receiver bound to loopback validates identity, signature, freshness, nonce, size, schema, and ordering before storing the report. +4. A TLS reverse proxy exposes the receiver. The receiver itself refuses a non-loopback bind. +5. A later console can use the authenticated read-only API. Mutation endpoints do not exist. + +The five-minute health timer is the normal submission schedule. The receiver rejects submissions less than 30 seconds apart and bodies larger than 32 KiB. + +## Version 1 contract + +The machine-readable contract is `schemas/status-report-v1.json`. It reports: + +- controller ID, controller build/engine version, host boot time, and SSH state; +- desired and applied configuration commits; +- reconciliation state and last successful reconciliation time; +- drift state; +- controller process state and restart count; +- reconciliation, drift, health, and cleanup timer states; +- current, busy, and configured-maximum runner counts; +- CPU use, logical CPU count, memory, swap, root/Docker disk and inode use, and 1/5/15-minute load; +- Docker availability and OOM evidence; +- one controlled error code/message, report generation time, and schema version. + +All times are Unix seconds. Commit values are empty when unavailable. Receiver validation rejects unknown fields and unsupported schema versions rather than guessing at compatibility. + +`error.message` is derived only from a controlled error code (`_` becomes a space). Raw exception text is never transmitted. + +## Authentication + +Each controller receives a unique random HMAC key. A key must contain 32-128 bytes and be owned by root with mode `0600` on the controller. The receiver keeps a separate copy owned by its service account with mode `0600`. Key files are read byte-for-byte, including leading or trailing whitespace; copy the same raw bytes to both sides. + +Every report includes: + +- `X-CI-Fleet-Controller`; +- `X-CI-Fleet-Timestamp`; +- a 128-bit random `X-CI-Fleet-Nonce`; +- an `Authorization` header using the `CI-Fleet-HMAC-SHA256` scheme. + +The signature covers method, fixed path, controller ID, timestamp, nonce, and SHA-256 body digest. It is valid only within five minutes. The receiver selects the key from the claimed controller ID, requires the signed ID to equal the payload ID, and records nonces until their authentication window expires. A controller therefore cannot sign as another controller unless that controller's independent key is compromised. + +Rotate a controller key by replacing the receiver-side copy, restarting the receiver so it reloads the `0600` auth configuration, and then replacing the controller-side copy within one reporting interval. Keys are not GitHub credentials and must not be committed to desired state. + +## Receiver + +Example service-account-owned configuration (`0600`): + +```json +{ + "controllers": { + "example-ci-01": "secrets/example-ci-01.key" + }, + "read_token_file": "secrets/read-api.token" +} +``` + +Start the stdlib receiver on loopback behind an HTTPS reverse proxy: + +```bash +python3 scripts/status_receiver.py \ + --auth-config /etc/ci-fleet-status/auth.json \ + --database /var/lib/ci-fleet-status/status.db \ + --bind 127.0.0.1 \ + --port 8080 +``` + +The database is mode `0600`. Defaults retain at most 288 reports per controller and no report older than seven days; whichever bound is reached first wins. At five-minute intervals, the count bound is approximately one day. Nonces are retained only for the request-authentication window. + +Read endpoints require `Authorization: Bearer `: + +- `GET /v1/controllers` returns the latest report for each controller; +- `GET /v1/controllers/` returns latest plus bounded history. + +`POST /v1/status` is the only write endpoint. There are no endpoints for configuration, shell execution, logs, Docker actions, runner actions, or arbitrary host operations. + +Controller-side configuration in `/etc/ci-fleet/monitoring.env` is: + +```text +CI_FLEET_HEALTH_STATUS_URL=https://status.example.invalid/v1/status +CI_FLEET_HEALTH_STATUS_KEY_FILE=/etc/ci-fleet/secrets/status-reporting.key +``` + +The URL must be HTTPS with the exact `/v1/status` path and no embedded credentials, query, or fragment. + +## Threat model + +Protected assets are controller identity, status integrity, status confidentiality, controller credentials, private desired state, and runner/reconciliation availability. + +Covered threats: + +- network modification or forgery: HTTPS plus body-bound HMAC; +- replay and stale overwrite: timestamp window, nonce uniqueness, and monotonically increasing report time; +- controller impersonation: independent keys and header/payload identity equality; +- storage abuse: strict schema, 32 KiB request cap, rate limit, count retention, and time retention; +- secret leakage: allowlisted fields and controlled error strings only; +- receiver exposure: loopback-only application bind, authenticated reads, and external HTTPS termination; +- monitoring dependency failure: delivery failure is a local warning and never blocks runner handling or reconciliation. + +Residual risks: + +- compromise of one controller exposes that controller's reporting key and permits forged reports for that identity until rotation; +- ordinary runner jobs are already host-root-equivalent through the Docker socket and can read any host-local reporting key; this channel prevents cross-controller impersonation but does not attest a controller against malicious code already running on that controller. Isolate job Docker onto a separate trust boundary before treating reports as adversarial to job code; +- compromise of the receiver exposes retained status and all receiver-side reporting keys; +- HMAC keys are symmetric; use a managed asymmetric identity service later if receiver compromise becomes part of the impersonation threat model; +- the read bearer token is suitable for the backend foundation, not browser distribution. A future console should terminate user authentication before this API. + +## Deliberately excluded + +Reports never contain: + +- GitHub App keys, installation tokens, reporting keys, read tokens, or other credentials; +- environment names or values; +- private desired-state contents or repository contents; +- command lines, process arguments, arbitrary logs, job output, or exception text; +- project source, runner registration material, network addresses, or provider inventory; +- Docker socket access or any host-control capability. + +The local full health result remains available for recovery, but only the status schema's allowlisted summary leaves the controller. diff --git a/schemas/status-report-v1.json b/schemas/status-report-v1.json new file mode 100644 index 00000000..d5a12ec6 --- /dev/null +++ b/schemas/status-report-v1.json @@ -0,0 +1,139 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/RandomDevelopment/ci-fleet/blob/main/schemas/status-report-v1.json", + "title": "ci-fleet controller status report v1", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "controller", "configuration", "reconciliation", "drift", "process", "timers", "runners", "metrics", "docker", "error", "generated_at"], + "properties": { + "schema_version": {"const": 1}, + "controller": { + "type": "object", "additionalProperties": false, + "required": ["id", "software_version", "boot_time", "ssh"], + "properties": { + "id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$"}, + "software_version": {"type": "string", "pattern": "^[A-Za-z0-9_.+-]{1,64}$"}, + "boot_time": {"type": "integer", "minimum": 0}, + "ssh": {"enum": ["enabled", "disabled", "unknown"]} + } + }, + "configuration": { + "type": "object", "additionalProperties": false, + "required": ["desired_commit", "applied_commit"], + "properties": { + "desired_commit": {"$ref": "#/$defs/commit"}, + "applied_commit": {"$ref": "#/$defs/commit"} + } + }, + "reconciliation": { + "type": "object", "additionalProperties": false, + "required": ["state", "last_success_at"], + "properties": { + "state": {"enum": ["bootstrap", "converged", "drift", "failed", "invalid", "missing", "pending", "reconciling", "rolled_back", "unknown"]}, + "last_success_at": {"type": ["integer", "null"], "minimum": 0} + } + }, + "drift": { + "type": "object", "additionalProperties": false, "required": ["state"], + "properties": {"state": {"enum": ["ok", "stale", "failed", "unknown"]}} + }, + "process": { + "type": "object", "additionalProperties": false, "required": ["state", "restart_count"], + "properties": { + "state": {"enum": ["created", "exited", "missing", "paused", "restarting", "running", "unknown"]}, + "restart_count": {"type": "integer", "minimum": 0} + } + }, + "timers": { + "type": "object", "additionalProperties": false, + "required": ["reconciliation", "drift", "health", "cleanup"], + "properties": { + "reconciliation": {"$ref": "#/$defs/unitState"}, + "drift": {"$ref": "#/$defs/unitState"}, + "health": {"$ref": "#/$defs/unitState"}, + "cleanup": {"$ref": "#/$defs/unitState"} + } + }, + "runners": { + "type": "object", "additionalProperties": false, + "required": ["current", "busy", "maximum"], + "properties": { + "current": {"type": "integer", "minimum": 0}, + "busy": {"type": "integer", "minimum": 0}, + "maximum": {"type": "integer", "minimum": 0} + } + }, + "metrics": { + "type": "object", "additionalProperties": false, + "required": ["cpu", "memory", "swap", "disk", "inodes", "load"], + "properties": { + "cpu": { + "type": "object", "additionalProperties": false, "required": ["logical", "used_percent"], + "properties": { + "logical": {"type": "integer", "minimum": 1}, + "used_percent": {"type": "number", "minimum": 0, "maximum": 100} + } + }, + "memory": { + "type": "object", "additionalProperties": false, "required": ["total_bytes", "available_bytes"], + "properties": { + "total_bytes": {"type": "integer", "minimum": 0}, + "available_bytes": {"type": "integer", "minimum": 0} + } + }, + "swap": {"$ref": "#/$defs/byteUsage"}, + "disk": { + "type": "object", "additionalProperties": false, "required": ["root", "docker"], + "properties": {"root": {"$ref": "#/$defs/byteUsage"}, "docker": {"$ref": "#/$defs/byteUsage"}} + }, + "inodes": { + "type": "object", "additionalProperties": false, "required": ["root", "docker"], + "properties": {"root": {"$ref": "#/$defs/inodeUsage"}, "docker": {"$ref": "#/$defs/inodeUsage"}} + }, + "load": { + "type": "object", "additionalProperties": false, "required": ["one", "five", "fifteen"], + "properties": { + "one": {"type": "number", "minimum": 0}, + "five": {"type": "number", "minimum": 0}, + "fifteen": {"type": "number", "minimum": 0} + } + } + } + }, + "docker": { + "type": "object", "additionalProperties": false, "required": ["healthy", "oom"], + "properties": {"healthy": {"type": "boolean"}, "oom": {"type": "boolean"}} + }, + "error": { + "oneOf": [ + {"type": "null"}, + { + "type": "object", "additionalProperties": false, "required": ["code", "message"], + "properties": { + "code": {"type": "string", "pattern": "^[a-z0-9_]{1,64}$"}, + "message": {"type": "string", "pattern": "^[a-z0-9 ]{1,64}$"} + } + } + ] + }, + "generated_at": {"type": "integer", "minimum": 0} + }, + "$defs": { + "commit": {"type": "string", "pattern": "^(|[0-9a-f]{40})$"}, + "unitState": {"enum": ["ok", "stale", "failed", "unknown"]}, + "byteUsage": { + "type": "object", "additionalProperties": false, "required": ["total_bytes", "used_bytes"], + "properties": { + "total_bytes": {"type": "integer", "minimum": 0}, + "used_bytes": {"type": "integer", "minimum": 0} + } + }, + "inodeUsage": { + "type": "object", "additionalProperties": false, "required": ["total", "used"], + "properties": { + "total": {"type": "integer", "minimum": 0}, + "used": {"type": "integer", "minimum": 0} + } + } + } +} diff --git a/scripts/health.py b/scripts/health.py index 04f3795a..7f847fbf 100644 --- a/scripts/health.py +++ b/scripts/health.py @@ -2,18 +2,28 @@ from __future__ import annotations import argparse +import http.client import json import os import re +import secrets import stat import subprocess import sys import time +import urllib.parse import urllib.request from dataclasses import dataclass from pathlib import Path from typing import Any, Callable +from status_auth import sign_headers + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, *_args: Any, **_kwargs: Any) -> None: + return None + @dataclass(frozen=True) class Thresholds: @@ -128,6 +138,8 @@ def add(check_id: str, severity: str, **details: Any) -> None: controller_state = snapshot["controller"]["state"] controller_ok = controller_state == "running" if desired == "active" else controller_state in {"missing", "exited", "created"} add("controller", "ok" if controller_ok else "critical", state=controller_state, desired_state=desired) + status_expected = desired == "active" and controller_state == "running" + add("controller_status", "warning" if status_expected and not snapshot.get("controller_status_valid", True) else "ok") restarts = snapshot["controller"]["restart_count"] add("restarts", "warning" if restarts >= thresholds.restart_warn_count else "ok", count=restarts) @@ -181,6 +193,71 @@ def add(check_id: str, severity: str, **details: Any) -> None: } +def build_status_report(snapshot: dict[str, Any], health_report: dict[str, Any], *, generated_at: int) -> dict[str, Any]: + reconciliation = snapshot.get("reconciliation") or {} + state = reconciliation.get("status", "missing") + status_expected = snapshot.get("desired_state") == "active" and snapshot.get("controller", {}).get("state") == "running" + error_code = "health_controller_status" if status_expected and snapshot.get("controller_status_valid") is False else "" + if not error_code: + error_code = f"reconciliation_{state}" if state in {"drift", "failed", "invalid", "rolled_back"} else "" + if not error_code: + checks = health_report.get("checks", []) + failed = next((check for check in checks if check.get("status") == "critical"), None) + failed = failed or next((check for check in checks if check.get("status") == "warning"), None) + error_code = f"health_{failed['id']}" if failed else "" + error = {"code": error_code, "message": error_code.replace("_", " ")} if error_code else None + timers = snapshot.get("timers", {}) + disks = snapshot["disks"] + process_state = snapshot["controller"].get("state", "unknown") + if process_state not in {"created", "exited", "missing", "paused", "restarting", "running"}: + process_state = "unknown" + commits = [reconciliation.get(name, "") for name in ("desired_commit", "applied_commit")] + commits = [commit if isinstance(commit, str) and (not commit or re.fullmatch(r"[0-9a-f]{40}", commit)) else "" for commit in commits] + last_success = reconciliation.get("last_success_at") + if not isinstance(last_success, int) or isinstance(last_success, bool) or last_success > generated_at: + last_success = None + return { + "schema_version": 1, + "controller": { + "id": snapshot["controller_id"], + "software_version": snapshot.get("software_version", "unknown"), + "boot_time": snapshot.get("boot_time", 0), + "ssh": snapshot.get("ssh", "unknown"), + }, + "configuration": { + "desired_commit": commits[0], + "applied_commit": commits[1], + }, + "reconciliation": {"state": state, "last_success_at": last_success}, + "drift": {"state": snapshot.get("services", {}).get("drift", "unknown")}, + "process": { + "state": process_state, + "restart_count": snapshot["controller"].get("restart_count", 0), + }, + "timers": { + "reconciliation": timers.get("reconcile", "unknown"), + "drift": timers.get("drift", "unknown"), + "health": timers.get("health", "unknown"), + "cleanup": timers.get("cleanup", "unknown"), + }, + "runners": snapshot.get("runners", {"current": 0, "busy": 0, "maximum": 0}), + "metrics": { + "cpu": snapshot.get("cpu", {"logical": 1, "used_percent": 0}), + "memory": snapshot.get("memory", {"total_bytes": 0, "available_bytes": 0}), + "swap": snapshot.get("swap", {"total_bytes": 0, "used_bytes": 0}), + "disk": {name: {"total_bytes": value.get("total_bytes", 0), "used_bytes": value.get("used_bytes", 0)} for name, value in disks.items()}, + "inodes": {name: {"total": value.get("inode_total", 0), "used": value.get("inode_used", 0)} for name, value in disks.items()}, + "load": snapshot.get("load", {"one": 0, "five": 0, "fifteen": 0}), + }, + "docker": { + "healthy": bool(snapshot.get("docker_available")), + "oom": bool(snapshot.get("recent_oom") or snapshot["controller"].get("oom_killed")), + }, + "error": error, + "generated_at": generated_at, + } + + Runner = Callable[[list[str]], subprocess.CompletedProcess[str]] @@ -201,6 +278,10 @@ def _disk(path: str) -> dict[str, int]: return { "used_percent": round(100 * used / max(value.f_blocks, 1)), "inode_used_percent": round(100 * iused / max(value.f_files, 1)), + "total_bytes": value.f_blocks * value.f_frsize, + "used_bytes": used * value.f_frsize, + "inode_total": value.f_files, + "inode_used": iused, } @@ -228,6 +309,15 @@ def _timespan_seconds(value: str) -> float | None: return sum(float(match.group(1)) * units[match.group(2)] for match in matches) +def _ssh_state(run: Runner) -> str: + results = [run(["systemctl", action, unit]) for unit in ("ssh.service", "ssh.socket", "sshd.service") for action in ("is-enabled", "is-active")] + states = [result.stdout.strip().lower() for result in results] + if any(result.returncode == 0 and state in {"active", "enabled"} for result, state in zip(results, states)): + return "enabled" + disabled = {"disabled", "inactive", "masked", "not-found", "failed", "static"} + return "disabled" if all(state in disabled for state in states) else "unknown" + + def _unit_state(run: Runner, unit: str, timer: bool = False, max_age_seconds: int = 0) -> str: if timer: if run(["systemctl", "is-active", unit]).returncode != 0: @@ -270,18 +360,70 @@ def _container(run: Runner, name: str) -> tuple[dict[str, Any], dict[str, int]]: }, capacity -def _memory(root: Path) -> tuple[int, int]: +def _memory_details(root: Path) -> tuple[dict[str, int], dict[str, int]]: values: dict[str, int] = {} try: for line in (root / "proc/meminfo").read_text().splitlines(): key, value = line.split(":", 1) - values[key] = int(value.split()[0]) + values[key] = int(value.split()[0]) * 1024 except (OSError, ValueError, IndexError): - return 0, 0 - available = round(100 * values.get("MemAvailable", 0) / max(values.get("MemTotal", 1), 1)) + return {"total_bytes": 0, "available_bytes": 0}, {"total_bytes": 0, "used_bytes": 0} swap_total = values.get("SwapTotal", 0) - swap = round(100 * (swap_total - values.get("SwapFree", 0)) / max(swap_total, 1)) if swap_total else 0 - return available, swap + return ( + {"total_bytes": values.get("MemTotal", 0), "available_bytes": values.get("MemAvailable", 0)}, + {"total_bytes": swap_total, "used_bytes": max(0, swap_total - values.get("SwapFree", 0))}, + ) + + +def _memory(root: Path) -> tuple[int, int]: + memory, swap = _memory_details(root) + available = round(100 * memory["available_bytes"] / max(memory["total_bytes"], 1)) + swap_used = round(100 * swap["used_bytes"] / max(swap["total_bytes"], 1)) if swap["total_bytes"] else 0 + return available, swap_used + + +def _cpu(root: Path) -> dict[str, float | int]: + # ponytail: cumulative boot-average CPU; persist the prior sample if interval utilization becomes necessary. + try: + fields = next(line for line in (root / "proc/stat").read_text().splitlines() if line.startswith("cpu ")).split()[1:] + counters = [int(value) for value in fields] + total = sum(counters[:8]) + used = total - counters[3] + percent = round(100 * used / max(total, 1), 1) + except (OSError, ValueError, IndexError, StopIteration): + percent = 0.0 + return {"logical": max(os.cpu_count() or 1, 1), "used_percent": percent} + + +def _boot_time(root: Path) -> int: + try: + line = next(line for line in (root / "proc/stat").read_text().splitlines() if line.startswith("btime ")) + return max(int(line.split()[1]), 0) + except (OSError, ValueError, IndexError, StopIteration): + return 0 + + +def _controller_status(run: Runner, name: str, controller: str, maximum: int) -> tuple[dict[str, int], str, bool]: + result = run(["docker", "exec", name, "cat", "/run/ci-fleet/status.json"]) + try: + value = json.loads(result.stdout) if result.returncode == 0 else {} + current, busy, reported_max = (value[key] for key in ("current", "busy", "maximum")) + version = value["software_version"] + generated_at = value["generated_at"] + valid = ( + value.get("controller") == controller + and all(isinstance(count, int) and not isinstance(count, bool) and count >= 0 for count in (current, busy, reported_max)) + and busy <= current <= reported_max == maximum + and isinstance(version, str) + and bool(re.fullmatch(r"[A-Za-z0-9_.+-]{1,64}", version)) + and isinstance(generated_at, int) and not isinstance(generated_at, bool) + and abs(int(time.time()) - generated_at) <= 120 + ) + if valid: + return {"current": current, "busy": busy, "maximum": reported_max}, version, True + except (KeyError, TypeError, ValueError, json.JSONDecodeError): + pass + return {"current": 0, "busy": 0, "maximum": maximum}, "unknown", False def _memory_pressure(root: Path) -> float | None: @@ -309,13 +451,14 @@ def _backup_state(values: dict[str, str], run: Runner) -> str: return "ok" if run([str(path)]).returncode == 0 else "failed" -def _reconcile_state(path: Path) -> dict[str, str]: +def _reconcile_state(path: Path) -> dict[str, Any]: + empty = {"status": "missing", "desired_commit": "", "applied_commit": "", "health": "", "last_success_at": None} try: value = json.loads(path.read_text()) except (OSError, json.JSONDecodeError): - return {"status": "missing", "desired_commit": "", "applied_commit": "", "health": ""} + return empty if not isinstance(value, dict): - return {"status": "invalid", "desired_commit": "", "applied_commit": "", "health": ""} + return {**empty, "status": "invalid"} status = value.get("status", "") if not isinstance(status, str) or status not in {"converged", "drift", "invalid", "pending", "reconciling", "rolled_back", "failed"}: status = "invalid" @@ -324,15 +467,23 @@ def _reconcile_state(path: Path) -> dict[str, str]: reported_health = value.get("health", "") if not isinstance(reported_health, str) or reported_health not in {"", "healthy", "warning", "unhealthy", "maintenance", "drift", "unknown"}: reported_health = "invalid" - return {"status": status, "desired_commit": commits[0], "applied_commit": commits[1], "health": reported_health} + last_success = value.get("last_success_at") + if not isinstance(last_success, int) or isinstance(last_success, bool) or last_success < 0: + last_success = None + return {"status": status, "desired_commit": commits[0], "applied_commit": commits[1], "health": reported_health, "last_success_at": last_success} def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Runner = _run) -> dict[str, Any]: docker_root = values.get("CI_FLEET_DOCKER_ROOT", "/var/lib/docker") available, swap = _memory(root) + memory, swap_metrics = _memory_details(root) + loads = os.getloadavg() docker_ok = run(["docker", "info"]).returncode == 0 controller_name = values.get("CI_FLEET_CONTROLLER_CONTAINER", "ci-fleet-controller-1") + instance = values.get("CI_FLEET_INSTANCE", "unknown") + configured = {"min": int(values.get("CI_FLEET_MIN_RUNNERS", 0)), "max": int(values.get("CI_FLEET_MAX_RUNNERS", 0))} controller, effective = _container(run, controller_name) if docker_ok else ({"state": "missing", "restart_count": 0, "oom_killed": False}, {"min": 0, "max": 0}) + runners, software_version, controller_status_valid = _controller_status(run, controller_name, instance, configured["max"]) if docker_ok else ({"current": 0, "busy": 0, "maximum": configured["max"]}, "unknown", False) managed = {"running": 0, "inactive": 0, "unhealthy": 0, "restarting": 0} if docker_ok: result = run(["docker", "ps", "-a", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true", "--format", "{{json .}}"]) @@ -346,7 +497,6 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run managed["unhealthy"] += int("unhealthy" in status) managed["restarting"] += int(state == "restarting" or "restarting" in status) oom = run(["journalctl", "--dmesg", "--since=-24h", "--grep=Out of memory|Killed process", "--quiet"]) - configured = {"min": int(values.get("CI_FLEET_MIN_RUNNERS", 0)), "max": int(values.get("CI_FLEET_MAX_RUNNERS", 0))} timer_ages = {"health": 900, "cleanup": 172800, "drift": 3600} remote_config = bool(re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", values.get("CI_FLEET_CONFIG_REPOSITORY", ""))) reconciliation = _reconcile_state(root / "var/lib/ci-fleet/reconcile/state.json") if remote_config else None @@ -370,7 +520,6 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run # ponytail: activation validates unit installation separately; scheduled runs verify maintenance state after commit. timers = {name: "ok" for name in timers} services = {name: "ok" for name in services} - instance = values.get("CI_FLEET_INSTANCE", "unknown") stale = _stale_resources(run, instance) if docker_ok else {"containers": 0, "networks": 0, "volumes": 0} stale["images"] = _count(run, ["docker", "images", "-q", "--filter", "dangling=true", "--filter", "label=io.randomdevelopment.ci-fleet.managed=true"]) if docker_ok else 0 stale["build_cache"] = _count(run, ["docker", "buildx", "du", "--filter", "until=168h", "--format", "json"]) if docker_ok else 0 @@ -378,8 +527,17 @@ def collect_snapshot(values: dict[str, str], *, root: Path = Path("/"), run: Run "controller_id": instance, "desired_state": values.get("CI_FLEET_CONTROLLER_STATE", "active"), "disks": {"root": _disk(str(root)), "docker": _disk(str(root / docker_root.lstrip("/")))}, + "cpu": _cpu(root), + "memory": memory, + "swap": swap_metrics, + "load": {"one": loads[0], "five": loads[1], "fifteen": loads[2]}, + "boot_time": _boot_time(root), + "ssh": _ssh_state(run), + "software_version": software_version if software_version != "unknown" else values.get("CI_FLEET_ENGINE_REF", "unknown"), + "runners": runners, + "controller_status_valid": controller_status_valid, "memory_available_percent": available, - "load_per_cpu": os.getloadavg()[2] / max(os.cpu_count() or 1, 1), + "load_per_cpu": loads[2] / max(os.cpu_count() or 1, 1), "swap_used_percent": swap if (pressure := _memory_pressure(root)) is None or pressure >= 0.1 else 0, "recent_oom": oom.returncode == 0 and bool(oom.stdout.strip()), "docker_available": docker_ok, @@ -425,11 +583,71 @@ def _write_report(path: Path, report: dict[str, Any]) -> None: temporary.replace(path) -def _send_heartbeat(values: dict[str, str], report: dict[str, Any]) -> int: +def _send_status( + values: dict[str, str], + report: dict[str, Any], + *, + now: int | None = None, + nonce: str | None = None, + opener: Callable[..., Any] | None = None, +) -> int: + url = values.get("CI_FLEET_HEALTH_STATUS_URL") + if not url: + return 0 + try: + parsed = urllib.parse.urlsplit(url) + parsed.port + except ValueError: + return 1 + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.path != "/v1/status" or parsed.query or parsed.fragment: + return 1 + key_file = values.get("CI_FLEET_HEALTH_STATUS_KEY_FILE") + if not key_file: + return 1 + path = Path(key_file) + try: + info = path.stat() + expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0 + if info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) & 0o077: + return 1 + key = path.read_bytes() + except OSError: + return 1 + if not 32 <= len(key) <= 128: + return 1 + body = json.dumps(report, separators=(",", ":"), sort_keys=True).encode() + if len(body) > 32_768: + return 1 + generated_at = int(now if now is not None else time.time()) + request = urllib.request.Request( + url, + data=body, + headers=sign_headers(report["controller"]["id"], body, key, timestamp=generated_at, nonce=nonce or secrets.token_hex(16)), + method="POST", + ) + try: + transport = opener or urllib.request.build_opener(_NoRedirect).open + with transport(request, timeout=10) as response: + return 0 if 200 <= response.status < 300 else 1 + except (OSError, ValueError, http.client.HTTPException): + return 1 + + +def _send_heartbeat( + values: dict[str, str], + report: dict[str, Any], + *, + opener: Callable[..., Any] | None = None, +) -> int: url = values.get("CI_FLEET_HEALTH_HEARTBEAT_URL") if not url: return 0 - if not url.startswith("https://"): + try: + parsed = urllib.parse.urlsplit(url) + parsed.port + except ValueError: + return 2 + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: return 2 headers = {"Content-Type": "application/json"} token_file = values.get("CI_FLEET_HEALTH_HEARTBEAT_TOKEN_FILE") @@ -437,30 +655,39 @@ def _send_heartbeat(values: dict[str, str], report: dict[str, Any]) -> int: path = Path(token_file) try: info = path.stat() - if info.st_uid != 0 or stat.S_IMODE(info.st_mode) & 0o077: + expected_owner = os.getuid() if os.environ.get("CI_FLEET_TESTING") == "1" else 0 + if info.st_uid != expected_owner or stat.S_IMODE(info.st_mode) & 0o077: return 2 headers["Authorization"] = f"Bearer {path.read_text().strip()}" except OSError: return 2 request = urllib.request.Request(url, data=json.dumps(report).encode(), headers=headers, method="POST") try: - with urllib.request.urlopen(request, timeout=10) as response: + transport = opener or urllib.request.build_opener(_NoRedirect).open + with transport(request, timeout=10) as response: return 0 if 200 <= response.status < 300 else 1 - except OSError: + except (OSError, ValueError, http.client.HTTPException): return 1 def _local(args: argparse.Namespace) -> int: values = dict(os.environ) values.update(load_monitoring_config(args.monitoring_config)) - report = evaluate(collect_snapshot(values), thresholds_from(values)) - report["timestamp"] = int(time.time()) - heartbeat = _send_heartbeat(values, report) - if heartbeat: - severity = "critical" if heartbeat == 2 else "warning" - report["checks"].append({"id": "heartbeat_delivery", "status": severity}) - if heartbeat > report["exit_code"]: - report["status"], report["exit_code"] = ("warning", 1) if heartbeat == 1 else ("unhealthy", 2) + snapshot = collect_snapshot(values) + report = evaluate(snapshot, thresholds_from(values)) + now = int(time.time()) + report["timestamp"] = now + delivery = 0 + if values.get("CI_FLEET_HEALTH_SUPPRESS_DELIVERY") != "1": + delivery = ( + _send_status(values, build_status_report(snapshot, report, generated_at=now), now=now) + if values.get("CI_FLEET_HEALTH_STATUS_URL") else _send_heartbeat(values, report) + ) + if delivery: + severity = "critical" if delivery == 2 else "warning" + report["checks"].append({"id": "status_delivery", "status": severity}) + if delivery > report["exit_code"]: + report["status"], report["exit_code"] = ("unhealthy", 2) if delivery == 2 else ("warning", 1) _write_report(args.output, report) print(json.dumps(report, sort_keys=True) if args.json else render_human(report)) return int(report["exit_code"]) diff --git a/scripts/install-worker-controller.sh b/scripts/install-worker-controller.sh index 37b3f252..6eb9f240 100755 --- a/scripts/install-worker-controller.sh +++ b/scripts/install-worker-controller.sh @@ -437,6 +437,9 @@ runtime_release_complete() { [[ -x "$path/scripts/preflight.sh" && -x "$path/scripts/healthcheck.sh" && -x "$path/scripts/cleanup.sh" ]] || return 1 if grep -Fq 'scripts/health.py' "$path/scripts/healthcheck.sh"; then [[ -f "$path/scripts/health.py" ]] || return 1 + if grep -Fq 'build_status_report' "$path/scripts/health.py"; then + [[ -f "$path/scripts/status_auth.py" && -f "$path/controller/status.go" ]] || return 1 + fi fi for required in controller/Dockerfile controller/go.mod controller/main.go controller/config.go controller/scaler.go controller/state.go runner/Dockerfile; do [[ -f "$path/$required" ]] || return 1 @@ -766,6 +769,7 @@ run_health_check() { . "$environment" set +a [[ "$bootstrap" != true ]] || export CI_FLEET_HEALTH_BOOTSTRAP=1 + export CI_FLEET_HEALTH_SUPPRESS_DELIVERY=1 "$release/scripts/healthcheck.sh" ) || result=$? ((result < 2)) diff --git a/scripts/remote-reconcile.sh b/scripts/remote-reconcile.sh index a4b5bb16..e4eb7521 100755 --- a/scripts/remote-reconcile.sh +++ b/scripts/remote-reconcile.sh @@ -74,19 +74,32 @@ require_commands() { # --- State persistence --- save_reconcile_state() { - local status=${1:-} desired_commit=${2:-} applied_commit=${3:-} health=${4:-} message=${5:-} + local status=${1:-} desired_commit=${2:-} applied_commit=${3:-} health=${4:-} message=${5:-} mark_success=${6:-false} install -d -m 0700 "$reconcile_state_dir" - python3 - "$reconcile_state_file" "$status" "$desired_commit" "$applied_commit" "$health" "$message" <<'PY' 2>/dev/null || true + python3 - "$reconcile_state_file" "$status" "$desired_commit" "$applied_commit" "$health" "$message" "$mark_success" <<'PY' 2>/dev/null || true import json, os, sys, tempfile path = sys.argv[1] +now = int(__import__("time").time()) +try: + previous = json.load(open(path, encoding="utf-8")) + if not isinstance(previous, dict): + raise ValueError("reconciliation state must be an object") + last_success_at = previous.get("last_success_at") + if not isinstance(last_success_at, int) or last_success_at < 0: + last_success_at = None +except (OSError, ValueError, TypeError): + last_success_at = None +if sys.argv[7] == "true": + last_success_at = now state = { "status": sys.argv[2], "desired_commit": sys.argv[3] or "", "applied_commit": sys.argv[4] or "", "health": sys.argv[5] or "", "message": sys.argv[6] or "", - "checked_at": int(__import__("time").time()), + "checked_at": now, + "last_success_at": last_success_at, } fd, tmp = tempfile.mkstemp(prefix=".reconcile-state.", dir=os.path.dirname(path), text=True) try: @@ -310,6 +323,7 @@ run_health_check() { # shellcheck disable=SC1090 [[ ! -f "$rendered_env" ]] || . "$rendered_env" set +a + export CI_FLEET_HEALTH_SUPPRESS_DELIVERY=1 python3 "$repo_root/scripts/health.py" local --output "$output" >/dev/null 2>&1 ) || true python3 -c "import json; print(json.load(open('$output'))['status'])" 2>/dev/null || echo "unknown" @@ -387,7 +401,7 @@ if [[ "$desired_commit" == "$installed_config_ref" ]]; then --ref "$installed_config_ref" \ --controller "$installed_controller" 2>"$temp_dir/drift_err"; then note "CONVERGED controller=${installed_controller} config_ref=${installed_config_ref}" - save_reconcile_state 'converged' "$desired_commit" "$installed_config_ref" 'healthy' 'no change, converged' + save_reconcile_state 'converged' "$desired_commit" "$installed_config_ref" 'healthy' 'no change, converged' true exit 0 fi fi @@ -451,7 +465,7 @@ if CI_FLEET_INSTALLER_LOCK_FD=9 "$installer" --upgrade \ save_reconcile_state 'converged' "$desired_commit" "$desired_commit" 'unknown' "reconciled to ${desired_commit}; checking health" health_status=$(run_health_check "$temp_dir/health.json") - save_reconcile_state 'converged' "$desired_commit" "$desired_commit" "$health_status" "reconciled to ${desired_commit}" + save_reconcile_state 'converged' "$desired_commit" "$desired_commit" "$health_status" "reconciled to ${desired_commit}" true note "RECONCILE_OK controller=${installed_controller} desired=${desired_commit} applied=${desired_commit} health=${health_status}" exit 0 else diff --git a/scripts/status_auth.py b/scripts/status_auth.py new file mode 100644 index 00000000..fab0bda3 --- /dev/null +++ b/scripts/status_auth.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import hashlib +import hmac +from typing import Mapping + +AUTH_SCHEME = "CI-Fleet-HMAC-SHA256" + + +def canonical_request(controller: str, timestamp: int, nonce: str, body: bytes) -> bytes: + digest = hashlib.sha256(body).hexdigest() + return f"POST\n/v1/status\n{controller}\n{timestamp}\n{nonce}\n{digest}".encode() + + +def sign_headers(controller: str, body: bytes, key: bytes, *, timestamp: int, nonce: str) -> dict[str, str]: + signature = hmac.new(key, canonical_request(controller, timestamp, nonce, body), hashlib.sha256).hexdigest() + return { + "Authorization": f"{AUTH_SCHEME} {signature}", + "Content-Type": "application/json", + "X-CI-Fleet-Controller": controller, + "X-CI-Fleet-Timestamp": str(timestamp), + "X-CI-Fleet-Nonce": nonce, + } + + +def verify_headers(headers: Mapping[str, str], body: bytes, key: bytes) -> tuple[str, int, str]: + values = {name.lower(): value for name, value in headers.items()} + controller = values.get("x-ci-fleet-controller", "") + nonce = values.get("x-ci-fleet-nonce", "") + try: + timestamp = int(values.get("x-ci-fleet-timestamp", "")) + except ValueError as error: + raise ValueError("invalid authentication timestamp") from error + authorization = values.get("authorization", "") + prefix = f"{AUTH_SCHEME} " + if not authorization.startswith(prefix): + raise ValueError("missing authentication signature") + expected = hmac.new(key, canonical_request(controller, timestamp, nonce, body), hashlib.sha256).hexdigest() + if not hmac.compare_digest(authorization[len(prefix):], expected): + raise ValueError("invalid authentication signature") + return controller, timestamp, nonce diff --git a/scripts/status_receiver.py b/scripts/status_receiver.py new file mode 100644 index 00000000..dd84950f --- /dev/null +++ b/scripts/status_receiver.py @@ -0,0 +1,466 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import hmac +import http.server +import ipaddress +import json +import os +import re +import socket +import sqlite3 +import threading +import time +import urllib.parse +from contextlib import closing +from pathlib import Path +from typing import Any, Mapping + +from status_auth import verify_headers + +CONTROLLER_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}") +NONCE = re.compile(r"[0-9a-f]{32}") + + +class StatusError(ValueError): + def __init__(self, status: int, code: str): + super().__init__(code) + self.status = status + self.code = code + + +class StatusReceiver: + def __init__( + self, + database: Path, + controller_keys: Mapping[str, bytes], + *, + read_token: str, + history_limit: int = 288, + retention_seconds: int = 604_800, + min_interval_seconds: int = 30, + max_payload_bytes: int = 32_768, + max_clock_skew_seconds: int = 300, + ) -> None: + if history_limit < 1 or retention_seconds < 1: + raise ValueError("history and retention bounds must be positive") + if len(set(controller_keys.values())) != len(controller_keys): + raise ValueError("controller authentication keys must be unique") + if any(hmac.compare_digest(key, read_token.encode()) for key in controller_keys.values()): + raise ValueError("read token must differ from controller authentication keys") + self.database = database + self.controller_keys = dict(controller_keys) + self.read_token = read_token + self.history_limit = history_limit + self.retention_seconds = retention_seconds + self.min_interval_seconds = min_interval_seconds + self.max_payload_bytes = max_payload_bytes + self.max_clock_skew_seconds = max_clock_skew_seconds + # ponytail: one receiver-wide lock; split by controller only if measured write contention warrants it. + self._write_lock = threading.Lock() + self._last_attempt: dict[str, int] = {} + self._clock = time.time + with closing(self._connect()) as connection, connection: + connection.executescript(""" + CREATE TABLE IF NOT EXISTS reports ( + controller TEXT NOT NULL, + generated_at INTEGER NOT NULL, + received_at INTEGER NOT NULL, + payload TEXT NOT NULL, + PRIMARY KEY (controller, generated_at) + ); + CREATE TABLE IF NOT EXISTS nonces ( + controller TEXT NOT NULL, + nonce TEXT NOT NULL, + authenticated_at INTEGER NOT NULL, + PRIMARY KEY (controller, nonce) + ); + """) + os.chmod(self.database, 0o600) + self.expire() + + def _connect(self) -> sqlite3.Connection: + self.database.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + try: + descriptor = os.open(self.database, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + pass + else: + os.close(descriptor) + return sqlite3.connect(self.database) + + def submit(self, body: bytes, headers: Mapping[str, str], *, now: int) -> None: + if len(body) > self.max_payload_bytes: + raise StatusError(413, "payload_too_large") + claimed = {name.lower(): value for name, value in headers.items()}.get("x-ci-fleet-controller", "") + key = self.controller_keys.get(claimed) + known_controller = key is not None + try: + controller, authenticated_at, nonce = verify_headers(headers, body, key or b"\0" * 32) + except ValueError as error: + raise StatusError(401, "authentication_failed") from error + if not known_controller: + raise StatusError(401, "authentication_failed") + if controller != claimed or abs(now - authenticated_at) > self.max_clock_skew_seconds or not NONCE.fullmatch(nonce): + raise StatusError(401, "authentication_stale") + try: + report = json.loads(body) + except (UnicodeDecodeError, json.JSONDecodeError) as error: + raise StatusError(400, "invalid_json") from error + self._validate_minimum(report, controller) + generated_at = report["generated_at"] + if abs(now - generated_at) > self.max_clock_skew_seconds: + raise StatusError(409, "report_time_stale") + encoded = json.dumps(report, separators=(",", ":"), sort_keys=True) + with self._write_lock, closing(self._connect()) as connection, connection: + if connection.execute("SELECT 1 FROM nonces WHERE controller=? AND nonce=?", (controller, nonce)).fetchone(): + raise StatusError(409, "replayed_report") + last_attempt = self._last_attempt.get(controller) + if last_attempt is not None and now - last_attempt < 1: + raise StatusError(429, "submission_too_frequent") + self._last_attempt[controller] = now + try: + connection.execute("INSERT INTO nonces VALUES (?, ?, ?)", (controller, nonce, authenticated_at)) + connection.commit() + except sqlite3.IntegrityError as error: + raise StatusError(409, "replayed_report") from error + last = connection.execute( + "SELECT generated_at, received_at FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT 1", + (controller,), + ).fetchone() + if last and generated_at <= last[0]: + raise StatusError(409, "stale_report") + if last and now - last[1] < self.min_interval_seconds: + raise StatusError(429, "submission_too_frequent") + connection.execute("INSERT INTO reports VALUES (?, ?, ?, ?)", (controller, generated_at, now, encoded)) + connection.execute("DELETE FROM reports WHERE received_at < ?", (now - self.retention_seconds,)) + connection.execute( + "DELETE FROM reports WHERE controller=? AND rowid NOT IN (SELECT rowid FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT ?)", + (controller, controller, self.history_limit), + ) + connection.execute("DELETE FROM nonces WHERE authenticated_at < ?", (now - self.max_clock_skew_seconds,)) + + def _expire(self, connection: sqlite3.Connection, now: int) -> None: + connection.execute("DELETE FROM reports WHERE received_at < ?", (now - self.retention_seconds,)) + connection.execute(""" + DELETE FROM reports WHERE rowid IN ( + SELECT rowid FROM ( + SELECT rowid, ROW_NUMBER() OVER (PARTITION BY controller ORDER BY generated_at DESC, rowid DESC) AS rank + FROM reports + ) WHERE rank > ? + ) + """, (self.history_limit,)) + connection.execute("DELETE FROM nonces WHERE authenticated_at < ?", (now - self.max_clock_skew_seconds,)) + + def expire(self) -> None: + with self._write_lock, closing(self._connect()) as connection, connection: + self._expire(connection, int(self._clock())) + + @staticmethod + def _validate_minimum(report: Any, controller: str) -> None: + def exact(value: Any, keys: set[str]) -> bool: + return isinstance(value, dict) and set(value) == keys + + def integer(value: Any, minimum: int = 0) -> bool: + return isinstance(value, int) and not isinstance(value, bool) and value >= minimum + + def number(value: Any, minimum: float = 0) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) and value >= minimum and value < float("inf") + + def enum(value: Any, choices: set[str]) -> bool: + return isinstance(value, str) and value in choices + + if not isinstance(report, dict) or type(report.get("schema_version")) is not int or report["schema_version"] != 1: + raise StatusError(400, "unsupported_schema") + root_keys = {"schema_version", "controller", "configuration", "reconciliation", "drift", "process", "timers", "runners", "metrics", "docker", "error", "generated_at"} + if set(report) != root_keys: + raise StatusError(400, "invalid_report") + identity = report["controller"] + if not exact(identity, {"id", "software_version", "boot_time", "ssh"}) or identity["id"] != controller or not CONTROLLER_ID.fullmatch(controller): + raise StatusError(403, "controller_identity_mismatch") + if not isinstance(identity["software_version"], str) or not re.fullmatch(r"[A-Za-z0-9_.+-]{1,64}", identity["software_version"]): + raise StatusError(400, "invalid_report") + if not integer(identity["boot_time"]) or not enum(identity["ssh"], {"enabled", "disabled", "unknown"}): + raise StatusError(400, "invalid_report") + generated_at = report["generated_at"] + if not integer(generated_at) or identity["boot_time"] > generated_at: + raise StatusError(400, "invalid_report") + + configuration = report["configuration"] + commit = lambda value: isinstance(value, str) and (value == "" or re.fullmatch(r"[0-9a-f]{40}", value)) + if not exact(configuration, {"desired_commit", "applied_commit"}) or not all(commit(configuration[name]) for name in configuration): + raise StatusError(400, "invalid_report") + reconciliation = report["reconciliation"] + reconcile_states = {"bootstrap", "converged", "drift", "failed", "invalid", "missing", "pending", "reconciling", "rolled_back", "unknown"} + if not exact(reconciliation, {"state", "last_success_at"}) or not enum(reconciliation["state"], reconcile_states): + raise StatusError(400, "invalid_report") + if reconciliation["last_success_at"] is not None and (not integer(reconciliation["last_success_at"]) or reconciliation["last_success_at"] > generated_at): + raise StatusError(400, "invalid_report") + if not exact(report["drift"], {"state"}) or not enum(report["drift"]["state"], {"ok", "stale", "failed", "unknown"}): + raise StatusError(400, "invalid_report") + process = report["process"] + if not exact(process, {"state", "restart_count"}) or not enum(process["state"], {"created", "exited", "missing", "paused", "restarting", "running", "unknown"}) or not integer(process["restart_count"]): + raise StatusError(400, "invalid_report") + timers = report["timers"] + if not exact(timers, {"reconciliation", "drift", "health", "cleanup"}) or any(not enum(value, {"ok", "stale", "failed", "unknown"}) for value in timers.values()): + raise StatusError(400, "invalid_report") + runners = report["runners"] + if not exact(runners, {"current", "busy", "maximum"}) or not all(integer(value) for value in runners.values()) or not (runners["busy"] <= runners["current"] <= runners["maximum"]): + raise StatusError(400, "invalid_report") + + metrics = report["metrics"] + if not exact(metrics, {"cpu", "memory", "swap", "disk", "inodes", "load"}): + raise StatusError(400, "invalid_report") + cpu = metrics["cpu"] + if not exact(cpu, {"logical", "used_percent"}) or not integer(cpu["logical"], 1) or not number(cpu["used_percent"]) or cpu["used_percent"] > 100: + raise StatusError(400, "invalid_report") + for name in ("memory", "swap"): + value = metrics[name] + total_key, part_key = ("total_bytes", "available_bytes") if name == "memory" else ("total_bytes", "used_bytes") + if not exact(value, {total_key, part_key}) or not integer(value[total_key]) or not integer(value[part_key]) or value[part_key] > value[total_key]: + raise StatusError(400, "invalid_report") + for group, keys in (("disk", {"total_bytes", "used_bytes"}), ("inodes", {"total", "used"})): + if not exact(metrics[group], {"root", "docker"}): + raise StatusError(400, "invalid_report") + total_key, used_key = tuple(keys) + if total_key.startswith("used"): + total_key, used_key = used_key, total_key + for value in metrics[group].values(): + if not exact(value, keys) or not integer(value[total_key]) or not integer(value[used_key]) or value[used_key] > value[total_key]: + raise StatusError(400, "invalid_report") + load = metrics["load"] + if not exact(load, {"one", "five", "fifteen"}) or not all(number(value) for value in load.values()): + raise StatusError(400, "invalid_report") + docker = report["docker"] + if not exact(docker, {"healthy", "oom"}) or not all(isinstance(value, bool) for value in docker.values()): + raise StatusError(400, "invalid_report") + error = report["error"] + if error is not None and (not exact(error, {"code", "message"}) or not isinstance(error["code"], str) or not re.fullmatch(r"[a-z0-9_]{1,64}", error["code"]) or error["message"] != error["code"].replace("_", " ")): + raise StatusError(400, "invalid_report") + + def _authorize_read(self, read_token: str) -> None: + if not hmac.compare_digest(read_token.encode(), self.read_token.encode()): + raise StatusError(401, "read_authentication_failed") + + def latest(self, controller: str, read_token: str) -> dict[str, Any] | None: + self._authorize_read(read_token) + with self._write_lock, closing(self._connect()) as connection, connection: + self._expire(connection, int(self._clock())) + row = connection.execute( + "SELECT payload FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT 1", (controller,) + ).fetchone() + return json.loads(row[0]) if row else None + + def history(self, controller: str, read_token: str, limit: int | None = None) -> list[dict[str, Any]]: + self._authorize_read(read_token) + count = min(max(limit or self.history_limit, 1), self.history_limit) + with self._write_lock, closing(self._connect()) as connection, connection: + self._expire(connection, int(self._clock())) + rows = connection.execute( + "SELECT payload FROM reports WHERE controller=? ORDER BY generated_at DESC LIMIT ?", (controller, count) + ).fetchall() + return [json.loads(row[0]) for row in rows] + + def latest_and_history(self, controller: str, read_token: str) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: + self._authorize_read(read_token) + with self._write_lock, closing(self._connect()) as connection, connection: + self._expire(connection, int(self._clock())) + rows = connection.execute( + "SELECT payload FROM reports WHERE controller = ? ORDER BY generated_at DESC LIMIT ?", + (controller, self.history_limit), + ).fetchall() + history = [json.loads(row[0]) for row in rows] + return (history[0] if history else None), history + + def list_latest(self, read_token: str) -> list[dict[str, Any]]: + self._authorize_read(read_token) + with self._write_lock, closing(self._connect()) as connection, connection: + self._expire(connection, int(self._clock())) + rows = connection.execute(""" + SELECT reports.payload FROM reports + JOIN (SELECT controller, MAX(generated_at) AS generated_at FROM reports GROUP BY controller) latest + USING (controller, generated_at) + ORDER BY reports.controller + """).fetchall() + return [json.loads(row[0]) for row in rows] + + +class _BoundedHTTPServer(http.server.ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, *args: Any, max_requests: int = 32, request_timeout: int = 15, **kwargs: Any) -> None: + self.max_requests = max_requests + self.request_timeout = request_timeout + self._slots = threading.BoundedSemaphore(max_requests) + self._deadlines: dict[Any, threading.Timer] = {} + super().__init__(*args, **kwargs) + + def process_request(self, request: Any, client_address: Any) -> None: + if not self._slots.acquire(blocking=False): + request.close() + return + try: + request.settimeout(self.request_timeout) + timer = threading.Timer(self.request_timeout, self._expire_request, (request,)) + timer.daemon = True + self._deadlines[request] = timer + timer.start() + super().process_request(request, client_address) + except Exception: + timer = self._deadlines.pop(request, None) + if timer: + timer.cancel() + self._slots.release() + raise + + @staticmethod + def _expire_request(request: Any) -> None: + try: + request.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + def process_request_thread(self, request: Any, client_address: Any) -> None: + try: + super().process_request_thread(request, client_address) + finally: + timer = self._deadlines.pop(request, None) + if timer: + timer.cancel() + self._slots.release() + + +def create_server(bind: str, port: int, receiver: StatusReceiver) -> _BoundedHTTPServer: + class Handler(http.server.BaseHTTPRequestHandler): + def send_json(self, status: int, value: Any) -> None: + body = json.dumps(value, separators=(",", ":"), sort_keys=True).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def bearer(self) -> str: + authorization = self.headers.get("Authorization", "") + return authorization[7:] if authorization.startswith("Bearer ") else "" + + def do_POST(self) -> None: + if self.path != "/v1/status": + self.send_json(404, {"error": "not_found"}) + return + try: + length = int(self.headers.get("Content-Length", "-1")) + if length < 0 or length > receiver.max_payload_bytes: + raise StatusError(413, "payload_too_large") + receiver.submit(self.rfile.read(length), dict(self.headers.items()), now=int(__import__("time").time())) + self.send_json(202, {"accepted": True}) + except (ValueError, StatusError) as error: + failure = error if isinstance(error, StatusError) else StatusError(400, "invalid_request") + self.send_json(failure.status, {"error": failure.code}) + + def do_GET(self) -> None: + path = urllib.parse.urlsplit(self.path) + try: + if path.query or path.fragment: + raise StatusError(400, "invalid_request") + if path.path == "/v1/controllers": + value = {"schema_version": 1, "controllers": receiver.list_latest(self.bearer())} + elif path.path.startswith("/v1/controllers/"): + controller = urllib.parse.unquote(path.path.removeprefix("/v1/controllers/")) + if not CONTROLLER_ID.fullmatch(controller): + raise StatusError(404, "not_found") + latest, history = receiver.latest_and_history(controller, self.bearer()) + value = { + "schema_version": 1, + "latest": latest, + "history": history, + } + else: + raise StatusError(404, "not_found") + self.send_json(200, value) + except StatusError as error: + self.send_json(error.status, {"error": error.code}) + + def log_message(self, format: str, *args: Any) -> None: + pass + + if ipaddress.ip_address(bind).version == 6: + class IPv6Server(_BoundedHTTPServer): + address_family = socket.AF_INET6 + return IPv6Server((bind, port), Handler) + return _BoundedHTTPServer((bind, port), Handler) + + +def _read_secret(path: Path, *, textual: bool = False) -> bytes: + info = path.stat() + if info.st_uid != os.getuid() or info.st_mode & 0o077: + raise ValueError(f"secret must be owned by the receiver user with mode 0600: {path}") + value = path.read_bytes() + if textual: + value = value.strip() + if any(byte < 0x21 or byte > 0x7e for byte in value): + raise ValueError(f"textual secret must contain visible ASCII only: {path}") + if not 32 <= len(value) <= 128: + raise ValueError(f"secret must contain 32-128 bytes: {path}") + return value + + +def load_auth_config(path: Path) -> tuple[dict[str, bytes], str]: + info = path.stat() + if info.st_uid != os.getuid() or info.st_mode & 0o077: + raise ValueError(f"auth config must be owned by the receiver user with mode 0600: {path}") + value = json.loads(path.read_text()) + if not isinstance(value, dict) or set(value) != {"controllers", "read_token_file"} or not isinstance(value["controllers"], dict): + raise ValueError("auth config must contain controllers and read_token_file only") + resolve = lambda name: Path(name) if Path(name).is_absolute() else path.parent / name + keys = {} + for controller, key_file in value["controllers"].items(): + if not isinstance(controller, str) or not CONTROLLER_ID.fullmatch(controller) or not isinstance(key_file, str): + raise ValueError("invalid controller auth mapping") + keys[controller] = _read_secret(resolve(key_file)) + if not keys or not isinstance(value["read_token_file"], str): + raise ValueError("auth config requires at least one controller and a read token") + return keys, _read_secret(resolve(value["read_token_file"]), textual=True).decode() + + +def _expiration_loop(receiver: StatusReceiver, stop: threading.Event, interval: float = 60) -> None: + while not stop.wait(interval): + try: + receiver.expire() + except (OSError, sqlite3.Error): + pass + + +def main() -> int: + parser = argparse.ArgumentParser(description="Authenticated ci-fleet status receiver") + parser.add_argument("--auth-config", type=Path, required=True) + parser.add_argument("--database", type=Path, required=True) + parser.add_argument("--bind", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8080) + parser.add_argument("--history-limit", type=int, default=288) + parser.add_argument("--retention-seconds", type=int, default=604_800) + args = parser.parse_args() + if not ipaddress.ip_address(args.bind).is_loopback: + parser.error("--bind must be a loopback address; terminate HTTPS in a reverse proxy") + keys, read_token = load_auth_config(args.auth_config) + receiver = StatusReceiver( + args.database, keys, read_token=read_token, + history_limit=args.history_limit, retention_seconds=args.retention_seconds, + ) + stop = threading.Event() + maintenance = threading.Thread(target=_expiration_loop, args=(receiver, stop), daemon=True) + maintenance.start() + server = create_server(args.bind, args.port, receiver) + try: + server.serve_forever() + finally: + stop.set() + maintenance.join() + server.server_close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/test-install-worker-controller.sh b/scripts/test-install-worker-controller.sh index 6cfc4af6..b75e5799 100755 --- a/scripts/test-install-worker-controller.sh +++ b/scripts/test-install-worker-controller.sh @@ -5,6 +5,7 @@ repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd) health_timer=$repo_root/host/systemd/ci-fleet-health.timer grep -Fqx 'OnActiveSec=2min' "$health_timer" || { printf 'FAIL: health timer lacks activation-relative initial trigger\n' >&2; exit 1; } ! grep -Fq 'OnBootSec=' "$health_timer" || { printf 'FAIL: health timer initial trigger is boot-relative\n' >&2; exit 1; } +grep -Fq 'export CI_FLEET_HEALTH_SUPPRESS_DELIVERY=1' "$repo_root/scripts/install-worker-controller.sh" || { printf 'FAIL: installer health check can submit monitoring reports\n' >&2; exit 1; } tmp=$(mktemp -d) trap 'rm -rf "$tmp"' EXIT fake_bin=$tmp/bin @@ -488,12 +489,15 @@ expect_success "$installer" --install "${base_args[@]}" --ref "$ref_one" >/dev/n [[ $(stat -c '%i' "$active_release") == "$complete_release_inode" ]] || fail 'complete immutable release was replaced instead of reused' warning_ref=$(write_config active 1 2) -printf 'CI_FLEET_HEALTH_DISK_WARN_PERCENT=0\n' >"$root/etc/ci-fleet/monitoring.env" +printf '%s\n' \ + 'CI_FLEET_HEALTH_DISK_WARN_PERCENT=0' \ + 'CI_FLEET_HEALTH_STATUS_URL=https://status.example.invalid/v1/status' >"$root/etc/ci-fleet/monitoring.env" chmod 600 "$root/etc/ci-fleet/monitoring.env" warning_output=$tmp/warning-upgrade.out "$installer" --upgrade "${base_args[@]}" --ref "$warning_ref" >"$warning_output" 2>&1 warning_upgrade=$(<"$warning_output") grep -Fq 'WARNING disk_root' <<<"$warning_upgrade" || fail 'warning health fixture did not produce a warning result' +if grep -Fq 'WARNING status_delivery' <<<"$warning_upgrade"; then fail 'ad-hoc reconciliation health check submitted duplicate status'; fi grep -Fq 'CONVERGED mode=upgrade' <<<"$warning_upgrade" || fail 'warning health result did not report convergence' grep -Fq "CI_FLEET_CONFIG_REF=$warning_ref" "$root/etc/ci-fleet/ci-fleet.env" || fail 'warning health result rolled back an otherwise healthy activation' rm -f "$root/etc/ci-fleet/monitoring.env" @@ -603,6 +607,7 @@ chmod 600 "$adopt_pem" cp "$repo_root/deploy/compose.yaml" "$adopt_root/opt/ci-fleet/deploy/compose.yaml" cp "$repo_root/scripts/healthcheck.sh" "$adopt_root/opt/ci-fleet/scripts/healthcheck.sh" cp "$repo_root/scripts/health.py" "$adopt_root/opt/ci-fleet/scripts/health.py" +cp "$repo_root/scripts/status_auth.py" "$adopt_root/opt/ci-fleet/scripts/status_auth.py" cp "$repo_root/scripts/cleanup.sh" "$adopt_root/opt/ci-fleet/scripts/cleanup.sh" chmod 0755 "$adopt_root/opt/ci-fleet/scripts/healthcheck.sh" "$adopt_root/opt/ci-fleet/scripts/cleanup.sh" printf '%s\n' \ diff --git a/scripts/test_health.py b/scripts/test_health.py index 8fa8e741..2cf89d3e 100644 --- a/scripts/test_health.py +++ b/scripts/test_health.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 import copy import importlib.util +import json +import os import sys import tempfile import unittest @@ -113,10 +115,17 @@ def test_malformed_reconciliation_state_is_observable(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "state.json" path.write_text('{"status":[],"health":{},"desired_commit":[],"applied_commit":null}\n') + reconciliation = health._reconcile_state(path) self.assertEqual( - health._reconcile_state(path), - {"status": "invalid", "desired_commit": "invalid", "applied_commit": "invalid", "health": "invalid"}, + reconciliation, + {"status": "invalid", "desired_commit": "invalid", "applied_commit": "invalid", "health": "invalid", "last_success_at": None}, ) + snapshot = healthy_snapshot() + snapshot["controller_id"] = "example-ci-01" + snapshot["reconciliation"] = reconciliation + report = health.build_status_report(snapshot, health.evaluate(snapshot, health.Thresholds()), generated_at=1_000) + self.assertEqual(report["configuration"], {"desired_commit": "", "applied_commit": ""}) + self.assertEqual(report["error"]["code"], "reconciliation_invalid") def test_external_heartbeats_detect_missing_and_stale_active_hosts(self) -> None: controllers = { @@ -198,6 +207,54 @@ def run(args): finally: health.os.getloadavg, health.os.cpu_count = original_load, original_cpus + def test_collector_builds_status_metrics_and_uses_controller_runner_state(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "proc").mkdir() + (root / "proc/stat").write_text("cpu 100 0 50 800 50 0 0 0 40 10\nbtime 900\n") + (root / "proc/meminfo").write_text("MemTotal: 1024 kB\nMemAvailable: 768 kB\nSwapTotal: 512 kB\nSwapFree: 384 kB\n") + generated_at = [int(health.time.time())] + + def run(args): + if args[:3] == ["docker", "exec", "controller"]: + return health.subprocess.CompletedProcess(args, 0, json.dumps({ + "controller": "example-ci-01", "software_version": "1" * 40, + "current": 2, "busy": 1, "maximum": 6, "generated_at": generated_at[0], + }), "") + if args[:2] == ["docker", "info"]: + return health.subprocess.CompletedProcess(args, 0, "", "") + if args[:2] == ["docker", "inspect"]: + outputs = {"{{.State.Status}}": "running\n", "{{.State.OOMKilled}}": "false\n", "{{.RestartCount}}": "0\n", "{{range .Config.Env}}{{println .}}{{end}}": "CI_FLEET_MIN_RUNNERS=0\nCI_FLEET_MAX_RUNNERS=6\n"} + return health.subprocess.CompletedProcess(args, 0, outputs.get(args[3], ""), "") + if args[:2] == ["systemctl", "is-enabled"] and args[-1] in {"ssh.service", "ssh.socket", "sshd.service"}: + return health.subprocess.CompletedProcess(args, 1, "disabled\n", "") + if args[:2] == ["systemctl", "is-active"] and args[-1] in {"ssh.service", "ssh.socket", "sshd.service"}: + return health.subprocess.CompletedProcess(args, 3, "inactive\n", "") + return health.subprocess.CompletedProcess(args, 0, "success\n", "") + + snapshot = health.collect_snapshot({ + "CI_FLEET_INSTANCE": "example-ci-01", "CI_FLEET_CONTROLLER_CONTAINER": "controller", + "CI_FLEET_MAX_RUNNERS": "6", "CI_FLEET_HEALTH_BOOTSTRAP": "1", + }, root=root, run=run) + self.assertEqual(snapshot["runners"], {"current": 2, "busy": 1, "maximum": 6}) + self.assertEqual(snapshot["software_version"], "1" * 40) + self.assertEqual(snapshot["boot_time"], 900) + self.assertEqual(snapshot["ssh"], "disabled") + self.assertEqual(snapshot["memory"], {"total_bytes": 1048576, "available_bytes": 786432}) + self.assertEqual(snapshot["swap"], {"total_bytes": 524288, "used_bytes": 131072}) + self.assertAlmostEqual(snapshot["cpu"]["used_percent"], 20.0) + generated_at[0] -= 121 + self.assertEqual(health._controller_status(run, "controller", "example-ci-01", 6), ({"current": 0, "busy": 0, "maximum": 6}, "unknown", False)) + stale = health.collect_snapshot({ + "CI_FLEET_INSTANCE": "example-ci-01", "CI_FLEET_CONTROLLER_CONTAINER": "controller", + "CI_FLEET_MAX_RUNNERS": "6", "CI_FLEET_HEALTH_BOOTSTRAP": "1", + }, root=root, run=run) + self.assertFalse(stale["controller_status_valid"]) + stale_report = health.evaluate(stale, health.Thresholds()) + self.assertEqual(next(check for check in stale_report["checks"] if check["id"] == "controller_status")["status"], "warning") + outbound = health.build_status_report(stale, stale_report, generated_at=int(health.time.time())) + self.assertEqual(outbound["error"]["code"], "health_controller_status") + def test_threshold_overrides_validate_ordering(self) -> None: self.assertAlmostEqual(health._timespan_seconds("3d 1h 41min 40.5s"), 265300.5) self.assertAlmostEqual(health._timespan_seconds("1y 2month 3w 4d 5h 6min 7.5s"), 365.25 * 86400 + 2 * 365.25 * 86400 / 12 + 3 * 7 * 86400 + 4 * 86400 + 5 * 3600 + 6 * 60 + 7.5) @@ -212,7 +269,7 @@ def test_human_output_is_redacted(self) -> None: output = health.render_human(report) self.assertIn("HEALTHY controller=example-ci-01", output) self.assertNotIn("SHOULD_NOT_PRINT", output) - self.assertEqual(health._send_heartbeat({"CI_FLEET_HEALTH_HEARTBEAT_URL": "http://unsafe.invalid"}, report), 2) + self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "http://unsafe.invalid"}, {"controller": {"id": "example"}}), 1) def test_probe_failures_are_results_and_missing_units_fail(self) -> None: for error in (FileNotFoundError(), health.subprocess.TimeoutExpired(["probe"], 30)): @@ -228,6 +285,146 @@ def missing(args): self.assertEqual(health._unit_state(missing, "missing.service"), "failed") + def test_controller_runtime_directory_is_writable_by_nonroot_process(self) -> None: + dockerfile = (ROOT / "controller/Dockerfile").read_text() + self.assertIn("install -d -o 65532 -g 65532 /run/ci-fleet", dockerfile) + self.assertIn("USER 65532:65532", dockerfile) + + def test_status_report_contract_redaction_and_disabled_ssh(self) -> None: + snapshot = healthy_snapshot() + snapshot.update({ + "software_version": "1" * 40, + "boot_time": 900, + "ssh": "disabled", + "cpu": {"logical": 8, "used_percent": 25.0}, + "memory": {"total_bytes": 1024, "available_bytes": 768}, + "swap": {"total_bytes": 512, "used_bytes": 0}, + "load": {"one": 0.1, "five": 0.2, "fifteen": 0.3}, + "runners": {"current": 1, "busy": 1, "maximum": 6}, + "reconciliation": { + "status": "failed", "desired_commit": "2" * 40, "applied_commit": "3" * 40, + "health": "unhealthy", "last_success_at": 950, + "message": "token=SUPER_SECRET https://private.invalid/path", + }, + }) + snapshot["controller"]["state"] = "dead" + for disk in snapshot["disks"].values(): + disk.update(total_bytes=4096, used_bytes=1024, inode_total=1000, inode_used=100) + report = health.build_status_report(snapshot, health.evaluate(snapshot, health.Thresholds()), generated_at=1_000) + self.assertEqual(report["schema_version"], 1) + self.assertEqual(report["controller"]["ssh"], "disabled") + self.assertEqual(report["configuration"], {"desired_commit": "2" * 40, "applied_commit": "3" * 40}) + self.assertEqual(report["runners"], {"current": 1, "busy": 1, "maximum": 6}) + self.assertEqual(report["process"]["state"], "unknown") + self.assertEqual(report["error"], {"code": "reconciliation_failed", "message": "reconciliation failed"}) + priority = copy.deepcopy(snapshot) + priority["reconciliation"]["status"] = "converged" + priority["controller_status_valid"] = True + prioritized = health.build_status_report(priority, {"checks": [ + {"id": "early_warning", "status": "warning"}, + {"id": "later_failure", "status": "critical"}, + ]}, generated_at=1_000) + self.assertEqual(prioritized["error"]["code"], "health_later_failure") + snapshot["reconciliation"]["last_success_at"] = 1_001 + future_success = health.build_status_report(snapshot, health.evaluate(snapshot, health.Thresholds()), generated_at=1_000) + self.assertIsNone(future_success["reconciliation"]["last_success_at"]) + snapshot.update({"desired_state": "disabled", "controller_status_valid": False}) + snapshot["controller"]["state"] = "exited" + snapshot["reconciliation"].update({"status": "converged", "last_success_at": 900}) + maintenance = health.build_status_report(snapshot, health.evaluate(snapshot, health.Thresholds()), generated_at=1_000) + self.assertNotEqual(maintenance["error"]["code"], "health_controller_status") + encoded = json.dumps(report) + self.assertNotIn("SUPER_SECRET", encoded) + self.assertNotIn("private.invalid", encoded) + + def test_ssh_state_requires_service_and_socket_to_be_disabled(self) -> None: + def disabled(args): + if args[:2] == ["systemctl", "is-enabled"]: + return health.subprocess.CompletedProcess(args, 1, "disabled\n", "") + if args[:2] == ["systemctl", "is-active"]: + return health.subprocess.CompletedProcess(args, 3, "inactive\n", "") + return health.subprocess.CompletedProcess(args, 1, "", "") + + self.assertEqual(health._ssh_state(disabled), "disabled") + + def socket_enabled(args): + active = args[-1] == "ssh.socket" + return health.subprocess.CompletedProcess(args, 0 if active else 1, "active\n" if active else "inactive\n", "") + + self.assertEqual(health._ssh_state(socket_enabled), "enabled") + + def sshd_enabled(args): + active = args[-1] == "sshd.service" + return health.subprocess.CompletedProcess(args, 0 if active else 1, "active\n" if active else "not-found\n", "") + + self.assertEqual(health._ssh_state(sshd_enabled), "enabled") + self.assertEqual(health._ssh_state(lambda args: health.subprocess.CompletedProcess(args, 127, "", "")), "unknown") + + def test_legacy_configuration_failure_remains_critical(self) -> None: + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "health.json" + old_collect, old_send = getattr(health, "collect_snapshot"), getattr(health, "_send_heartbeat") + old_status_url = os.environ.pop("CI_FLEET_HEALTH_STATUS_URL", None) + setattr(health, "collect_snapshot", lambda _values: healthy_snapshot()) + setattr(health, "_send_heartbeat", lambda _values, _report: 2) + try: + result = health._local(health.argparse.Namespace( + monitoring_config=Path(directory) / "missing.env", output=output, json=True, + )) + report = json.loads(output.read_text()) + self.assertEqual((result, report["status"]), (2, "unhealthy")) + self.assertEqual(report["checks"][-1], {"id": "status_delivery", "status": "critical"}) + finally: + setattr(health, "collect_snapshot", old_collect) + setattr(health, "_send_heartbeat", old_send) + if old_status_url is not None: + os.environ["CI_FLEET_HEALTH_STATUS_URL"] = old_status_url + + def test_status_delivery_is_signed_and_outage_is_non_disruptive(self) -> None: + with tempfile.TemporaryDirectory() as directory: + key = Path(directory) / "status.key" + key.write_text("controller-key-32-bytes-long-0001\n") + key.chmod(0o600) + values = { + "CI_FLEET_HEALTH_STATUS_URL": "https://status.example.invalid/v1/status", + "CI_FLEET_HEALTH_STATUS_KEY_FILE": str(key), + } + report = {"controller": {"id": "example-ci-01"}, "generated_at": 1_000} + captured = {} + + class Response: + status = 202 + def __enter__(self): return self + def __exit__(self, *_): return None + + def opener(request, timeout): + captured.update(url=request.full_url, headers=dict(request.header_items()), body=request.data, timeout=timeout) + return Response() + + old = os.environ.get("CI_FLEET_TESTING") + os.environ["CI_FLEET_TESTING"] = "1" + try: + self.assertEqual(health._send_status(values, report, now=1_000, nonce="a" * 32, opener=opener), 0) + self.assertIn("CI-Fleet-HMAC-SHA256", captured["headers"]["Authorization"]) + expected = health.sign_headers("example-ci-01", captured["body"], key.read_bytes(), timestamp=1_000, nonce="a" * 32) + self.assertEqual(captured["headers"]["Authorization"], expected["Authorization"]) + original = copy.deepcopy(report) + self.assertEqual(health._send_status(values, report, now=1_000, nonce="b" * 32, opener=lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError())), 1) + self.assertEqual(health._send_status(values, report, now=1_000, nonce="c" * 32, opener=lambda *_args, **_kwargs: (_ for _ in ()).throw(health.http.client.BadStatusLine("bad"))), 1) + self.assertEqual(report, original) + self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "http://unsafe.invalid"}, report), 1) + self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "https://[bad/v1/status"}, report), 1) + self.assertEqual(health._send_status({"CI_FLEET_HEALTH_STATUS_URL": "https://status.example.invalid:bad/v1/status"}, report), 1) + legacy = {"CI_FLEET_HEALTH_HEARTBEAT_URL": "https://legacy.invalid"} + self.assertEqual(health._send_status(legacy, report), 0) + self.assertEqual(health._send_heartbeat(legacy, report, opener=opener), 0) + self.assertIsNone(health._NoRedirect().redirect_request(None, None, 302, None, {}, None)) + finally: + if old is None: + os.environ.pop("CI_FLEET_TESTING", None) + else: + os.environ["CI_FLEET_TESTING"] = old + def test_expired_active_resources_and_stopped_capacity_are_observable(self) -> None: cleanup = "KEEP container runner state=running expired=1 (routine cleanup never removes active containers)\nWOULD_REMOVE volume old expired=1\n" run = lambda args: health.subprocess.CompletedProcess(args, 0, cleanup, "") diff --git a/scripts/test_remote_reconcile.py b/scripts/test_remote_reconcile.py index c2308297..1348a9af 100644 --- a/scripts/test_remote_reconcile.py +++ b/scripts/test_remote_reconcile.py @@ -226,6 +226,39 @@ def test_reconcile_state_saved_on_failure(self): # Should exist even on failure self.assertTrue(state_path.exists() or result.returncode != 0) + def test_reconcile_failure_preserves_last_success_timestamp(self): + state_path = Path(self.env["CI_FLEET_RECONCILE_STATE_DIR"]) / "state.json" + state_path.parent.mkdir(parents=True) + state_path.write_text(json.dumps({ + "status": "converged", "desired_commit": "0" * 40, "applied_commit": "0" * 40, + "health": "healthy", "message": "ok", "checked_at": 777, "last_success_at": 777, + })) + subprocess.run([str(RECONCILE_SCRIPT)], capture_output=True, text=True, env=self.env) + self.assertEqual(json.loads(state_path.read_text())["last_success_at"], 777) + + def test_non_object_reconcile_state_is_recovered(self): + state_path = Path(self.env["CI_FLEET_RECONCILE_STATE_DIR"]) / "state.json" + state_path.parent.mkdir(parents=True) + state_path.write_text("[]\n") + subprocess.run([str(RECONCILE_SCRIPT)], capture_output=True, text=True, env=self.env) + state = json.loads(state_path.read_text()) + self.assertIsInstance(state, dict) + self.assertIsNone(state["last_success_at"]) + + def test_reconciliation_health_probe_suppresses_delivery(self): + content = RECONCILE_SCRIPT.read_text() + probe = content.split("run_health_check()", 1)[1].split("}", 1)[0] + self.assertIn("export CI_FLEET_HEALTH_SUPPRESS_DELIVERY=1", probe) + + def test_no_op_does_not_advance_last_success_timestamp(self): + content = RECONCILE_SCRIPT.read_text() + no_op = content.split('if [[ "$no_op" == true ]]; then', 2)[2].split("exit 0", 1)[0] + self.assertNotIn("'no change, converged' true", no_op) + self.assertIn('if sys.argv[7] == "true":', content) + self.assertEqual(content.count("save_reconcile_state 'converged'"), 4) + self.assertEqual(content.count("'no change, converged' true"), 1) + self.assertEqual(content.count('"reconciled to ${desired_commit}" true'), 1) + def test_validate_schema_output_no_secrets(self): """Sanitized log output must not contain actual key material or token values.""" result = subprocess.run( diff --git a/scripts/test_status_receiver.py b/scripts/test_status_receiver.py new file mode 100644 index 00000000..c756a55d --- /dev/null +++ b/scripts/test_status_receiver.py @@ -0,0 +1,401 @@ +#!/usr/bin/env python3 +import importlib.util +import json +import os +import sqlite3 +import sys +import tempfile +import threading +import time +import unittest +import urllib.request +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + + +def load(name: str): + spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py") + module = importlib.util.module_from_spec(spec) + assert spec.loader + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +status_auth = load("status_auth") +status_receiver = load("status_receiver") + + +def valid_report(controller: str = "example-ci-01", generated_at: int = 1_000) -> dict: + return { + "schema_version": 1, + "controller": { + "id": controller, + "software_version": "1" * 40, + "boot_time": 900, + "ssh": "disabled", + }, + "configuration": {"desired_commit": "2" * 40, "applied_commit": "2" * 40}, + "reconciliation": {"state": "converged", "last_success_at": 990}, + "drift": {"state": "ok"}, + "process": {"state": "running", "restart_count": 0}, + "timers": {name: "ok" for name in ("reconciliation", "drift", "health", "cleanup")}, + "runners": {"current": 0, "busy": 0, "maximum": 6}, + "metrics": { + "cpu": {"logical": 8, "used_percent": 25.0}, + "memory": {"total_bytes": 1024, "available_bytes": 768}, + "swap": {"total_bytes": 512, "used_bytes": 0}, + "disk": {name: {"total_bytes": 4096, "used_bytes": 1024} for name in ("root", "docker")}, + "inodes": {name: {"total": 1000, "used": 100} for name in ("root", "docker")}, + "load": {"one": 0.1, "five": 0.2, "fifteen": 0.3}, + }, + "docker": {"healthy": True, "oom": False}, + "error": None, + "generated_at": generated_at, + } + + +class StatusReceiverTests(unittest.TestCase): + def setUp(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.key = b"controller-key" + self.receiver = status_receiver.StatusReceiver( + Path(self.temporary.name) / "status.db", + {"example-ci-01": self.key, "other-ci-01": b"other-key"}, + read_token="reader-token", + history_limit=3, + retention_seconds=3_600, + min_interval_seconds=0, + ) + self.receiver._clock = lambda: 1_000 + self.assertEqual(self.receiver.database.stat().st_mode & 0o777, 0o600) + + def signed(self, report: dict, *, timestamp: int = 1_000, nonce: str = "a" * 32, + controller: str = "example-ci-01", key: bytes | None = None) -> tuple[bytes, dict[str, str]]: + body = json.dumps(report, separators=(",", ":"), sort_keys=True).encode() + return body, status_auth.sign_headers(controller, body, key or self.key, timestamp=timestamp, nonce=nonce) + + def submit(self, report: dict, **kwargs) -> None: + body, headers = self.signed(report, **kwargs) + self.receiver.submit(body, headers, now=kwargs.get("timestamp", 1_000)) + + def assert_status_error(self, status: int, code: str, call) -> None: + with self.assertRaises(status_receiver.StatusError) as caught: + call() + self.assertEqual((caught.exception.status, caught.exception.code), (status, code)) + + def test_authenticated_report_is_stored_and_read_as_latest(self) -> None: + report = valid_report() + self.submit(report) + self.assertEqual(self.receiver.latest("example-ci-01", "reader-token"), report) + + def test_concurrent_report_writes_are_serialized(self) -> None: + body, headers = self.signed( + valid_report("other-ci-01"), controller="other-ci-01", key=b"other-key" + ) + finished = threading.Event() + errors = [] + + def submit() -> None: + try: + self.receiver.submit(body, headers, now=1_000) + except Exception as error: + errors.append(error) + finally: + finished.set() + + self.receiver._write_lock.acquire() + thread = threading.Thread(target=submit) + thread.start() + try: + self.assertFalse(finished.wait(0.05)) + finally: + self.receiver._write_lock.release() + thread.join() + self.assertEqual(errors, []) + + def test_read_token_cannot_reuse_controller_key(self) -> None: + with self.assertRaisesRegex(ValueError, "read token"): + status_receiver.StatusReceiver( + Path(self.temporary.name) / "shared-secret.db", + {"example-ci-01": b"shared-secret"}, + read_token="shared-secret", + ) + + def test_duplicate_controller_keys_are_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "unique"): + status_receiver.StatusReceiver( + Path(self.temporary.name) / "duplicate.db", + {"example-ci-01": self.key, "other-ci-01": self.key}, + read_token="reader-token", + ) + + def test_retention_bounds_must_be_positive(self) -> None: + for values in ({"history_limit": 0}, {"retention_seconds": -1}): + with self.assertRaisesRegex(ValueError, "positive"): + status_receiver.StatusReceiver( + Path(self.temporary.name) / "invalid.db", {"example-ci-01": self.key}, + read_token="reader-token", **values, + ) + + def test_authentication_rejects_tampering_and_unknown_controller(self) -> None: + body, headers = self.signed(valid_report()) + self.assert_status_error(401, "authentication_failed", lambda: self.receiver.submit(body + b" ", headers, now=1_000)) + headers["X-CI-Fleet-Controller"] = "missing" + self.assert_status_error(401, "authentication_failed", lambda: self.receiver.submit(body, headers, now=1_000)) + + def test_controller_identity_isolation(self) -> None: + report = valid_report("other-ci-01") + body, headers = self.signed(report) + self.assert_status_error(403, "controller_identity_mismatch", lambda: self.receiver.submit(body, headers, now=1_000)) + + def test_replay_stale_authentication_and_stale_report_are_rejected(self) -> None: + report = valid_report() + body, headers = self.signed(report) + self.receiver.submit(body, headers, now=1_000) + self.assert_status_error(409, "replayed_report", lambda: self.receiver.submit(body, headers, now=1_000)) + old_body, old_headers = self.signed(valid_report(generated_at=1_001), timestamp=1_000, nonce="b" * 32) + self.assert_status_error(401, "authentication_stale", lambda: self.receiver.submit(old_body, old_headers, now=1_301)) + delayed = valid_report(generated_at=600) + delayed["controller"]["boot_time"] = 500 + delayed["reconciliation"]["last_success_at"] = 590 + delayed_body, delayed_headers = self.signed(delayed, timestamp=1_000, nonce="d" * 32) + self.assert_status_error(409, "report_time_stale", lambda: self.receiver.submit(delayed_body, delayed_headers, now=1_000)) + stale_body, stale_headers = self.signed(valid_report(), timestamp=1_001, nonce="c" * 32) + self.assert_status_error(409, "stale_report", lambda: self.receiver.submit(stale_body, stale_headers, now=1_001)) + + def test_rejected_submissions_close_database_connections(self) -> None: + connections = [] + + class TrackingConnection(sqlite3.Connection): + closed = False + + def close(self) -> None: + self.closed = True + super().close() + + def connect() -> sqlite3.Connection: + connection = sqlite3.connect(self.receiver.database, factory=TrackingConnection) + connections.append(connection) + return connection + + self.receiver._connect = connect + body, headers = self.signed(valid_report()) + self.receiver.submit(body, headers, now=1_000) + self.assert_status_error(409, "replayed_report", lambda: self.receiver.submit(body, headers, now=1_000)) + self.assertEqual(len(connections), 2) + self.assertTrue(all(connection.closed for connection in connections)) + + def test_payload_and_submission_frequency_are_bounded(self) -> None: + small = status_receiver.StatusReceiver( + Path(self.temporary.name) / "small.db", {"example-ci-01": self.key}, + read_token="reader-token", max_payload_bytes=16, min_interval_seconds=0, + ) + body, headers = self.signed(valid_report()) + self.assert_status_error(413, "payload_too_large", lambda: small.submit(body, headers, now=1_000)) + + limited = status_receiver.StatusReceiver( + Path(self.temporary.name) / "limited.db", {"example-ci-01": self.key}, + read_token="reader-token", min_interval_seconds=30, + ) + limited.submit(body, headers, now=1_000) + second_body, second_headers = self.signed(valid_report(generated_at=1_001), timestamp=1_001, nonce="b" * 32) + self.assert_status_error(429, "submission_too_frequent", lambda: limited.submit(second_body, second_headers, now=1_001)) + third_body, third_headers = self.signed(valid_report(generated_at=1_002), timestamp=1_002, nonce="c" * 32) + self.assert_status_error(429, "submission_too_frequent", lambda: limited.submit(third_body, third_headers, now=1_001)) + with sqlite3.connect(limited.database) as connection: + self.assertEqual(connection.execute("SELECT COUNT(*) FROM nonces").fetchone()[0], 2) + self.assert_status_error(409, "replayed_report", lambda: limited.submit(second_body, second_headers, now=1_031)) + + def test_schema_compatibility_and_malformed_metrics(self) -> None: + future = valid_report() + future["schema_version"] = 2 + body, headers = self.signed(future) + self.assert_status_error(400, "unsupported_schema", lambda: self.receiver.submit(body, headers, now=1_000)) + + boolean = valid_report() + boolean["schema_version"] = True + body, headers = self.signed(boolean, nonce="f" * 32) + self.assert_status_error(400, "unsupported_schema", lambda: self.receiver.submit(body, headers, now=1_000)) + + for mutate in ( + lambda report: report["controller"].update(ssh={}), + lambda report: report["reconciliation"].update(state=[]), + lambda report: report["drift"].update(state={}), + lambda report: report["process"].update(state=[]), + lambda report: report["timers"].update(health={}), + ): + malformed = valid_report() + mutate(malformed) + body, headers = self.signed(malformed, nonce="e" * 32) + self.assert_status_error(400, "invalid_report", lambda body=body, headers=headers: self.receiver.submit(body, headers, now=1_000)) + + malformed = valid_report() + malformed["metrics"]["memory"]["available_bytes"] = -1 + body, headers = self.signed(malformed, nonce="b" * 32) + self.assert_status_error(400, "invalid_report", lambda: self.receiver.submit(body, headers, now=1_000)) + + extra = valid_report() + extra["secret"] = "must not be accepted" + body, headers = self.signed(extra, nonce="c" * 32) + self.assert_status_error(400, "invalid_report", lambda: self.receiver.submit(body, headers, now=1_000)) + + def test_history_and_retention_are_bounded(self) -> None: + for offset, nonce in enumerate(("a", "b", "c", "d")): + generated = 1_000 + offset + self.submit(valid_report(generated_at=generated), timestamp=generated, nonce=nonce * 32) + self.assertEqual([item["generated_at"] for item in self.receiver.history("example-ci-01", "reader-token")], [1_003, 1_002, 1_001]) + + self.submit(valid_report(generated_at=5_000), timestamp=5_000, nonce="e" * 32) + self.assertEqual([item["generated_at"] for item in self.receiver.history("example-ci-01", "reader-token")], [5_000]) + latest, history = self.receiver.latest_and_history("example-ci-01", "reader-token") + self.assertEqual(latest, history[0]) + + def test_smaller_history_limit_is_applied_on_restart(self) -> None: + database = Path(self.temporary.name) / "smaller-history.db" + receiver = status_receiver.StatusReceiver( + database, {"example-ci-01": self.key}, read_token="reader-token", + history_limit=3, retention_seconds=3_600, min_interval_seconds=0, + ) + now = int(time.time()) + for offset, nonce in enumerate(("a", "b", "c")): + body, headers = self.signed(valid_report(generated_at=now + offset), timestamp=now + offset, nonce=nonce * 32) + receiver.submit(body, headers, now=now + offset) + status_receiver.StatusReceiver( + database, {"example-ci-01": self.key}, read_token="reader-token", + history_limit=1, retention_seconds=3_600, min_interval_seconds=0, + ) + with sqlite3.connect(database) as connection: + self.assertEqual(connection.execute("SELECT COUNT(*) FROM reports").fetchone()[0], 1) + + def test_time_retention_is_enforced_without_traffic(self) -> None: + receiver = status_receiver.StatusReceiver( + Path(self.temporary.name) / "expiry.db", {"example-ci-01": self.key}, + read_token="reader-token", retention_seconds=10, min_interval_seconds=0, + ) + body, headers = self.signed(valid_report()) + receiver.submit(body, headers, now=1_000) + receiver._clock = lambda: 1_011 + stop = threading.Event() + thread = threading.Thread(target=status_receiver._expiration_loop, args=(receiver, stop, 0.01)) + thread.start() + try: + deadline = time.time() + 1 + while time.time() < deadline: + with sqlite3.connect(receiver.database) as connection: + if connection.execute("SELECT COUNT(*) FROM reports").fetchone()[0] == 0: + break + time.sleep(0.01) + else: + self.fail("expired report remained without request traffic") + finally: + stop.set() + thread.join() + + def test_retention_worker_retries_transient_database_errors(self) -> None: + stop = threading.Event() + + class FlakyReceiver: + calls = 0 + + def expire(self) -> None: + self.calls += 1 + if self.calls == 1: + raise sqlite3.OperationalError("locked") + stop.set() + + receiver = FlakyReceiver() + status_receiver._expiration_loop(receiver, stop, 0.001) + self.assertEqual(receiver.calls, 2) + + def test_read_token_must_be_http_header_safe(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + key_file, token_file, config_file = root / "key", root / "token", root / "auth.json" + key_file.write_bytes(b"k" * 32) + config_file.write_text(json.dumps({"controllers": {"example-ci-01": "key"}, "read_token_file": "token"})) + for path in (key_file, token_file, config_file): + path.touch(exist_ok=True) + path.chmod(0o600) + for value in ("€" * 32, "a" * 32 + "\n" + "b" * 32): + token_file.write_text(value) + with self.assertRaisesRegex(ValueError, "visible ASCII"): + status_receiver.load_auth_config(config_file) + + def test_receiver_restart_reloads_rotated_key(self) -> None: + directory = Path(self.temporary.name) + key_file, token_file, config_file = directory / "key", directory / "token", directory / "auth.json" + raw_key = b"\n" + b"a" * 30 + b" " + key_file.write_bytes(raw_key) + token_file.write_bytes(b"r" * 32) + config_file.write_text(json.dumps({"controllers": {"example-ci-01": "key"}, "read_token_file": "token"})) + for path in (key_file, token_file, config_file): + path.chmod(0o600) + first, _ = status_receiver.load_auth_config(config_file) + key_file.write_bytes(b"b" * 32) + second, _ = status_receiver.load_auth_config(config_file) + self.assertEqual((first["example-ci-01"], second["example-ci-01"]), (raw_key, b"b" * 32)) + + def test_http_post_and_read_only_api(self) -> None: + server = status_receiver.create_server("127.0.0.1", 0, self.receiver) + thread = threading.Thread(target=server.serve_forever) + thread.start() + self.addCleanup(thread.join) + self.addCleanup(server.server_close) + self.addCleanup(server.shutdown) + base = f"http://127.0.0.1:{server.server_port}" + now = int(time.time()) + report = valid_report(generated_at=now) + report["controller"]["boot_time"] = now - 100 + report["reconciliation"]["last_success_at"] = now - 10 + body, headers = self.signed(report, timestamp=now) + request = urllib.request.Request(base + "/v1/status", data=body, headers=headers, method="POST") + with urllib.request.urlopen(request) as response: + self.assertEqual(response.status, 202) + request = urllib.request.Request(base + "/v1/controllers", headers={"Authorization": "Bearer reader-token"}) + with urllib.request.urlopen(request) as response: + payload = json.load(response) + self.assertEqual(payload, {"schema_version": 1, "controllers": [report]}) + request = urllib.request.Request(base + "/v1/controllers/example-ci-01", headers={"Authorization": "Bearer reader-token"}) + with urllib.request.urlopen(request) as response: + payload = json.load(response) + self.assertEqual(payload["latest"], report) + self.assertEqual(payload["history"], [report]) + + def test_read_api_authentication_and_controller_listing(self) -> None: + self.submit(valid_report()) + self.assert_status_error(401, "read_authentication_failed", lambda: self.receiver.latest("example-ci-01", "wrong")) + self.assert_status_error(401, "read_authentication_failed", lambda: self.receiver.latest("example-ci-01", "tök")) + self.assertEqual(self.receiver.list_latest("reader-token"), [valid_report()]) + + def test_http_server_bounds_slow_clients(self) -> None: + server = status_receiver.create_server("127.0.0.1", 0, self.receiver) + self.addCleanup(server.server_close) + self.assertEqual((server.max_requests, server.request_timeout), (32, 15)) + for _ in range(server.max_requests): + self.assertTrue(server._slots.acquire(blocking=False)) + self.assertFalse(server._slots.acquire(blocking=False)) + for _ in range(server.max_requests): + server._slots.release() + + expired = threading.Event() + class SlowRequest: + def shutdown(self, how: int) -> None: + self.how = how + expired.set() + request = SlowRequest() + timer = threading.Timer(0.01, server._expire_request, (request,)) + timer.start() + self.assertTrue(expired.wait(1)) + self.assertEqual(request.how, status_receiver.socket.SHUT_RDWR) + + if status_receiver.socket.has_ipv6: + ipv6 = status_receiver.create_server("::1", 0, self.receiver) + self.addCleanup(ipv6.server_close) + self.assertEqual(ipv6.address_family, status_receiver.socket.AF_INET6) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/validate.sh b/scripts/validate.sh index 8a1f94e9..56f2fa8d 100755 --- a/scripts/validate.sh +++ b/scripts/validate.sh @@ -10,14 +10,19 @@ python3 -m py_compile \ .github/actions/plan/test_plan.py \ scripts/desired_state.py \ scripts/health.py \ + scripts/status_auth.py \ + scripts/status_receiver.py \ scripts/scan_committed_secrets.py \ scripts/test_desired_state.py \ scripts/test_health.py \ + scripts/test_status_receiver.py \ scripts/test_quickstart.py python3 .github/actions/plan/test_plan.py python3 scripts/test_desired_state.py python3 scripts/test_health.py +python3 scripts/test_status_receiver.py python3 scripts/test_quickstart.py +python3 -m json.tool schemas/status-report-v1.json >/dev/null python3 .github/actions/plan/plan.py --plan examples/project/scripts/ci/plan.json --group fast >/dev/null python3 .github/actions/plan/plan.py --plan examples/project/scripts/ci/plan.json --group full >/dev/null scripts/test-capacity-preflight.sh