diff --git a/internal/build/build.go b/internal/build/build.go new file mode 100644 index 0000000..89ff26f --- /dev/null +++ b/internal/build/build.go @@ -0,0 +1,505 @@ +// Package build implements the image build subsystem (spec §16) — "the +// platform's biggest security surface". A SysAdmin uploads a Dockerfile and a +// context tarball; felis-api starts an in-cluster Kaniko Job that builds and +// pushes to the internal registry, after which a Trivy scan gates admission to +// the image whitelist. +// +// Trust model (spec §16, §22): we trust the SysAdmin at the *ingress* (only an +// admin through Zero Trust may submit a build) but never trust the *Dockerfile +// at runtime* — an arbitrary Dockerfile is build-time RCE whose victim is the +// cluster, not the uploader. So the build Pod runs with a deliberately weak +// service account in an isolated namespace that can only push to the registry +// and cannot touch the minecraft namespace, the felis database, or the K8s API +// (spec §21). Those isolation guarantees live in the Job/NetworkPolicy specs +// (jobspec.go) and are asserted by unit tests, since no cluster runs here. +// +// The Trivy gate is enforced as the build Pod's *exit code*: a kaniko +// initContainer builds and pushes, then a trivy container scans the pushed ref +// with `--exit-code 1 --severity CRITICAL`. Therefore "Job Succeeded" is +// equivalent to "pushed AND no CRITICAL CVE". felis-api observes the Job phase +// and performs the database writes — the build Pod itself never has database +// credentials (the weak-SA red line). On success the image is admitted to +// image_whitelist with enabled=true (recording added_by); on failure the build +// is marked failed and nothing is admitted (spec §16: the only retained +// automatic gate). +// +// The Builder depends on the Store and Jobs interfaces, so submission, the +// scan-gate translation, cancellation, and image admission are all unit-tested +// against in-memory fakes. The Postgres (pgStore) and controller-runtime +// (k8sJobs) implementations compile here but are exercised only by integration +// tests against a live database / cluster. +package build + +import ( + "context" + "errors" + "fmt" + "strings" + "time" +) + +// Status mirrors the build_status enum (spec §6). +type Status string + +const ( + StatusPending Status = "pending" + StatusBuilding Status = "building" + StatusSucceeded Status = "succeeded" + StatusFailed Status = "failed" + StatusCancelled Status = "cancelled" +) + +// terminal reports whether a status is final and no longer reconciled. +func (s Status) terminal() bool { + switch s { + case StatusSucceeded, StatusFailed, StatusCancelled: + return true + default: + return false + } +} + +// JobPhase is the build Pod's lifecycle as observed from the K8s Job, decoupled +// from any K8s type so the scan-gate translation stays unit-testable. +type JobPhase int + +const ( + // JobUnknown means the Job was not found (e.g. GC'd); treated as failed. + JobUnknown JobPhase = iota + JobPending + JobRunning + // JobSucceeded means kaniko pushed AND trivy found no CRITICAL CVE — the + // scan gate passed (spec §16). + JobSucceeded + // JobFailed means kaniko failed OR trivy found a CRITICAL CVE — the build + // is rejected and nothing is admitted. + JobFailed +) + +// image admission sources (spec §6 image_whitelist.source). +const ( + SourceBuilt = "built" + SourceExternal = "external" +) + +// ErrNotFound is returned when a build id / image ref does not exist. +var ErrNotFound = errors.New("build: not found") + +// ErrAlreadyTerminal is returned by Cancel when the build has already finished. +var ErrAlreadyTerminal = errors.New("build: already in a terminal state") + +// ErrInvalid wraps every request-validation failure (bad image ref, missing / +// oversize Dockerfile, missing context). Callers map it to a 400; it is kept +// distinct from store/cluster failures so those surface as 500. +var ErrInvalid = errors.New("build: invalid request") + +// Request is the validated POST /images/build input (spec §16). The dockerfile +// and context are archived for audit; the target ref must address the internal +// registry (enforced in Validate). +type Request struct { + // ImageRef is the push target, e.g. registry.felis.svc:5000/foo:1.0. It must + // be under the configured internal registry — a build can never push + // elsewhere. + ImageRef string + // Dockerfile is the uploaded build recipe (size-capped). + Dockerfile string + // ContextRef locates the uploaded tar.gz context in object storage / a PVC + // (spec §17: Kaniko pulls it; Git context is intentionally not supported). + ContextRef string + // BaseImage is the resolved FROM, recorded for audit only — it is NOT a hard + // gate (spec §16: base FROM is not hard-gated; the scan + egress lock cover + // poisoned bases). + BaseImage string + // RequestedBy is the admin Access email, used as the audit actor and the + // added_by of any admitted image. + RequestedBy string +} + +// Build mirrors an image_builds row (spec §6). +type Build struct { + ID string `json:"id"` + ImageRef string `json:"image_ref"` + Status Status `json:"status"` + Dockerfile string `json:"dockerfile,omitempty"` + ContextRef string `json:"context_ref,omitempty"` + BaseImage string `json:"base_image,omitempty"` + RequestedBy string `json:"requested_by"` + JobName string `json:"job_name,omitempty"` + LogRef string `json:"log_ref,omitempty"` + Error string `json:"error,omitempty"` + CreatedAt time.Time `json:"created_at"` + FinishedAt *time.Time `json:"finished_at,omitempty"` +} + +// Image mirrors an image_whitelist row (spec §6): the dynamic, auditable image +// admission list that the create-server form reads from. +type Image struct { + ImageRef string `json:"image_ref"` + Source string `json:"source"` + BuildID string `json:"build_id,omitempty"` + AddedBy string `json:"added_by"` + Enabled bool `json:"enabled"` + AddedAt time.Time `json:"added_at"` +} + +// Store is the business-layer persistence the Builder depends on (image_builds +// + image_whitelist). It is an interface so the Builder is tested against an +// in-memory fake; the Postgres implementation (pgStore) is integration-tested +// only. +type Store interface { + // CreateBuild inserts a new image_builds row (status pending). + CreateBuild(ctx context.Context, b *Build) error + // GetBuild loads one build, or ErrNotFound. + GetBuild(ctx context.Context, id string) (*Build, error) + // SetBuildJob records the Job name and advances status to building. + SetBuildJob(ctx context.Context, id, jobName string) error + // FinishBuild sets a terminal status, an optional error, and finished_at. + FinishBuild(ctx context.Context, id string, status Status, errMsg string, at time.Time) error + // ListUnfinishedBuilds returns builds still being reconciled (status pending + // or building), oldest first — the work list for SyncAll. + ListUnfinishedBuilds(ctx context.Context) ([]Build, error) + // AdmitBuiltImage upserts an image_whitelist row with enabled=true and + // source=built (the scan-gate success path, spec §16). It records added_by. + AdmitBuiltImage(ctx context.Context, img Image) error + // ListImages returns the image whitelist. + ListImages(ctx context.Context) ([]Image, error) + // AddExternalImage upserts an externally-pushed image (spec §15 external + // admission; source=external, no build_id). + AddExternalImage(ctx context.Context, img Image) error + // RemoveImage deletes an image_whitelist row, or ErrNotFound. + RemoveImage(ctx context.Context, imageRef string) error +} + +// Jobs is the cluster-side build lifecycle the Builder depends on. It is an +// interface so the scan-gate translation is tested against a fake; the +// controller-runtime implementation (k8sJobs) is integration-tested only — it +// requires a live cluster. +type Jobs interface { + // CreateBuildJob starts the Kaniko+Trivy Job for p in the felis-build + // namespace and returns the Job name. + CreateBuildJob(ctx context.Context, p JobParams) (jobName string, err error) + // JobPhase reports the current phase of a previously-created Job. + JobPhase(ctx context.Context, jobName string) (JobPhase, error) + // CancelBuildJob deletes the Job (and its pods), tolerating not-found. + CancelBuildJob(ctx context.Context, jobName string) error +} + +// Config parameterises the build subsystem from felis.toml (spec §24 [registry] +// + safety limits). It is validated by withDefaults before use. +type Config struct { + // Namespace is the isolated build namespace (spec §16: felis-build). + Namespace string + // ServiceAccount is the weak SA the build Pod runs as. It MUST NOT be the + // felis-api SA (spec §16 red line). + ServiceAccount string + // RegistryURL is the internal registry the build pushes to and Trivy scans + // (spec §17). Image refs are validated to be under it. + RegistryURL string + // KanikoImage / TrivyImage are the executor images. + KanikoImage string + TrivyImage string + // Deadline caps a build's wall-clock (spec §16: activeDeadlineSeconds). + Deadline time.Duration + // MaxDockerfileBytes caps the uploaded Dockerfile (spec §16: context size + // limits). Zero applies the default. + MaxDockerfileBytes int + // CPULimit / MemLimit cap each build container (spec §16: resource limits). + CPULimit string + MemLimit string +} + +// Defaults applied when a Config field is left zero. +const ( + defaultNamespace = "felis-build" + defaultServiceAccount = "felis-build" + defaultKanikoImage = "gcr.io/kaniko-project/executor:latest" + defaultTrivyImage = "aquasec/trivy:latest" + defaultDeadline = 30 * time.Minute + defaultMaxDockerfile = 256 * 1024 // 256 KiB + defaultCPULimit = "2" + defaultMemLimit = "4Gi" +) + +// withDefaults returns a copy of c with zero fields filled, so a partially +// configured Config (or the zero value, in tests) is always usable. +func (c Config) withDefaults() Config { + if c.Namespace == "" { + c.Namespace = defaultNamespace + } + if c.ServiceAccount == "" { + c.ServiceAccount = defaultServiceAccount + } + if c.KanikoImage == "" { + c.KanikoImage = defaultKanikoImage + } + if c.TrivyImage == "" { + c.TrivyImage = defaultTrivyImage + } + if c.Deadline <= 0 { + c.Deadline = defaultDeadline + } + if c.MaxDockerfileBytes <= 0 { + c.MaxDockerfileBytes = defaultMaxDockerfile + } + if c.CPULimit == "" { + c.CPULimit = defaultCPULimit + } + if c.MemLimit == "" { + c.MemLimit = defaultMemLimit + } + return c +} + +// Builder orchestrates the build subsystem. It holds no mutable state; the +// clock and id generator are injectable for hermetic tests. +type Builder struct { + Store Store + Jobs Jobs + Config Config + + // Now is the clock, injectable for tests. Defaults to time.Now. + Now func() time.Time + // IDGen mints build ids. Defaults to a time-based generator. + IDGen func() string +} + +func (b *Builder) now() time.Time { + if b.Now != nil { + return b.Now() + } + return time.Now() +} + +func (b *Builder) newID() string { + if b.IDGen != nil { + return b.IDGen() + } + return fmt.Sprintf("bld-%d", time.Now().UnixNano()) +} + +// Submit validates req, records a pending build, and starts the Kaniko+Trivy +// Job (spec §16). The build is returned in the building state once the Job is +// created; if Job creation fails the build is marked failed so it never lingers +// pending. The caller (felis-api) drives the build to a terminal state by +// polling Sync / SyncAll. +func (b *Builder) Submit(ctx context.Context, req Request) (*Build, error) { + cfg := b.Config.withDefaults() + if err := Validate(req, cfg); err != nil { + return nil, err + } + + now := b.now() + bld := &Build{ + ID: b.newID(), + ImageRef: req.ImageRef, + Status: StatusPending, + Dockerfile: req.Dockerfile, + ContextRef: req.ContextRef, + BaseImage: req.BaseImage, + RequestedBy: req.RequestedBy, + CreatedAt: now, + } + if err := b.Store.CreateBuild(ctx, bld); err != nil { + return nil, err + } + + jobName, err := b.Jobs.CreateBuildJob(ctx, b.jobParams(bld, cfg)) + if err != nil { + // The pending row exists; mark it failed so it is not reconciled forever. + _ = b.Store.FinishBuild(ctx, bld.ID, StatusFailed, "job creation failed: "+err.Error(), b.now()) + bld.Status = StatusFailed + bld.Error = "job creation failed: " + err.Error() + return bld, fmt.Errorf("build: create job: %w", err) + } + + if err := b.Store.SetBuildJob(ctx, bld.ID, jobName); err != nil { + return nil, err + } + bld.JobName = jobName + bld.Status = StatusBuilding + return bld, nil +} + +// jobParams projects a build + config onto the inputs jobspec.go renders. +func (b *Builder) jobParams(bld *Build, cfg Config) JobParams { + return JobParams{ + BuildID: bld.ID, + ImageRef: bld.ImageRef, + ContextRef: bld.ContextRef, + Namespace: cfg.Namespace, + ServiceAccount: cfg.ServiceAccount, + RegistryURL: cfg.RegistryURL, + KanikoImage: cfg.KanikoImage, + TrivyImage: cfg.TrivyImage, + Deadline: cfg.Deadline, + CPULimit: cfg.CPULimit, + MemLimit: cfg.MemLimit, + } +} + +// Get returns a build by id, or ErrNotFound. +func (b *Builder) Get(ctx context.Context, id string) (*Build, error) { + return b.Store.GetBuild(ctx, id) +} + +// Sync reconciles one non-terminal build against its Job phase — the scan-gate +// translation (spec §16). A terminal build is returned unchanged (idempotent). +// +// - JobSucceeded → status=succeeded AND the image is admitted to the whitelist +// with enabled=true (kaniko pushed and trivy found no CRITICAL CVE). +// - JobFailed / JobUnknown → status=failed, nothing admitted (a CRITICAL CVE +// surfaces here as a failed Job, since trivy runs with --exit-code 1). +// - JobPending / JobRunning → no change. +// +// The image admission is performed by felis-api (this code path), never by the +// build Pod, which holds no database credentials. +func (b *Builder) Sync(ctx context.Context, id string) (*Build, error) { + bld, err := b.Store.GetBuild(ctx, id) + if err != nil { + return nil, err + } + if bld.Status.terminal() { + return bld, nil + } + if bld.JobName == "" { + // Created but the Job name was never recorded; treat as failed rather + // than reconcile forever against a phantom Job. + return b.finish(ctx, bld, StatusFailed, "no build job recorded") + } + + phase, err := b.Jobs.JobPhase(ctx, bld.JobName) + if err != nil { + return nil, err + } + switch phase { + case JobSucceeded: + now := b.now() + // Admit the image first; only then mark the build succeeded, so a + // succeeded build always has its whitelist row (no admitted-but-not- + // recorded window if the second write fails). + if err := b.Store.AdmitBuiltImage(ctx, Image{ + ImageRef: bld.ImageRef, + Source: SourceBuilt, + BuildID: bld.ID, + AddedBy: bld.RequestedBy, + Enabled: true, + AddedAt: now, + }); err != nil { + return nil, err + } + return b.finishAt(ctx, bld, StatusSucceeded, "", now) + case JobFailed, JobUnknown: + return b.finish(ctx, bld, StatusFailed, "build job failed or scan found a CRITICAL CVE") + default: // JobPending / JobRunning + return bld, nil + } +} + +// SyncAll reconciles every unfinished build and returns the count advanced to a +// terminal state. felis-api calls this periodically (spec §16: the scan gate is +// observed, not pushed by the build Pod). +func (b *Builder) SyncAll(ctx context.Context) (int, error) { + builds, err := b.Store.ListUnfinishedBuilds(ctx) + if err != nil { + return 0, err + } + advanced := 0 + for i := range builds { + bld, err := b.Sync(ctx, builds[i].ID) + if err != nil { + return advanced, err + } + if bld.Status.terminal() { + advanced++ + } + } + return advanced, nil +} + +// Cancel stops an in-flight build: delete its Job and mark it cancelled. A +// build that has already finished returns ErrAlreadyTerminal. +func (b *Builder) Cancel(ctx context.Context, id string) (*Build, error) { + bld, err := b.Store.GetBuild(ctx, id) + if err != nil { + return nil, err + } + if bld.Status.terminal() { + return nil, ErrAlreadyTerminal + } + if bld.JobName != "" { + if err := b.Jobs.CancelBuildJob(ctx, bld.JobName); err != nil { + return nil, err + } + } + return b.finish(ctx, bld, StatusCancelled, "cancelled by administrator") +} + +// finish marks a build terminal at the current clock and returns the updated +// view without a second round-trip. +func (b *Builder) finish(ctx context.Context, bld *Build, status Status, msg string) (*Build, error) { + return b.finishAt(ctx, bld, status, msg, b.now()) +} + +func (b *Builder) finishAt(ctx context.Context, bld *Build, status Status, msg string, at time.Time) (*Build, error) { + if err := b.Store.FinishBuild(ctx, bld.ID, status, msg, at); err != nil { + return nil, err + } + bld.Status = status + bld.Error = msg + finished := at + bld.FinishedAt = &finished + return bld, nil +} + +// ListImages returns the image whitelist (spec §15 create-server form source). +func (b *Builder) ListImages(ctx context.Context) ([]Image, error) { + return b.Store.ListImages(ctx) +} + +// AddExternalImage admits an externally-pushed image (spec §15). It is enabled +// immediately; external images bypass the build pipeline but are still recorded +// with added_by for audit. +func (b *Builder) AddExternalImage(ctx context.Context, imageRef, addedBy string) (*Image, error) { + if err := ValidateImageRef(imageRef); err != nil { + return nil, err + } + img := Image{ + ImageRef: imageRef, + Source: SourceExternal, + AddedBy: addedBy, + Enabled: true, + AddedAt: b.now(), + } + if err := b.Store.AddExternalImage(ctx, img); err != nil { + return nil, err + } + return &img, nil +} + +// RemoveImage withdraws an image from the whitelist (spec §22: dynamic, +// auditable). It does not delete the underlying registry blob. +func (b *Builder) RemoveImage(ctx context.Context, imageRef string) error { + return b.Store.RemoveImage(ctx, imageRef) +} + +// ImageAdmitted reports whether a concrete image reference is on the whitelist +// and enabled (spec §15: the create-server form may only choose an admitted +// image). A disabled row never admits. A wildcard whitelist entry +// ("registry/foo:*") admits any concrete tag on that repo (imageMatches); the +// caller always passes a concrete ref, never a wildcard. An empty ref is never +// admitted. +func (b *Builder) ImageAdmitted(ctx context.Context, imageRef string) (bool, error) { + if strings.TrimSpace(imageRef) == "" { + return false, nil + } + images, err := b.Store.ListImages(ctx) + if err != nil { + return false, err + } + for _, img := range images { + if img.Enabled && imageMatches(imageRef, img.ImageRef) { + return true, nil + } + } + return false, nil +} diff --git a/internal/build/build_test.go b/internal/build/build_test.go new file mode 100644 index 0000000..2d0b859 --- /dev/null +++ b/internal/build/build_test.go @@ -0,0 +1,478 @@ +package build + +import ( + "context" + "errors" + "testing" + "time" +) + +var testNow = time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + +// ---- in-memory fakes ---- + +type fakeStore struct { + builds map[string]*Build + images map[string]Image + // call recorders + admitted []Image + finished []string // "id:status" + removeErr error + createErr error +} + +func newFakeStore() *fakeStore { + return &fakeStore{builds: map[string]*Build{}, images: map[string]Image{}} +} + +func (f *fakeStore) CreateBuild(_ context.Context, b *Build) error { + if f.createErr != nil { + return f.createErr + } + cp := *b + f.builds[b.ID] = &cp + return nil +} + +func (f *fakeStore) GetBuild(_ context.Context, id string) (*Build, error) { + b, ok := f.builds[id] + if !ok { + return nil, ErrNotFound + } + cp := *b + return &cp, nil +} + +func (f *fakeStore) SetBuildJob(_ context.Context, id, jobName string) error { + b, ok := f.builds[id] + if !ok { + return ErrNotFound + } + b.JobName = jobName + b.Status = StatusBuilding + return nil +} + +func (f *fakeStore) FinishBuild(_ context.Context, id string, status Status, errMsg string, at time.Time) error { + b, ok := f.builds[id] + if !ok { + return ErrNotFound + } + b.Status = status + b.Error = errMsg + finished := at + b.FinishedAt = &finished + f.finished = append(f.finished, id+":"+string(status)) + return nil +} + +func (f *fakeStore) ListUnfinishedBuilds(_ context.Context) ([]Build, error) { + var out []Build + for _, b := range f.builds { + if !b.Status.terminal() { + out = append(out, *b) + } + } + return out, nil +} + +func (f *fakeStore) AdmitBuiltImage(_ context.Context, img Image) error { + f.images[img.ImageRef] = img + f.admitted = append(f.admitted, img) + return nil +} + +func (f *fakeStore) ListImages(_ context.Context) ([]Image, error) { + var out []Image + for _, img := range f.images { + out = append(out, img) + } + return out, nil +} + +func (f *fakeStore) AddExternalImage(_ context.Context, img Image) error { + f.images[img.ImageRef] = img + return nil +} + +func (f *fakeStore) RemoveImage(_ context.Context, ref string) error { + if f.removeErr != nil { + return f.removeErr + } + if _, ok := f.images[ref]; !ok { + return ErrNotFound + } + delete(f.images, ref) + return nil +} + +type fakeJobs struct { + phase JobPhase + phaseErr error + createErr error + + created []JobParams + cancelled []string +} + +func (f *fakeJobs) CreateBuildJob(_ context.Context, p JobParams) (string, error) { + if f.createErr != nil { + return "", f.createErr + } + f.created = append(f.created, p) + return BuildJobName(p.BuildID), nil +} + +func (f *fakeJobs) JobPhase(_ context.Context, _ string) (JobPhase, error) { + return f.phase, f.phaseErr +} + +func (f *fakeJobs) CancelBuildJob(_ context.Context, jobName string) error { + f.cancelled = append(f.cancelled, jobName) + return nil +} + +// newBuilder wires a Builder over fresh fakes with a frozen clock and a +// deterministic id generator. +func newBuilder() (*Builder, *fakeStore, *fakeJobs) { + st := newFakeStore() + jb := &fakeJobs{phase: JobPending} + n := 0 + b := &Builder{ + Store: st, + Jobs: jb, + Config: Config{ + RegistryURL: "registry.felis.svc:5000", + }, + Now: func() time.Time { return testNow }, + IDGen: func() string { + n++ + return "bld-" + string(rune('0'+n)) + }, + } + return b, st, jb +} + +func goodRequest() Request { + return Request{ + ImageRef: "registry.felis.svc:5000/mc-paper:1.0", + Dockerfile: "FROM eclipse-temurin:21\nCOPY . /data\n", + ContextRef: "tar://contexts/abc.tar.gz", + RequestedBy: "admin@example.net", + } +} + +// ---- submission ---- + +func TestSubmitCreatesPendingBuildAndStartsJob(t *testing.T) { + b, st, jb := newBuilder() + bld, err := b.Submit(context.Background(), goodRequest()) + if err != nil { + t.Fatalf("Submit: %v", err) + } + if bld.Status != StatusBuilding { + t.Errorf("status = %q, want building", bld.Status) + } + if bld.JobName == "" { + t.Error("job name not recorded") + } + if got := st.builds[bld.ID]; got == nil || got.Dockerfile == "" { + t.Error("build row not persisted with dockerfile") + } + if len(jb.created) != 1 { + t.Fatalf("CreateBuildJob calls = %d, want 1", len(jb.created)) + } + // The Job must be parameterised with the weak build SA and the namespace, + // never the api identity — this is the §16 red line, asserted at the seam. + p := jb.created[0] + if p.ServiceAccount != defaultServiceAccount { + t.Errorf("job SA = %q, want %q", p.ServiceAccount, defaultServiceAccount) + } + if p.Namespace != defaultNamespace { + t.Errorf("job namespace = %q, want %q", p.Namespace, defaultNamespace) + } + if p.RegistryURL != "registry.felis.svc:5000" { + t.Errorf("job registry = %q", p.RegistryURL) + } + if p.Deadline <= 0 { + t.Error("job deadline not set") + } +} + +func TestSubmitRejectsExternalRegistryTarget(t *testing.T) { + b, st, jb := newBuilder() + req := goodRequest() + req.ImageRef = "docker.io/library/evil:latest" + if _, err := b.Submit(context.Background(), req); err == nil { + t.Fatal("expected rejection of non-internal registry target") + } + if len(st.builds) != 0 { + t.Error("a rejected build must not be persisted") + } + if len(jb.created) != 0 { + t.Error("a rejected build must not start a job") + } +} + +func TestSubmitRejectsEmptyAndOversizeDockerfile(t *testing.T) { + b, _, _ := newBuilder() + req := goodRequest() + req.Dockerfile = " " + if _, err := b.Submit(context.Background(), req); err == nil { + t.Error("expected rejection of empty dockerfile") + } + + b2, _, _ := newBuilder() + b2.Config.MaxDockerfileBytes = 16 + req2 := goodRequest() + req2.Dockerfile = "FROM scratch\n# padding padding padding padding\n" + if _, err := b2.Submit(context.Background(), req2); err == nil { + t.Error("expected rejection of oversize dockerfile") + } +} + +func TestSubmitRequiresContext(t *testing.T) { + b, _, _ := newBuilder() + req := goodRequest() + req.ContextRef = "" + if _, err := b.Submit(context.Background(), req); err == nil { + t.Error("expected rejection when context reference is missing") + } +} + +func TestSubmitMarksBuildFailedWhenJobCreationFails(t *testing.T) { + b, st, jb := newBuilder() + jb.createErr = errors.New("apiserver down") + bld, err := b.Submit(context.Background(), goodRequest()) + if err == nil { + t.Fatal("expected error when job creation fails") + } + if bld == nil || bld.Status != StatusFailed { + t.Fatalf("build should be marked failed, got %+v", bld) + } + // The persisted row must not be left pending forever. + if got := st.builds[bld.ID]; got == nil || got.Status != StatusFailed { + t.Errorf("persisted status = %v, want failed", got) + } +} + +// ---- the scan gate (Job phase -> DB) ---- + +func TestSyncSucceededAdmitsImageWithAddedBy(t *testing.T) { + b, st, jb := newBuilder() + bld, _ := b.Submit(context.Background(), goodRequest()) + + jb.phase = JobSucceeded + got, err := b.Sync(context.Background(), bld.ID) + if err != nil { + t.Fatalf("Sync: %v", err) + } + if got.Status != StatusSucceeded { + t.Errorf("status = %q, want succeeded", got.Status) + } + img, ok := st.images[bld.ImageRef] + if !ok { + t.Fatal("succeeded build did not admit its image to the whitelist") + } + if !img.Enabled { + t.Error("admitted image must be enabled=true") + } + if img.Source != SourceBuilt { + t.Errorf("source = %q, want built", img.Source) + } + if img.AddedBy != "admin@example.net" { + t.Errorf("added_by = %q, want the requester", img.AddedBy) + } + if img.BuildID != bld.ID { + t.Errorf("build_id = %q, want %q", img.BuildID, bld.ID) + } +} + +func TestSyncFailedDoesNotAdmitImage(t *testing.T) { + // A failed Job is exactly the CRITICAL-CVE case: trivy --exit-code 1 fails + // the Pod, so the gate is enforced as the Job verdict (spec §16). + b, st, jb := newBuilder() + bld, _ := b.Submit(context.Background(), goodRequest()) + + jb.phase = JobFailed + got, err := b.Sync(context.Background(), bld.ID) + if err != nil { + t.Fatalf("Sync: %v", err) + } + if got.Status != StatusFailed { + t.Errorf("status = %q, want failed", got.Status) + } + if len(st.images) != 0 { + t.Error("a failed/CRITICAL build must NOT admit any image") + } + if len(st.admitted) != 0 { + t.Error("AdmitBuiltImage must not be called on failure") + } +} + +func TestSyncRunningIsNoOp(t *testing.T) { + b, _, jb := newBuilder() + bld, _ := b.Submit(context.Background(), goodRequest()) + jb.phase = JobRunning + got, err := b.Sync(context.Background(), bld.ID) + if err != nil { + t.Fatalf("Sync: %v", err) + } + if got.Status != StatusBuilding { + t.Errorf("status = %q, want building (unchanged)", got.Status) + } +} + +func TestSyncIsIdempotentOnTerminalBuild(t *testing.T) { + b, st, jb := newBuilder() + bld, _ := b.Submit(context.Background(), goodRequest()) + jb.phase = JobSucceeded + if _, err := b.Sync(context.Background(), bld.ID); err != nil { + t.Fatalf("first Sync: %v", err) + } + // Flip the phase to a value that would admit again; a terminal build must + // not be re-reconciled. + before := len(st.admitted) + if _, err := b.Sync(context.Background(), bld.ID); err != nil { + t.Fatalf("second Sync: %v", err) + } + if len(st.admitted) != before { + t.Error("terminal build was re-admitted on a second Sync") + } +} + +func TestSyncAllAdvancesUnfinishedBuilds(t *testing.T) { + b, _, jb := newBuilder() + if _, err := b.Submit(context.Background(), goodRequest()); err != nil { + t.Fatalf("submit: %v", err) + } + jb.phase = JobSucceeded + n, err := b.SyncAll(context.Background()) + if err != nil { + t.Fatalf("SyncAll: %v", err) + } + if n != 1 { + t.Errorf("advanced = %d, want 1", n) + } +} + +// ---- cancellation ---- + +func TestCancelDeletesJobAndMarksCancelled(t *testing.T) { + b, _, jb := newBuilder() + bld, _ := b.Submit(context.Background(), goodRequest()) + got, err := b.Cancel(context.Background(), bld.ID) + if err != nil { + t.Fatalf("Cancel: %v", err) + } + if got.Status != StatusCancelled { + t.Errorf("status = %q, want cancelled", got.Status) + } + if len(jb.cancelled) != 1 { + t.Errorf("CancelBuildJob calls = %d, want 1", len(jb.cancelled)) + } +} + +func TestCancelTerminalBuildFails(t *testing.T) { + b, _, jb := newBuilder() + bld, _ := b.Submit(context.Background(), goodRequest()) + jb.phase = JobSucceeded + if _, err := b.Sync(context.Background(), bld.ID); err != nil { + t.Fatalf("Sync: %v", err) + } + if _, err := b.Cancel(context.Background(), bld.ID); !errors.Is(err, ErrAlreadyTerminal) { + t.Errorf("Cancel on terminal build err = %v, want ErrAlreadyTerminal", err) + } +} + +// ---- external admission ---- + +func TestAddExternalImageIsEnabledAndRecorded(t *testing.T) { + b, st, _ := newBuilder() + img, err := b.AddExternalImage(context.Background(), "registry.felis.svc:5000/ext:1", "admin@example.net") + if err != nil { + t.Fatalf("AddExternalImage: %v", err) + } + if img.Source != SourceExternal || !img.Enabled { + t.Errorf("external image = %+v, want enabled external", img) + } + if _, ok := st.images["registry.felis.svc:5000/ext:1"]; !ok { + t.Error("external image not persisted") + } +} + +func TestAddExternalImageRejectsMalformedRef(t *testing.T) { + b, _, _ := newBuilder() + if _, err := b.AddExternalImage(context.Background(), "not a ref!!", "admin@example.net"); err == nil { + t.Error("expected rejection of malformed image ref") + } +} + +func TestRemoveImage(t *testing.T) { + b, st, _ := newBuilder() + st.images["registry.felis.svc:5000/x:1"] = Image{ImageRef: "registry.felis.svc:5000/x:1"} + if err := b.RemoveImage(context.Background(), "registry.felis.svc:5000/x:1"); err != nil { + t.Fatalf("RemoveImage: %v", err) + } + if _, ok := st.images["registry.felis.svc:5000/x:1"]; ok { + t.Error("image not removed") + } +} + +// ---- whitelist admission for the create-server form (spec §15) ---- + +func TestImageMatches(t *testing.T) { + cases := []struct { + ref, pattern string + want bool + }{ + // exact match + {"registry.felis.svc:5000/mc:1.0", "registry.felis.svc:5000/mc:1.0", true}, + // exact mismatch on tag + {"registry.felis.svc:5000/mc:1.0", "registry.felis.svc:5000/mc:2.0", false}, + // tag wildcard admits any concrete tag on the repo + {"registry.felis.svc:5000/mc:1.0", "registry.felis.svc:5000/mc:*", true}, + {"registry.felis.svc:5000/mc:anything", "registry.felis.svc:5000/mc:*", true}, + // wildcard does not cross repos + {"registry.felis.svc:5000/other:1", "registry.felis.svc:5000/mc:*", false}, + // a registry-port colon is not a tag separator: an untagged ref under a + // ported host has no tag, so a wildcard (which needs a non-empty tag) misses + {"registry.felis.svc:5000/mc", "registry.felis.svc:5000/mc:*", false}, + // a non-wildcard pattern never matches via the prefix branch + {"registry.felis.svc:5000/mc:1", "registry.felis.svc:5000/mc", false}, + } + for _, c := range cases { + if got := imageMatches(c.ref, c.pattern); got != c.want { + t.Errorf("imageMatches(%q, %q) = %v, want %v", c.ref, c.pattern, got, c.want) + } + } +} + +func TestImageAdmitted(t *testing.T) { + b, st, _ := newBuilder() + st.images["registry.felis.svc:5000/exact:1"] = Image{ImageRef: "registry.felis.svc:5000/exact:1", Enabled: true} + st.images["registry.felis.svc:5000/wild:*"] = Image{ImageRef: "registry.felis.svc:5000/wild:*", Enabled: true} + st.images["registry.felis.svc:5000/off:1"] = Image{ImageRef: "registry.felis.svc:5000/off:1", Enabled: false} + + cases := []struct { + ref string + want bool + }{ + {"registry.felis.svc:5000/exact:1", true}, // exact, enabled + {"registry.felis.svc:5000/exact:2", false}, // wrong tag + {"registry.felis.svc:5000/wild:99", true}, // wildcard, enabled, port-colon safe + {"registry.felis.svc:5000/off:1", false}, // present but disabled + {"registry.felis.svc:5000/unknown:1", false}, // not on the list + {"", false}, // empty ref + {" ", false}, // blank ref + } + for _, c := range cases { + got, err := b.ImageAdmitted(context.Background(), c.ref) + if err != nil { + t.Fatalf("ImageAdmitted(%q): %v", c.ref, err) + } + if got != c.want { + t.Errorf("ImageAdmitted(%q) = %v, want %v", c.ref, got, c.want) + } + } +} diff --git a/internal/build/jobspec.go b/internal/build/jobspec.go new file mode 100644 index 0000000..1539d9f --- /dev/null +++ b/internal/build/jobspec.go @@ -0,0 +1,284 @@ +package build + +import ( + "fmt" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" +) + +// Label keys applied to build objects. ManagedBy doubles as the NetworkPolicy +// pod selector, so every build Pod is captured by the egress lock. +const ( + LabelManagedBy = "app.kubernetes.io/managed-by" + LabelComponent = "app.kubernetes.io/component" + LabelBuildID = "felis.lolicon.best/build-id" + + managedByValue = "felis-build" + componentValue = "image-build" +) + +// Container names within the build Pod. Kaniko is the initContainer that builds +// and pushes the image — its log IS the "build log" an admin watches (spec §16); +// Trivy is the main container whose CRITICAL-CVE verdict gates admission and is +// surfaced via the build status, not the log stream. Exported so the build-log +// streamer (internal/api.K8sBuildLogStreamer, spec §416 日志流复用 §8) follows the +// same container this Job defines — one source of truth for the name. +const ( + ContainerKaniko = "kaniko" + ContainerTrivy = "trivy" +) + +// JobParams are the rendered inputs to a build Job. They are derived from a +// Build + Config by the Builder; jobspec is a pure function of them so the +// security-critical Job shape is unit-tested without a cluster. +type JobParams struct { + BuildID string + ImageRef string + ContextRef string + Namespace string + ServiceAccount string + RegistryURL string + KanikoImage string + TrivyImage string + Deadline time.Duration + CPULimit string + MemLimit string +} + +// BuildJobName is the deterministic Job name for a build id. +func BuildJobName(buildID string) string { return "build-" + buildID } + +func buildLabels(p JobParams) map[string]string { + return map[string]string{ + LabelManagedBy: managedByValue, + LabelComponent: componentValue, + LabelBuildID: p.BuildID, + } +} + +// BuildJob renders the Kaniko+Trivy build Job (spec §16). Every isolation +// guarantee the spec demands is encoded here and asserted by jobspec_test.go, +// because no cluster runs in this environment: +// +// - runs in the isolated felis-build namespace with the weak felis-build SA +// (never the felis-api SA) and does NOT mount the SA token, so it cannot +// reach the K8s API (spec §16, §21); +// - no privileged container — Kaniko builds the Dockerfile without a daemon, +// so docker-in-docker / privileged is never needed (spec §16, §22); +// - activeDeadlineSeconds + backoffLimit=0 + per-container resource limits so +// a runaway or poisoned build cannot exhaust the cluster (spec §16); +// - the Trivy step runs with `--exit-code 1 --severity CRITICAL`, so a +// CRITICAL CVE fails the Pod and therefore the Job — the only retained +// automatic admission gate (spec §16). +// +// Sequencing: kaniko runs as an initContainer (build + push to the internal +// registry) and trivy as the main container (scan the pushed ref). The Pod +// succeeds only if kaniko pushed AND trivy found no CRITICAL CVE. +func BuildJob(p JobParams) (*batchv1.Job, error) { + limits, err := resourceLimits(p.CPULimit, p.MemLimit) + if err != nil { + return nil, err + } + deadline := int64(p.Deadline / time.Second) + if deadline <= 0 { + deadline = int64(defaultDeadline / time.Second) + } + + // Hardened container security context shared by both build containers: no + // privilege, no privilege escalation, drop all capabilities. Kaniko needs a + // writable root filesystem to unpack layers, so we do not force read-only + // root here, but it gains no privilege. + sec := &corev1.SecurityContext{ + Privileged: boolPtr(false), + AllowPrivilegeEscalation: boolPtr(false), + Capabilities: &corev1.Capabilities{Drop: []corev1.Capability{"ALL"}}, + } + + kaniko := corev1.Container{ + Name: ContainerKaniko, + Image: p.KanikoImage, + Args: []string{ + "--dockerfile=Dockerfile", + "--context=" + p.ContextRef, + "--destination=" + p.ImageRef, + // The internal registry is in-cluster only and may serve plain HTTP; + // it is never a public ingress (spec §17). + "--insecure", + "--skip-tls-verify", + }, + Resources: corev1.ResourceRequirements{Limits: limits, Requests: limits}, + SecurityContext: sec, + } + + trivy := corev1.Container{ + Name: ContainerTrivy, + Image: p.TrivyImage, + Args: []string{ + "image", + "--exit-code", "1", + "--severity", "CRITICAL", + "--no-progress", + "--insecure", + p.ImageRef, + }, + Resources: corev1.ResourceRequirements{Limits: limits, Requests: limits}, + SecurityContext: sec, + } + + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: BuildJobName(p.BuildID), + Namespace: p.Namespace, + Labels: buildLabels(p), + }, + Spec: batchv1.JobSpec{ + // A poisoned build must not loop — one shot, then a terminal verdict. + BackoffLimit: int32Ptr(0), + ActiveDeadlineSeconds: int64Ptr(deadline), + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{Labels: buildLabels(p)}, + Spec: corev1.PodSpec{ + RestartPolicy: corev1.RestartPolicyNever, + ServiceAccountName: p.ServiceAccount, + AutomountServiceAccountToken: boolPtr(false), + InitContainers: []corev1.Container{kaniko}, + Containers: []corev1.Container{trivy}, + }, + }, + }, + } + return job, nil +} + +// NetPolParams parameterises the build-namespace egress lock. +type NetPolParams struct { + Namespace string + RegistryNamespace string + RegistryPort int32 + // PackageSourceCIDRs is an optional, explicit allowlist of external package + // mirrors (spec §16: egress 仅 registry + 包源). Empty means the most + // locked-down default — no internet egress at all (默认拒外网). + PackageSourceCIDRs []string +} + +// BuildNetworkPolicy renders the default-deny egress policy for build Pods +// (spec §16, §21: build ns egress 仅放 registry + 包源,默认拒外网). It selects +// build Pods by the managed-by label, denies all ingress, and allows egress +// only to DNS, the internal registry, and any explicitly configured package +// mirrors. There is deliberately no allow-all egress rule. +func BuildNetworkPolicy(p NetPolParams) *networkingv1.NetworkPolicy { + port := p.RegistryPort + if port == 0 { + port = 5000 + } + dnsUDP := corev1.ProtocolUDP + dnsTCP := corev1.ProtocolTCP + dns53 := intstr.FromInt32(53) + regPort := intstr.FromInt32(port) + + egress := []networkingv1.NetworkPolicyEgressRule{ + // DNS resolution: port-restricted to 53, so this is not an open-internet + // hole — name resolution only. + { + Ports: []networkingv1.NetworkPolicyPort{ + {Protocol: &dnsUDP, Port: &dns53}, + {Protocol: &dnsTCP, Port: &dns53}, + }, + }, + // The internal registry, selected by the namespace's immutable + // kubernetes.io/metadata.name label, on the registry port only. + { + To: []networkingv1.NetworkPolicyPeer{{ + NamespaceSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{"kubernetes.io/metadata.name": p.RegistryNamespace}, + }, + }}, + Ports: []networkingv1.NetworkPolicyPort{ + {Protocol: &dnsTCP, Port: ®Port}, + }, + }, + } + // Explicit package-mirror CIDRs, when configured. No CIDR ⇒ no internet. + for _, cidr := range p.PackageSourceCIDRs { + egress = append(egress, networkingv1.NetworkPolicyEgressRule{ + To: []networkingv1.NetworkPolicyPeer{{ + IPBlock: &networkingv1.IPBlock{CIDR: cidr}, + }}, + }) + } + + return &networkingv1.NetworkPolicy{ + ObjectMeta: metav1.ObjectMeta{ + Name: "felis-build-egress", + Namespace: p.Namespace, + Labels: map[string]string{ + LabelManagedBy: managedByValue, + LabelComponent: componentValue, + }, + }, + Spec: networkingv1.NetworkPolicySpec{ + PodSelector: metav1.LabelSelector{ + MatchLabels: map[string]string{LabelManagedBy: managedByValue}, + }, + PolicyTypes: []networkingv1.PolicyType{ + networkingv1.PolicyTypeIngress, + networkingv1.PolicyTypeEgress, + }, + // Empty Ingress slice = deny all ingress: nothing connects to a + // build Pod. + Ingress: []networkingv1.NetworkPolicyIngressRule{}, + Egress: egress, + }, + } +} + +// BuildServiceAccount renders the weak build SA (spec §16, §21). It is the most +// dangerous identity in the platform if mis-scoped, so it is created bare: no +// secrets, token auto-mounting disabled, and — by virtue of having no Role or +// RoleBinding anywhere — zero K8s API permissions. Its only capability is +// network reachability to push to the registry, which RBAC does not grant. +func BuildServiceAccount(namespace, name string) *corev1.ServiceAccount { + return &corev1.ServiceAccount{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: namespace, + Labels: map[string]string{ + LabelManagedBy: managedByValue, + LabelComponent: componentValue, + }, + }, + AutomountServiceAccountToken: boolPtr(false), + } +} + +// resourceLimits parses the CPU/memory limits into a ResourceList. +func resourceLimits(cpu, mem string) (corev1.ResourceList, error) { + if cpu == "" { + cpu = defaultCPULimit + } + if mem == "" { + mem = defaultMemLimit + } + cpuQty, err := resource.ParseQuantity(cpu) + if err != nil { + return nil, fmt.Errorf("build: invalid cpu limit %q: %w", cpu, err) + } + memQty, err := resource.ParseQuantity(mem) + if err != nil { + return nil, fmt.Errorf("build: invalid memory limit %q: %w", mem, err) + } + return corev1.ResourceList{ + corev1.ResourceCPU: cpuQty, + corev1.ResourceMemory: memQty, + }, nil +} + +func boolPtr(b bool) *bool { return &b } +func int32Ptr(i int32) *int32 { return &i } +func int64Ptr(i int64) *int64 { return &i } diff --git a/internal/build/jobspec_test.go b/internal/build/jobspec_test.go new file mode 100644 index 0000000..e66b290 --- /dev/null +++ b/internal/build/jobspec_test.go @@ -0,0 +1,306 @@ +package build + +import ( + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + networkingv1 "k8s.io/api/networking/v1" +) + +func sampleJobParams() JobParams { + return JobParams{ + BuildID: "bld-1", + ImageRef: "registry.felis.svc:5000/mc-paper:1.0", + ContextRef: "tar://contexts/abc.tar.gz", + Namespace: defaultNamespace, + ServiceAccount: defaultServiceAccount, + RegistryURL: "registry.felis.svc:5000", + KanikoImage: defaultKanikoImage, + TrivyImage: defaultTrivyImage, + Deadline: 30 * time.Minute, + CPULimit: "2", + MemLimit: "4Gi", + } +} + +// The Job must run in the isolated build namespace under the weak build SA — +// never the api namespace/identity, and never the minecraft namespace. This is +// the central §16/§21 red line, asserted on the rendered spec because no cluster +// runs here. +func TestBuildJobRunsIsolatedUnderWeakSA(t *testing.T) { + job, err := BuildJob(sampleJobParams()) + if err != nil { + t.Fatalf("BuildJob: %v", err) + } + if job.Namespace != "felis-build" { + t.Errorf("namespace = %q, want felis-build (never minecraft/felis-system)", job.Namespace) + } + sa := job.Spec.Template.Spec.ServiceAccountName + if sa != "felis-build" { + t.Errorf("service account = %q, want felis-build (never felis-api)", sa) + } + if sa == "felis-api" { + t.Fatal("build Pod must NOT run as the felis-api SA") + } + // The SA token must not be mounted: with no token, the Pod cannot reach the + // K8s API even if a Role were mis-bound. + if amt := job.Spec.Template.Spec.AutomountServiceAccountToken; amt == nil || *amt { + t.Error("AutomountServiceAccountToken must be explicitly false") + } +} + +// A poisoned build must terminate and not loop or run unbounded. +func TestBuildJobIsBoundedAndOneShot(t *testing.T) { + job, err := BuildJob(sampleJobParams()) + if err != nil { + t.Fatalf("BuildJob: %v", err) + } + if job.Spec.BackoffLimit == nil || *job.Spec.BackoffLimit != 0 { + t.Error("BackoffLimit must be 0 — a poisoned build must not retry") + } + if job.Spec.ActiveDeadlineSeconds == nil || *job.Spec.ActiveDeadlineSeconds <= 0 { + t.Error("ActiveDeadlineSeconds must be a positive wall-clock cap") + } + if got := *job.Spec.ActiveDeadlineSeconds; got != int64((30 * time.Minute).Seconds()) { + t.Errorf("ActiveDeadlineSeconds = %d, want 1800", got) + } + if job.Spec.Template.Spec.RestartPolicy != corev1.RestartPolicyNever { + t.Error("RestartPolicy must be Never") + } +} + +// No build container may be privileged or able to escalate, and all containers +// must carry resource limits. +func TestBuildJobContainersAreHardened(t *testing.T) { + job, err := BuildJob(sampleJobParams()) + if err != nil { + t.Fatalf("BuildJob: %v", err) + } + all := append([]corev1.Container{}, job.Spec.Template.Spec.InitContainers...) + all = append(all, job.Spec.Template.Spec.Containers...) + if len(all) < 2 { + t.Fatalf("expected kaniko initContainer + trivy container, got %d containers", len(all)) + } + for _, c := range all { + sc := c.SecurityContext + if sc == nil { + t.Fatalf("container %q has no security context", c.Name) + } + if sc.Privileged == nil || *sc.Privileged { + t.Errorf("container %q must not be privileged (Kaniko needs no daemon)", c.Name) + } + if sc.AllowPrivilegeEscalation == nil || *sc.AllowPrivilegeEscalation { + t.Errorf("container %q must set allowPrivilegeEscalation=false", c.Name) + } + if sc.Capabilities == nil || len(sc.Capabilities.Drop) == 0 { + t.Errorf("container %q must drop capabilities", c.Name) + } else if string(sc.Capabilities.Drop[0]) != "ALL" { + t.Errorf("container %q must drop ALL capabilities, got %v", c.Name, sc.Capabilities.Drop) + } + if c.Resources.Limits.Cpu().IsZero() || c.Resources.Limits.Memory().IsZero() { + t.Errorf("container %q must carry CPU+memory limits", c.Name) + } + } +} + +// kaniko builds and pushes to the request's exact target; trivy gates admission +// with --exit-code 1 --severity CRITICAL on that same ref. +func TestBuildJobKanikoPushesAndTrivyGates(t *testing.T) { + p := sampleJobParams() + job, err := BuildJob(p) + if err != nil { + t.Fatalf("BuildJob: %v", err) + } + if len(job.Spec.Template.Spec.InitContainers) != 1 { + t.Fatalf("expected exactly one (kaniko) initContainer") + } + kaniko := job.Spec.Template.Spec.InitContainers[0] + if kaniko.Name != "kaniko" { + t.Errorf("init container = %q, want kaniko", kaniko.Name) + } + if !hasArg(kaniko.Args, "--destination="+p.ImageRef) { + t.Errorf("kaniko must push to %q, args=%v", p.ImageRef, kaniko.Args) + } + + if len(job.Spec.Template.Spec.Containers) != 1 { + t.Fatalf("expected exactly one (trivy) main container") + } + trivy := job.Spec.Template.Spec.Containers[0] + if trivy.Name != "trivy" { + t.Errorf("main container = %q, want trivy", trivy.Name) + } + // The scan gate: a CRITICAL CVE must fail the Pod (and thus the Job). + if !argPairPresent(trivy.Args, "--exit-code", "1") { + t.Errorf("trivy must run with --exit-code 1, args=%v", trivy.Args) + } + if !argPairPresent(trivy.Args, "--severity", "CRITICAL") { + t.Errorf("trivy must gate on --severity CRITICAL, args=%v", trivy.Args) + } + if !hasArg(trivy.Args, p.ImageRef) { + t.Errorf("trivy must scan the pushed ref %q, args=%v", p.ImageRef, trivy.Args) + } +} + +// The build namespace egress lock must be default-deny: deny all ingress, and +// allow egress only to DNS + the internal registry — never an allow-all rule. +func TestBuildNetworkPolicyIsDefaultDeny(t *testing.T) { + np := BuildNetworkPolicy(NetPolParams{ + Namespace: "felis-build", + RegistryNamespace: "felis-system", + RegistryPort: 5000, + }) + + if !hasPolicyType(np, networkingv1.PolicyTypeEgress) || !hasPolicyType(np, networkingv1.PolicyTypeIngress) { + t.Fatal("policy must govern both Ingress and Egress") + } + // Ingress: empty rule slice = deny all. + if len(np.Spec.Ingress) != 0 { + t.Errorf("ingress must be empty (deny all), got %d rules", len(np.Spec.Ingress)) + } + // The selector must capture every build Pod by the managed-by label. + if np.Spec.PodSelector.MatchLabels[LabelManagedBy] != managedByValue { + t.Errorf("pod selector must match managed-by=%s", managedByValue) + } + // No egress rule may be an allow-all (a rule with neither To peers nor Ports + // would permit unrestricted egress). + for i, rule := range np.Spec.Egress { + if len(rule.To) == 0 && len(rule.Ports) == 0 { + t.Errorf("egress rule %d is allow-all (no To, no Ports) — internet would be open", i) + } + } + // The registry must be reachable (by namespace selector), and DNS allowed. + if !egressAllowsNamespace(np, "felis-system") { + t.Error("egress must allow the registry namespace") + } + if !egressAllowsPort(np, 53) { + t.Error("egress must allow DNS (port 53)") + } +} + +// With no package-source CIDRs configured, there must be zero IPBlock egress — +// the most locked-down default (no open internet). +func TestBuildNetworkPolicyDefaultsToNoInternet(t *testing.T) { + np := BuildNetworkPolicy(NetPolParams{ + Namespace: "felis-build", + RegistryNamespace: "felis-system", + }) + for i, rule := range np.Spec.Egress { + for _, peer := range rule.To { + if peer.IPBlock != nil { + t.Errorf("egress rule %d has an IPBlock but no package sources were configured", i) + } + } + } +} + +func TestBuildNetworkPolicyAllowsConfiguredPackageMirrors(t *testing.T) { + cidr := "192.0.2.0/24" + np := BuildNetworkPolicy(NetPolParams{ + Namespace: "felis-build", + RegistryNamespace: "felis-system", + PackageSourceCIDRs: []string{cidr}, + }) + found := false + for _, rule := range np.Spec.Egress { + for _, peer := range rule.To { + if peer.IPBlock != nil && peer.IPBlock.CIDR == cidr { + found = true + } + } + } + if !found { + t.Errorf("configured package-mirror CIDR %q not present in egress", cidr) + } +} + +// The build SA is the most dangerous identity if mis-scoped: it must be bare — +// no secrets, token automount disabled, and (by having no Role anywhere) no API +// rights. We assert the spec-level properties the renderer controls. +func TestBuildServiceAccountIsBare(t *testing.T) { + sa := BuildServiceAccount("felis-build", "felis-build") + if sa.Namespace != "felis-build" { + t.Errorf("SA namespace = %q, want felis-build", sa.Namespace) + } + if sa.AutomountServiceAccountToken == nil || *sa.AutomountServiceAccountToken { + t.Error("SA must disable token automounting") + } + if len(sa.Secrets) != 0 { + t.Errorf("SA must carry no secrets, got %d", len(sa.Secrets)) + } + if len(sa.ImagePullSecrets) != 0 { + t.Errorf("SA must carry no image-pull secrets, got %d", len(sa.ImagePullSecrets)) + } +} + +// An invalid resource limit must surface as an error rather than render a Job +// with no limits. +func TestBuildJobRejectsBadResourceLimit(t *testing.T) { + p := sampleJobParams() + p.CPULimit = "not-a-quantity" + if _, err := BuildJob(p); err == nil { + t.Error("expected error for an unparseable CPU limit") + } +} + +// ---- helpers ---- + +func hasArg(args []string, want string) bool { + for _, a := range args { + if a == want { + return true + } + } + return false +} + +// argPairPresent reports whether flag is immediately followed by val (the +// `--exit-code 1` two-token form). +func argPairPresent(args []string, flag, val string) bool { + for i := 0; i < len(args)-1; i++ { + if args[i] == flag && args[i+1] == val { + return true + } + } + return false +} + +func hasPolicyType(np *networkingv1.NetworkPolicy, t networkingv1.PolicyType) bool { + for _, pt := range np.Spec.PolicyTypes { + if pt == t { + return true + } + } + return false +} + +func egressAllowsNamespace(np *networkingv1.NetworkPolicy, ns string) bool { + for _, rule := range np.Spec.Egress { + for _, peer := range rule.To { + if peer.NamespaceSelector != nil && + peer.NamespaceSelector.MatchLabels["kubernetes.io/metadata.name"] == ns { + return true + } + } + } + return false +} + +func egressAllowsPort(np *networkingv1.NetworkPolicy, port int32) bool { + for _, rule := range np.Spec.Egress { + for _, p := range rule.Ports { + if p.Port != nil && p.Port.IntVal == port { + return true + } + } + } + return false +} + +// guard against accidental shorthand: the test image ref must be host-qualified. +func TestSampleRefIsHostQualified(t *testing.T) { + if !strings.Contains(sampleJobParams().ImageRef, "/") { + t.Fatal("sample image ref must be host-qualified") + } +} diff --git a/internal/build/k8sjobs.go b/internal/build/k8sjobs.go new file mode 100644 index 0000000..4aaa680 --- /dev/null +++ b/internal/build/k8sjobs.go @@ -0,0 +1,87 @@ +package build + +import ( + "context" + + batchv1 "k8s.io/api/batch/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// K8sJobs is the production Jobs backed by a controller-runtime client (spec +// §16). It creates the Kaniko+Trivy build Job, reads its phase for the scan-gate +// translation, and deletes it on cancel — nothing more. The cluster-bootstrap +// objects (the felis-build namespace, the weak SA, and the egress +// NetworkPolicy) are installed once by the deployment manifests (spec §21), not +// per build, so this binding never needs to create them. It is integration- +// tested against a live cluster, not the hermetic build_test.go suite. +type K8sJobs struct { + c client.Client + cfg Config +} + +// NewK8sJobs builds a Jobs over c using cfg for the namespace and image refs. +func NewK8sJobs(c client.Client, cfg Config) *K8sJobs { + return &K8sJobs{c: c, cfg: cfg.withDefaults()} +} + +// CreateBuildJob renders and applies the build Job, returning its name. The Job +// is the security-critical object; its shape is fixed by BuildJob (jobspec.go) +// and asserted by jobspec_test.go. +func (k *K8sJobs) CreateBuildJob(ctx context.Context, p JobParams) (string, error) { + job, err := BuildJob(p) + if err != nil { + return "", err + } + if err := k.c.Create(ctx, job); err != nil { + return "", err + } + return job.Name, nil +} + +// JobPhase reads the Job and maps its status to a JobPhase. A missing Job +// (GC'd, never created) is JobUnknown, which Sync treats as failed. The mapping +// is deliberately conservative: a Job is Succeeded only when the Complete +// condition is true, so a half-finished Job is never admitted. +func (k *K8sJobs) JobPhase(ctx context.Context, jobName string) (JobPhase, error) { + var job batchv1.Job + if err := k.c.Get(ctx, types.NamespacedName{Namespace: k.cfg.Namespace, Name: jobName}, &job); err != nil { + if apierrors.IsNotFound(err) { + return JobUnknown, nil + } + return JobUnknown, err + } + for _, cond := range job.Status.Conditions { + if cond.Status != "True" { + continue + } + switch cond.Type { + case batchv1.JobComplete: + return JobSucceeded, nil + case batchv1.JobFailed: + // Covers a CRITICAL CVE (trivy --exit-code 1), a kaniko failure, and + // DeadlineExceeded — all are a rejected build. + return JobFailed, nil + } + } + if job.Status.Active > 0 { + return JobRunning, nil + } + return JobPending, nil +} + +// CancelBuildJob deletes the Job and, via background propagation, its pods. A +// missing Job is not an error: cancellation is idempotent. +func (k *K8sJobs) CancelBuildJob(ctx context.Context, jobName string) error { + bg := metav1.DeletePropagationBackground + obj := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{Namespace: k.cfg.Namespace, Name: jobName}, + } + if err := k.c.Delete(ctx, obj, &client.DeleteOptions{PropagationPolicy: &bg}); err != nil && + !apierrors.IsNotFound(err) { + return err + } + return nil +} diff --git a/internal/build/pgstore.go b/internal/build/pgstore.go new file mode 100644 index 0000000..8a86110 --- /dev/null +++ b/internal/build/pgstore.go @@ -0,0 +1,186 @@ +package build + +import ( + "context" + "database/sql" + "time" +) + +// PGStore is the production Store backed by Postgres (spec §6, §16). It writes +// the two tables of the build subsystem — image_builds and image_whitelist — +// and is the *only* component that holds database credentials: the build Pod +// never does (the weak-SA red line). The SQL here is exercised by integration +// tests against a live database, not the hermetic build_test.go suite. Every +// statement is a narrow operation; there is no generic UPDATE escape hatch. +type PGStore struct { + db *sql.DB +} + +// NewPGStore wraps an existing pool (from store.PostgresDriver.DB()). +func NewPGStore(db *sql.DB) *PGStore { return &PGStore{db: db} } + +func (s *PGStore) CreateBuild(ctx context.Context, b *Build) error { + const q = `INSERT INTO image_builds + (id, image_ref, status, dockerfile, context_ref, base_image, requested_by, created_at) + VALUES ($1, $2, $3, $4, NULLIF($5, ''), NULLIF($6, ''), $7, $8)` + _, err := s.db.ExecContext(ctx, q, + b.ID, b.ImageRef, string(b.Status), b.Dockerfile, b.ContextRef, b.BaseImage, + b.RequestedBy, b.CreatedAt) + return err +} + +func (s *PGStore) GetBuild(ctx context.Context, id string) (*Build, error) { + const q = `SELECT id, image_ref, status, dockerfile, context_ref, base_image, + requested_by, job_name, log_ref, error, created_at, finished_at + FROM image_builds WHERE id = $1` + return s.scanBuild(s.db.QueryRowContext(ctx, q, id)) +} + +func (s *PGStore) scanBuild(row *sql.Row) (*Build, error) { + var ( + b Build + status string + ctxRef, base, jobName, logRef, eMsg sql.NullString + finished sql.NullTime + ) + switch err := row.Scan(&b.ID, &b.ImageRef, &status, &b.Dockerfile, &ctxRef, &base, + &b.RequestedBy, &jobName, &logRef, &eMsg, &b.CreatedAt, &finished); { + case err == sql.ErrNoRows: + return nil, ErrNotFound + case err != nil: + return nil, err + } + b.Status = Status(status) + b.ContextRef = ctxRef.String + b.BaseImage = base.String + b.JobName = jobName.String + b.LogRef = logRef.String + b.Error = eMsg.String + if finished.Valid { + t := finished.Time + b.FinishedAt = &t + } + return &b, nil +} + +func (s *PGStore) SetBuildJob(ctx context.Context, id, jobName string) error { + const q = `UPDATE image_builds SET job_name = $2, status = 'building' + WHERE id = $1 AND status = 'pending'` + res, err := s.db.ExecContext(ctx, q, id, jobName) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +func (s *PGStore) FinishBuild(ctx context.Context, id string, status Status, errMsg string, at time.Time) error { + const q = `UPDATE image_builds SET status = $2, error = NULLIF($3, ''), finished_at = $4 + WHERE id = $1` + res, err := s.db.ExecContext(ctx, q, id, string(status), errMsg, at) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} + +func (s *PGStore) ListUnfinishedBuilds(ctx context.Context) ([]Build, error) { + const q = `SELECT id, image_ref, status, dockerfile, context_ref, base_image, + requested_by, job_name, log_ref, error, created_at, finished_at + FROM image_builds WHERE status IN ('pending', 'building') ORDER BY created_at ASC` + rows, err := s.db.QueryContext(ctx, q) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Build + for rows.Next() { + var ( + b Build + status string + ctxRef, base, jobName, logRef, eMsg sql.NullString + finished sql.NullTime + ) + if err := rows.Scan(&b.ID, &b.ImageRef, &status, &b.Dockerfile, &ctxRef, &base, + &b.RequestedBy, &jobName, &logRef, &eMsg, &b.CreatedAt, &finished); err != nil { + return nil, err + } + b.Status = Status(status) + b.ContextRef = ctxRef.String + b.BaseImage = base.String + b.JobName = jobName.String + b.LogRef = logRef.String + b.Error = eMsg.String + if finished.Valid { + t := finished.Time + b.FinishedAt = &t + } + out = append(out, b) + } + return out, rows.Err() +} + +// AdmitBuiltImage upserts the whitelist row on scan-gate success. ON CONFLICT +// re-enables and re-stamps a previously-removed or superseded ref, so a rebuild +// of the same tag re-admits it (spec §16). +func (s *PGStore) AdmitBuiltImage(ctx context.Context, img Image) error { + const q = `INSERT INTO image_whitelist + (image_ref, source, build_id, added_by, enabled, added_at) + VALUES ($1, 'built', NULLIF($2, ''), $3, true, $4) + ON CONFLICT (image_ref) DO UPDATE + SET source = 'built', build_id = EXCLUDED.build_id, added_by = EXCLUDED.added_by, + enabled = true, added_at = EXCLUDED.added_at` + _, err := s.db.ExecContext(ctx, q, img.ImageRef, img.BuildID, img.AddedBy, img.AddedAt) + return err +} + +func (s *PGStore) ListImages(ctx context.Context) ([]Image, error) { + const q = `SELECT image_ref, source, build_id, added_by, enabled, added_at + FROM image_whitelist ORDER BY added_at DESC` + rows, err := s.db.QueryContext(ctx, q) + if err != nil { + return nil, err + } + defer rows.Close() + var out []Image + for rows.Next() { + var ( + img Image + buildID sql.NullString + ) + if err := rows.Scan(&img.ImageRef, &img.Source, &buildID, &img.AddedBy, + &img.Enabled, &img.AddedAt); err != nil { + return nil, err + } + img.BuildID = buildID.String + out = append(out, img) + } + return out, rows.Err() +} + +func (s *PGStore) AddExternalImage(ctx context.Context, img Image) error { + const q = `INSERT INTO image_whitelist + (image_ref, source, added_by, enabled, added_at) + VALUES ($1, 'external', $2, true, $3) + ON CONFLICT (image_ref) DO UPDATE + SET source = 'external', build_id = NULL, added_by = EXCLUDED.added_by, + enabled = true, added_at = EXCLUDED.added_at` + _, err := s.db.ExecContext(ctx, q, img.ImageRef, img.AddedBy, img.AddedAt) + return err +} + +func (s *PGStore) RemoveImage(ctx context.Context, imageRef string) error { + res, err := s.db.ExecContext(ctx, `DELETE FROM image_whitelist WHERE image_ref = $1`, imageRef) + if err != nil { + return err + } + if n, _ := res.RowsAffected(); n == 0 { + return ErrNotFound + } + return nil +} diff --git a/internal/build/validate.go b/internal/build/validate.go new file mode 100644 index 0000000..686bf65 --- /dev/null +++ b/internal/build/validate.go @@ -0,0 +1,150 @@ +package build + +import ( + "fmt" + "regexp" + "strings" +) + +// invalidf builds a validation error wrapping ErrInvalid so the API layer maps +// every malformed-request case to a single 400 path. +func invalidf(format string, a ...any) error { + return fmt.Errorf("%w: "+format, append([]any{ErrInvalid}, a...)...) +} + +// imageNameRE matches the path+tag of an image reference under the registry +// host, e.g. "foo/bar:1.0" or "mc-paper:latest". It is intentionally strict: +// lowercase path segments, an optional tag of the same alphabet, no digests, no +// shell metacharacters that could escape into the Kaniko/Trivy argv. +var imageNameRE = regexp.MustCompile(`^[a-z0-9]([a-z0-9._/-]*[a-z0-9])?(:[a-zA-Z0-9._-]+)?$`) + +// Validate enforces the §16 admission rules on a build request: the target must +// address the internal registry, the Dockerfile must be present and within the +// size cap, and a context reference is required (Kaniko pulls it, spec §17). +func Validate(req Request, cfg Config) error { + cfg = cfg.withDefaults() + if err := validateRegistryTarget(req.ImageRef, cfg.RegistryURL); err != nil { + return err + } + if strings.TrimSpace(req.Dockerfile) == "" { + return invalidf("dockerfile is required") + } + if len(req.Dockerfile) > cfg.MaxDockerfileBytes { + return invalidf("dockerfile exceeds %d bytes", cfg.MaxDockerfileBytes) + } + if strings.TrimSpace(req.ContextRef) == "" { + return invalidf("context reference is required") + } + return nil +} + +// ValidateImageRef checks a bare image reference (used by external admission, +// where there is no registry-target constraint beyond well-formedness). A +// whitelist entry may be a tag wildcard ("registry/foo:*", a legitimate +// image_whitelist value per spec §15): the trailing ":*" is stripped before the +// path is validated. This is safe because a wildcard is only ever compared +// against concrete refs by imageMatches — it never reaches a Kaniko/Trivy argv, +// unlike a build push target (validateRegistryTarget stays strictly concrete). +func ValidateImageRef(ref string) error { + if ref == "" { + return invalidf("image reference is required") + } + host, rest, ok := splitRegistryHost(ref) + if !ok { + return invalidf("image reference %q must be host-qualified (host/path:tag)", ref) + } + if repo, isWildcard := strings.CutSuffix(rest, ":*"); isWildcard { + rest = repo + } + if !imageNameRE.MatchString(rest) { + return invalidf("invalid image path/tag %q", rest) + } + _ = host + return nil +} + +// splitTag splits an image reference's path from its tag. It is registry-port +// safe: only a colon *after* the final path separator is a tag separator, so +// "registry:5000/foo" splits to ("registry:5000/foo", ""), never a bogus tag. +func splitTag(ref string) (repo, tag string) { + slash := strings.LastIndexByte(ref, '/') + colon := strings.LastIndexByte(ref, ':') + if colon > slash { + return ref[:colon], ref[colon+1:] + } + return ref, "" +} + +// imageMatches reports whether a concrete image ref is admitted by a whitelist +// pattern. A pattern is either exact ("registry/foo:1.0") or a tag wildcard +// ("registry/foo:*", spec §15) that matches any non-empty tag on the same repo. +// A wildcard never matches an untagged ref — admission is always to a concrete +// tag. +func imageMatches(ref, pattern string) bool { + if ref == pattern { + return true + } + prefix, ok := strings.CutSuffix(pattern, ":*") + if !ok { + return false + } + repo, tag := splitTag(ref) + return repo == prefix && tag != "" +} + +// validateRegistryTarget enforces that a build pushes only to the configured +// internal registry — never an arbitrary external host (spec §16: the build can +// never push elsewhere; the registry is not a public ingress). When RegistryURL +// is unset (tests / not configured) the host constraint is skipped but the +// path/tag are still validated. +func validateRegistryTarget(ref, registryURL string) error { + if ref == "" { + return invalidf("image reference is required") + } + host, rest, ok := splitRegistryHost(ref) + if !ok { + return invalidf("image reference %q must target the internal registry (host/path:tag)", ref) + } + if !imageNameRE.MatchString(rest) { + return invalidf("invalid image path/tag %q", rest) + } + if registryURL != "" && host != registryHost(registryURL) { + return invalidf("image reference %q must target the internal registry %q, not %q", + ref, registryHost(registryURL), host) + } + return nil +} + +// splitRegistryHost separates the registry host from the remaining path+tag. A +// reference is host-qualified only if the first segment looks like a registry +// host — it contains a '.' or ':' (port), matching containerd's heuristic. +// "foo/bar:1" (Docker Hub shorthand) is rejected: builds must be explicit about +// the internal registry. +func splitRegistryHost(ref string) (host, rest string, ok bool) { + slash := strings.IndexByte(ref, '/') + if slash < 0 { + return "", "", false + } + host = ref[:slash] + rest = ref[slash+1:] + if !strings.ContainsAny(host, ".:") { + return "", "", false + } + if rest == "" { + return "", "", false + } + return host, rest, true +} + +// registryHost strips any scheme and path from a configured registry URL, +// leaving the host[:port] that an image reference must match. +func registryHost(registryURL string) string { + h := registryURL + if i := strings.Index(h, "://"); i >= 0 { + h = h[i+3:] + } + if i := strings.IndexByte(h, '/'); i >= 0 { + h = h[:i] + } + return h +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..70ad78c --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,204 @@ +// Package config loads and validates felis.toml (spec §24). root_domain lives +// here and nowhere else in code: every FQDN is composed at runtime as +// subdomain + "." + root_domain, so changing the deployment domain is a +// one-line config edit and the source tree stays domain-agnostic. +package config + +import ( + "fmt" + "strings" + + "github.com/BurntSushi/toml" +) + +// Config is the parsed felis.toml. +type Config struct { + Server ServerConfig `toml:"server"` + Database DatabaseConfig `toml:"database"` + Velocity VelocityConfig `toml:"velocity"` + Auth AuthConfig `toml:"auth"` + K8s K8sConfig `toml:"k8s"` + Registry RegistryConfig `toml:"registry"` + Archive ArchiveConfig `toml:"archive"` +} + +// ServerConfig is the [server] table. +type ServerConfig struct { + Listen string `toml:"listen"` + RootDomain string `toml:"root_domain"` +} + +// DatabaseConfig is the [database] table. +type DatabaseConfig struct { + URL string `toml:"url"` +} + +// VelocityConfig is the [velocity] table. +type VelocityConfig struct { + PublicIP string `toml:"public_ip"` + ServiceTokenRef string `toml:"service_token_ref"` +} + +// AuthConfig is the [auth] table: the two privileged faces and the access-JWT +// audience the API enforces. +type AuthConfig struct { + AdminHostname string `toml:"admin_hostname"` + PanelHostname string `toml:"panel_hostname"` + AccessJWTAud string `toml:"access_jwt_aud"` +} + +// K8sConfig is the [k8s] table. +type K8sConfig struct { + Namespace string `toml:"namespace"` + EgressMode string `toml:"egress_mode"` + MetalLBPool string `toml:"metallb_pool"` +} + +// RegistryConfig is the [registry] table. +type RegistryConfig struct { + URL string `toml:"url"` + BuildNamespace string `toml:"build_namespace"` + // UserUploadsContext is the object-store base under which a user-submitted + // modpack's Kaniko build context is pinned. It belongs to the §16 build + // subsystem's input domain (the build-context store), introduced by the + // user-directed modpack approval lane (see internal/submit package doc). The + // lane derives {UserUploadsContext}/{submissionID}/context.tar.gz; the upload + // transport that places the blob there is a separate, deferred integration + // (INTEGRATION-ONLY). It is kept distinct from [archive] on purpose — a world + // archive (§19 WorldArchiver) and a build context (§16) are different artifacts + // with different lifecycles, so the two must not share a store binding. + UserUploadsContext string `toml:"user_uploads_context"` +} + +// ArchiveConfig is the [archive] table plus its [archive.s3] subtable (spec §19). +type ArchiveConfig struct { + Store string `toml:"store"` + LocalPath string `toml:"local_path"` + Retention string `toml:"retention"` + WarnBefore []string `toml:"warn_before"` + MaxLocalBytes string `toml:"max_local_bytes"` + S3 ArchiveS3Config `toml:"s3"` +} + +// ArchiveS3Config is the [archive.s3] subtable. +type ArchiveS3Config struct { + Endpoint string `toml:"endpoint"` + Bucket string `toml:"bucket"` + AccessKeyRef string `toml:"access_key_ref"` + SecretKeyRef string `toml:"secret_key_ref"` +} + +// archive store backends recognized by §19. +var archiveStores = map[string]struct{}{ + "tarLocal": {}, + "tarS3": {}, + "volumeSnapshot": {}, + "longhorn": {}, +} + +// archive store backends this build can actually honor. §19 names four, but only +// tarLocal is implemented: the reaper's buildArchiver, the `felis restore` +// command, and the felis-api restore executor all construct tarLocal and nothing +// else. A config naming a recognized-but-unimplemented store is a footgun — it +// clears the "is this a real store name" check yet silently breaks retention (the +// reaper CronJob fails every run) and restore (503), while felis-api otherwise +// looks healthy. Validate rejects it so every binary that loads config (migrate, +// api, reaper) fails fast at startup with a clear remediation instead. (`felis +// restore` enforces the same invariant on its own --store flag: it runs inside +// the sandboxed weak-SA restore Job and by design never loads felis.toml or holds +// DB credentials, so it cannot lean on this load-time check.) +var implementedArchiveStores = map[string]struct{}{ + "tarLocal": {}, +} + +// Defaults that callers get when the field is omitted. +const ( + defaultListen = "0.0.0.0:8080" + defaultNamespace = "minecraft" + defaultEgressMode = "loadbalancer" + defaultStore = "tarLocal" + // defaultUserUploadsContext is a non-empty, platform-namespaced placeholder so + // the modpack approval lane's derived context ref is well-formed even before a + // deployment points it at a real object store. The blob transport is deferred, + // so this base only has to be a sensible, parseable prefix (see the §16 build + // subsystem and the internal/submit package doc for the lane's provenance). + defaultUserUploadsContext = "s3://felis-user-uploads" +) + +// Load reads and validates a felis.toml from path. +func Load(path string) (*Config, error) { + var cfg Config + md, err := toml.DecodeFile(path, &cfg) + if err != nil { + return nil, fmt.Errorf("config: decode %s: %w", path, err) + } + if undecoded := md.Undecoded(); len(undecoded) > 0 { + // Surface typos rather than silently ignoring unknown keys. + keys := make([]string, len(undecoded)) + for i, k := range undecoded { + keys[i] = k.String() + } + return nil, fmt.Errorf("config: unknown keys in %s: %s", path, strings.Join(keys, ", ")) + } + cfg.applyDefaults() + if err := cfg.Validate(); err != nil { + return nil, err + } + return &cfg, nil +} + +func (c *Config) applyDefaults() { + if c.Server.Listen == "" { + c.Server.Listen = defaultListen + } + if c.K8s.Namespace == "" { + c.K8s.Namespace = defaultNamespace + } + if c.K8s.EgressMode == "" { + c.K8s.EgressMode = defaultEgressMode + } + if c.Archive.Store == "" { + c.Archive.Store = defaultStore + } + if c.Registry.UserUploadsContext == "" { + c.Registry.UserUploadsContext = defaultUserUploadsContext + } +} + +// Validate enforces the mandatory fields (spec §24: database.url is 强制) and +// the closed value sets. +func (c *Config) Validate() error { + if c.Database.URL == "" { + return fmt.Errorf("config: [database] url is required") + } + if c.Server.RootDomain == "" { + return fmt.Errorf("config: [server] root_domain is required") + } + if !strings.Contains(c.Server.RootDomain, ".") { + return fmt.Errorf("config: [server] root_domain %q is not a domain", c.Server.RootDomain) + } + if _, ok := archiveStores[c.Archive.Store]; !ok { + return fmt.Errorf("config: [archive] store %q is not one of tarLocal|tarS3|volumeSnapshot|longhorn", c.Archive.Store) + } + if _, ok := implementedArchiveStores[c.Archive.Store]; !ok { + return fmt.Errorf("config: [archive] store %q is recognized by §19 but not implemented in this build — only tarLocal is supported; set store = \"tarLocal\"", c.Archive.Store) + } + switch c.K8s.EgressMode { + case "loadbalancer", "nodeport": + default: + return fmt.Errorf("config: [k8s] egress_mode %q must be loadbalancer or nodeport", c.K8s.EgressMode) + } + // The registry URL is a bare host[:port] (spec §24: url="registry.felis.svc:5000"), + // never a scheme-qualified URL. This is not cosmetic: two consumers read it with + // different robustness. The admin build path normalizes via registryHost() (which + // strips a scheme), but the user-modpack approval lane derives its push target by + // string concatenation (submit.Manager.deriveImageRef → "{url}/user-uploads/{id}:latest") + // with no stripping. A "http://" prefix would make the lane's pre-CAS build.Validate + // reject every derived ref (imageNameRE forbids the leading "http:/…") and collapse + // EVERY approve to 500 while admin builds keep working — a silent split-brain. Fail + // fast at load instead, with the contract spelled out. + if c.Registry.URL != "" && strings.Contains(c.Registry.URL, "://") { + return fmt.Errorf("config: [registry] url %q must be a bare host[:port] with no scheme (e.g. registry.felis.svc:5000); a scheme breaks the user-modpack build lane's derived push target", c.Registry.URL) + } + return nil +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..6049a92 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,213 @@ +package config_test + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "felis.lolicon.best/internal/config" +) + +// writeTOML writes content to a temp felis.toml and returns its path. +func writeTOML(t *testing.T, content string) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "felis.toml") + if err := os.WriteFile(path, []byte(content), 0o600); err != nil { + t.Fatalf("write toml: %v", err) + } + return path +} + +// Note: tests use the neutral example domain mc.example.net, never a real +// deployment domain, to keep the source tree clean of domain literals. +const validTOML = ` +[server] +listen = "0.0.0.0:9090" +root_domain = "mc.example.net" + +[database] +url = "postgres://felis:secret@db:5432/felis" + +[velocity] +public_ip = "203.0.113.4" +service_token_ref = "felis-velocity-token" + +[auth] +admin_hostname = "admin.example.net" +panel_hostname = "panel.example.net" +access_jwt_aud = "felis-panel" + +[k8s] +namespace = "minecraft" +egress_mode = "loadbalancer" +metallb_pool = "192.0.2.200-250" + +[registry] +url = "registry.felis.svc:5000" +build_namespace = "felis-build" + +[archive] +store = "tarLocal" +local_path = "backup-pvc" +retention = "3mo" +warn_before = ["3d", "1d"] +max_local_bytes = "200Gi" +[archive.s3] +endpoint = "" +bucket = "felis-backups" +access_key_ref = "" +secret_key_ref = "" +` + +func TestLoadValid(t *testing.T) { + cfg, err := config.Load(writeTOML(t, validTOML)) + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.Server.Listen != "0.0.0.0:9090" { + t.Errorf("listen = %q", cfg.Server.Listen) + } + if cfg.Server.RootDomain != "mc.example.net" { + t.Errorf("root_domain = %q", cfg.Server.RootDomain) + } + if cfg.Database.URL == "" { + t.Error("database url empty") + } + if cfg.Archive.Store != "tarLocal" { + t.Errorf("archive store = %q", cfg.Archive.Store) + } + if len(cfg.Archive.WarnBefore) != 2 || cfg.Archive.WarnBefore[0] != "3d" { + t.Errorf("warn_before = %v", cfg.Archive.WarnBefore) + } + if cfg.Archive.S3.Bucket != "felis-backups" { + t.Errorf("s3 bucket = %q", cfg.Archive.S3.Bucket) + } +} + +func TestLoadAppliesDefaults(t *testing.T) { + cfg, err := config.Load(writeTOML(t, ` +[server] +root_domain = "mc.example.net" +[database] +url = "postgres://felis@db/felis" +`)) + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.Server.Listen != "0.0.0.0:8080" { + t.Errorf("default listen = %q, want 0.0.0.0:8080", cfg.Server.Listen) + } + if cfg.K8s.Namespace != "minecraft" { + t.Errorf("default namespace = %q, want minecraft", cfg.K8s.Namespace) + } + if cfg.K8s.EgressMode != "loadbalancer" { + t.Errorf("default egress_mode = %q", cfg.K8s.EgressMode) + } + if cfg.Archive.Store != "tarLocal" { + t.Errorf("default archive store = %q", cfg.Archive.Store) + } +} + +func TestLoadRejectsMissingDatabaseURL(t *testing.T) { + _, err := config.Load(writeTOML(t, ` +[server] +root_domain = "mc.example.net" +`)) + if err == nil { + t.Fatal("expected error when database.url is missing") + } +} + +func TestLoadRejectsMissingRootDomain(t *testing.T) { + _, err := config.Load(writeTOML(t, ` +[database] +url = "postgres://felis@db/felis" +`)) + if err == nil { + t.Fatal("expected error when root_domain is missing") + } +} + +func TestLoadRejectsUnknownArchiveStore(t *testing.T) { + _, err := config.Load(writeTOML(t, ` +[server] +root_domain = "mc.example.net" +[database] +url = "postgres://felis@db/felis" +[archive] +store = "magicbox" +`)) + if err == nil { + t.Fatal("expected error for unknown archive store") + } +} + +// TestLoadRejectsUnimplementedArchiveStore guards the §19/build-reality gap: +// tarS3, volumeSnapshot and longhorn are recognized store names but only +// tarLocal is implemented in this build. A config naming one of them must be +// rejected at load — otherwise felis-api boots green while the reaper CronJob +// fails every run and restore silently 503s. The error must point the operator +// at the fix (tarLocal), distinct from the "unknown store" message. +func TestLoadRejectsUnimplementedArchiveStore(t *testing.T) { + for _, store := range []string{"tarS3", "volumeSnapshot", "longhorn"} { + t.Run(store, func(t *testing.T) { + _, err := config.Load(writeTOML(t, ` +[server] +root_domain = "mc.example.net" +[database] +url = "postgres://felis@db/felis" +[archive] +store = "`+store+`" +`)) + if err == nil { + t.Fatalf("expected error for recognized-but-unimplemented store %q", store) + } + if !strings.Contains(err.Error(), "tarLocal") { + t.Errorf("error for %q should point at the tarLocal remediation, got: %v", store, err) + } + }) + } +} + +// TestLoadRejectsSchemeQualifiedRegistryURL guards the §24 split-brain: the +// registry url is a bare host[:port], read scheme-tolerantly by the admin build +// path (registryHost strips the scheme) but scheme-INtolerantly by the +// user-modpack lane (deriveImageRef concatenates raw). A scheme-qualified url +// would boot felis-api green and 500 every approve while admin builds keep +// working, so it must be rejected at load with the bare-host contract spelled +// out. Both http:// and https:// are caught (the check is on "://"). +func TestLoadRejectsSchemeQualifiedRegistryURL(t *testing.T) { + for _, url := range []string{"http://registry.felis.svc:5000", "https://registry.felis.svc:5000"} { + t.Run(url, func(t *testing.T) { + _, err := config.Load(writeTOML(t, ` +[server] +root_domain = "mc.example.net" +[database] +url = "postgres://felis@db/felis" +[registry] +url = "`+url+`" +`)) + if err == nil { + t.Fatalf("expected error for scheme-qualified registry url %q", url) + } + if !strings.Contains(err.Error(), "scheme") { + t.Errorf("error for %q should explain the bare-host contract, got: %v", url, err) + } + }) + } +} + +func TestLoadRejectsUnknownKeys(t *testing.T) { + _, err := config.Load(writeTOML(t, ` +[server] +root_domain = "mc.example.net" +typo_field = "oops" +[database] +url = "postgres://felis@db/felis" +`)) + if err == nil { + t.Fatal("expected error for unknown key") + } +} diff --git a/internal/naming/naming.go b/internal/naming/naming.go new file mode 100644 index 0000000..99393f1 --- /dev/null +++ b/internal/naming/naming.go @@ -0,0 +1,96 @@ +// Package naming enforces the portability and admission rules (spec §2, §22): +// a server name matches ^[a-z0-9-]{3,32}$ and is non-reserved, and every +// hostname must be a single label under the configured root_domain. The root +// domain is never hardcoded — it is always supplied by config — so this package +// stays free of any deployment-specific domain. +package naming + +import ( + "fmt" + "regexp" + "strings" +) + +var ( + serverNameRE = regexp.MustCompile(`^[a-z0-9-]{3,32}$`) + dnsLabelRE = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$`) +) + +// reserved subdomains/server names that users may not claim: proxy/lobby and +// the platform's own faces. +var reserved = map[string]struct{}{ + "lobby": {}, + "admin": {}, + "panel": {}, + "api": {}, + "felis": {}, + "velocity": {}, + "registry": {}, + "internal": {}, + "www": {}, +} + +// ValidateServerName checks the §22 name rule and reservation list. +func ValidateServerName(name string) error { + if !serverNameRE.MatchString(name) { + return fmt.Errorf("naming: invalid server name %q: must match ^[a-z0-9-]{3,32}$", name) + } + if strings.HasPrefix(name, "-") || strings.HasSuffix(name, "-") { + return fmt.Errorf("naming: server name %q must not start or end with '-'", name) + } + if _, ok := reserved[name]; ok { + return fmt.Errorf("naming: server name %q is reserved", name) + } + return nil +} + +// IsReserved reports whether label is on the reserved list. +func IsReserved(label string) bool { + _, ok := reserved[label] + return ok +} + +// worldVolumeName mirrors operator.dataVolumeName: the per-server StatefulSet's +// volumeClaimTemplate is named "world", so a single-replica server's world PVC +// is "world--0". This is the one naming convention shared by the operator +// (which creates the PVC), the reaper (which deletes it), and restore (which +// mounts it), so it lives here rather than being duplicated per subsystem. +const worldVolumeName = "world" + +// WorldPVCName returns the world PersistentVolumeClaim name for a server, +// matching the operator's StatefulSet volumeClaimTemplate naming +// ("world--0" for the sole replica). +func WorldPVCName(server string) string { + return worldVolumeName + "-" + server + "-0" +} + +// Hostname composes subdomain.rootDomain after validating the subdomain. +func Hostname(subdomain, rootDomain string) (string, error) { + if err := ValidateServerName(subdomain); err != nil { + return "", err + } + if rootDomain == "" { + return "", fmt.Errorf("naming: root domain is empty") + } + return subdomain + "." + rootDomain, nil +} + +// ValidateHostname enforces the §2 invariant that host is a single label +// directly under rootDomain. +func ValidateHostname(host, rootDomain string) error { + if rootDomain == "" { + return fmt.Errorf("naming: root domain is empty") + } + suffix := "." + rootDomain + if !strings.HasSuffix(host, suffix) { + return fmt.Errorf("naming: hostname %q must be under %q", host, rootDomain) + } + label := strings.TrimSuffix(host, suffix) + if label == "" || strings.Contains(label, ".") { + return fmt.Errorf("naming: hostname %q must be a single label under %q", host, rootDomain) + } + if !dnsLabelRE.MatchString(label) { + return fmt.Errorf("naming: invalid hostname label %q", label) + } + return nil +} diff --git a/internal/naming/naming_test.go b/internal/naming/naming_test.go new file mode 100644 index 0000000..6b21f36 --- /dev/null +++ b/internal/naming/naming_test.go @@ -0,0 +1,95 @@ +package naming_test + +import ( + "testing" + + "felis.lolicon.best/internal/naming" +) + +func TestValidateServerName(t *testing.T) { + cases := []struct { + name string + ok bool + }{ + {"survival", true}, + {"creative-2", true}, + {"abc", true}, + {"a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6", true}, // 32 chars + {"ab", false}, // too short + {"a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6q", false}, // 33 chars + {"Survival", false}, // uppercase + {"has_underscore", false}, // illegal char + {"has space", false}, // illegal char + {"-leading", false}, // leading hyphen + {"trailing-", false}, // trailing hyphen + {"lobby", false}, // reserved + {"admin", false}, // reserved + {"api", false}, // reserved + } + for _, c := range cases { + err := naming.ValidateServerName(c.name) + if c.ok && err != nil { + t.Errorf("ValidateServerName(%q) = %v, want ok", c.name, err) + } + if !c.ok && err == nil { + t.Errorf("ValidateServerName(%q) = nil, want error", c.name) + } + } +} + +func TestWorldPVCName(t *testing.T) { + cases := map[string]string{ + "survival": "world-survival-0", + "creative-2": "world-creative-2-0", + } + for server, want := range cases { + if got := naming.WorldPVCName(server); got != want { + t.Errorf("WorldPVCName(%q) = %q, want %q", server, got, want) + } + } +} + +func TestHostname(t *testing.T) { + const root = "mc.example.net" + got, err := naming.Hostname("survival", root) + if err != nil { + t.Fatalf("Hostname: %v", err) + } + if got != "survival.mc.example.net" { + t.Errorf("Hostname = %q, want survival.mc.example.net", got) + } + if _, err := naming.Hostname("lobby", root); err == nil { + t.Error("Hostname should reject a reserved subdomain") + } + if _, err := naming.Hostname("survival", ""); err == nil { + t.Error("Hostname should reject an empty root domain") + } +} + +func TestValidateHostname(t *testing.T) { + const root = "mc.example.net" + cases := []struct { + host string + ok bool + }{ + {"survival.mc.example.net", true}, + {"a.mc.example.net", true}, + {"deep.sub.mc.example.net", false}, // not a single label under root + {"survival.evil.example.org", false}, + {"mc.example.net", false}, // bare root, no label + {".mc.example.net", false}, + {"-bad.mc.example.net", false}, + } + for _, c := range cases { + err := naming.ValidateHostname(c.host, root) + if c.ok && err != nil { + t.Errorf("ValidateHostname(%q) = %v, want ok", c.host, err) + } + if !c.ok && err == nil { + t.Errorf("ValidateHostname(%q) = nil, want error", c.host) + } + } + if err := naming.ValidateHostname("x.mc.example.net", ""); err == nil { + t.Error("ValidateHostname should reject an empty root domain") + } +} diff --git a/internal/rcon/rcon.go b/internal/rcon/rcon.go new file mode 100644 index 0000000..8f29f59 --- /dev/null +++ b/internal/rcon/rcon.go @@ -0,0 +1,188 @@ +// Package rcon implements a minimal Source RCON client (spec §5, §7). +// +// The operator uses it for two purposes: +// - Readiness probing: a successful Dial (TCP connect + auth) is the +// loader-agnostic "RCON 探通" gate. A status ping is never sufficient. +// - Graceful shutdown: Execute("save-all flush") then Execute("stop") from +// the operator-injected preStop hook. +// +// Multi-packet responses (a single command whose reply exceeds one ~4 KiB +// packet) are not reassembled; Phase-1 commands ("list", "save-all", "stop") +// always fit in one packet. This is intentional and documented rather than +// silently truncating large replies. +package rcon + +import ( + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "time" +) + +// Source RCON packet types. +const ( + typeResponseValue = 0 // SERVERDATA_RESPONSE_VALUE + typeExecCommand = 2 // SERVERDATA_EXECCOMMAND + typeAuthResponse = 2 // SERVERDATA_AUTH_RESPONSE (same id as EXECCOMMAND) + typeAuth = 3 // SERVERDATA_AUTH +) + +// authFailedID is the request id a server returns when auth fails. +const authFailedID int32 = -1 + +// Packet length bounds. minPacketLen is 4 (id) + 4 (type) + 2 (terminators). +const ( + minPacketLen = 10 + maxPacketLen = 4096 +) + +// ErrAuthFailed is returned by Dial when the RCON password is rejected. +var ErrAuthFailed = errors.New("rcon: authentication failed") + +// DefaultPort is the conventional RCON port. +const DefaultPort = 25575 + +// Conn is an authenticated RCON connection. It is not safe for concurrent use. +type Conn struct { + conn net.Conn + reqID int32 +} + +// Dial opens a TCP connection to addr and authenticates with password. The +// timeout, if > 0, bounds the whole connect+auth handshake; it is cleared on +// the returned Conn so subsequent calls block unless SetDeadline is used. +func Dial(addr, password string, timeout time.Duration) (*Conn, error) { + var ( + netConn net.Conn + err error + ) + if timeout > 0 { + netConn, err = net.DialTimeout("tcp", addr, timeout) + } else { + netConn, err = net.Dial("tcp", addr) + } + if err != nil { + return nil, err + } + + c := &Conn{conn: netConn} + if timeout > 0 { + if err := netConn.SetDeadline(time.Now().Add(timeout)); err != nil { + netConn.Close() + return nil, err + } + } + if err := c.auth(password); err != nil { + netConn.Close() + return nil, err + } + // Clear the handshake deadline so the connection is reusable. + if err := netConn.SetDeadline(time.Time{}); err != nil { + netConn.Close() + return nil, err + } + return c, nil +} + +// SetDeadline sets an absolute deadline for subsequent Execute calls. +func (c *Conn) SetDeadline(t time.Time) error { return c.conn.SetDeadline(t) } + +// Close closes the underlying connection. +func (c *Conn) Close() error { return c.conn.Close() } + +// Execute runs a single command and returns the server's reply body. +func (c *Conn) Execute(cmd string) (string, error) { + id := c.nextID() + if err := writePacket(c.conn, id, typeExecCommand, cmd); err != nil { + return "", err + } + respID, _, body, err := readPacket(c.conn) + if err != nil { + return "", err + } + if respID != id { + return "", fmt.Errorf("rcon: response id mismatch: got %d want %d", respID, id) + } + return body, nil +} + +// auth performs the SERVERDATA_AUTH handshake. Some servers emit an empty +// RESPONSE_VALUE before the auth response, so non-auth packets are skipped. +func (c *Conn) auth(password string) error { + id := c.nextID() + if err := writePacket(c.conn, id, typeAuth, password); err != nil { + return err + } + for { + respID, respType, _, err := readPacket(c.conn) + if err != nil { + return err + } + if respType != typeAuthResponse { + // Empty RESPONSE_VALUE echo; keep reading for the real answer. + continue + } + if respID == authFailedID { + return ErrAuthFailed + } + if respID != id { + return fmt.Errorf("rcon: auth id mismatch: got %d want %d", respID, id) + } + return nil + } +} + +func (c *Conn) nextID() int32 { + c.reqID++ + if c.reqID < 0 { + c.reqID = 1 + } + return c.reqID +} + +// writePacket encodes one RCON packet: little-endian length, id, type, the +// null-terminated body, and a trailing null byte. +func writePacket(w io.Writer, id, typ int32, body string) error { + bodyBytes := []byte(body) + length := int32(4 + 4 + len(bodyBytes) + 2) + if length > maxPacketLen { + return fmt.Errorf("rcon: outgoing packet too large: %d bytes", length) + } + buf := make([]byte, 0, 4+length) + buf = appendInt32(buf, length) + buf = appendInt32(buf, id) + buf = appendInt32(buf, typ) + buf = append(buf, bodyBytes...) + buf = append(buf, 0, 0) + _, err := w.Write(buf) + return err +} + +// readPacket decodes one RCON packet. +func readPacket(r io.Reader) (id, typ int32, body string, err error) { + var lenBuf [4]byte + if _, err = io.ReadFull(r, lenBuf[:]); err != nil { + return 0, 0, "", err + } + length := int32(binary.LittleEndian.Uint32(lenBuf[:])) + if length < minPacketLen || length > maxPacketLen { + return 0, 0, "", fmt.Errorf("rcon: invalid packet length %d", length) + } + payload := make([]byte, length) + if _, err = io.ReadFull(r, payload); err != nil { + return 0, 0, "", err + } + id = int32(binary.LittleEndian.Uint32(payload[0:4])) + typ = int32(binary.LittleEndian.Uint32(payload[4:8])) + // Strip the two trailing null bytes from the body. + body = string(payload[8 : length-2]) + return id, typ, body, nil +} + +func appendInt32(buf []byte, v int32) []byte { + var b [4]byte + binary.LittleEndian.PutUint32(b[:], uint32(v)) + return append(buf, b[:]...) +} diff --git a/internal/rcon/rcon_test.go b/internal/rcon/rcon_test.go new file mode 100644 index 0000000..5280b43 --- /dev/null +++ b/internal/rcon/rcon_test.go @@ -0,0 +1,179 @@ +package rcon_test + +import ( + "encoding/binary" + "errors" + "io" + "net" + "sync" + "testing" + "time" + + "felis.lolicon.best/internal/rcon" +) + +// fakeRCON is a minimal Source RCON server used to exercise the client +// hermetically (no real Minecraft server, no network beyond loopback). +type fakeRCON struct { + ln net.Listener + password string + replies map[string]string + wg sync.WaitGroup +} + +func startFakeRCON(t *testing.T, password string, replies map[string]string) *fakeRCON { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + f := &fakeRCON{ln: ln, password: password, replies: replies} + f.wg.Add(1) + go f.serve() + return f +} + +func (f *fakeRCON) addr() string { return f.ln.Addr().String() } + +func (f *fakeRCON) stop() { + f.ln.Close() + f.wg.Wait() +} + +func (f *fakeRCON) serve() { + defer f.wg.Done() + for { + conn, err := f.ln.Accept() + if err != nil { + return + } + go f.handle(conn) + } +} + +func (f *fakeRCON) handle(conn net.Conn) { + defer conn.Close() + authed := false + for { + id, typ, body, err := readFramePacket(conn) + if err != nil { + return + } + switch typ { + case 3: // AUTH + if body == f.password { + authed = true + _ = writeFramePacket(conn, id, 2, "") // AUTH_RESPONSE, echo id + } else { + _ = writeFramePacket(conn, -1, 2, "") // failure + } + case 2: // EXECCOMMAND + if !authed { + _ = writeFramePacket(conn, -1, 0, "") + continue + } + reply := f.replies[body] + _ = writeFramePacket(conn, id, 0, reply) + default: + _ = writeFramePacket(conn, id, 0, "") + } + } +} + +func writeFramePacket(w io.Writer, id, typ int32, body string) error { + b := []byte(body) + length := int32(4 + 4 + len(b) + 2) + buf := make([]byte, 0, 4+length) + buf = putI32(buf, length) + buf = putI32(buf, id) + buf = putI32(buf, typ) + buf = append(buf, b...) + buf = append(buf, 0, 0) + _, err := w.Write(buf) + return err +} + +func readFramePacket(r io.Reader) (id, typ int32, body string, err error) { + var lenBuf [4]byte + if _, err = io.ReadFull(r, lenBuf[:]); err != nil { + return 0, 0, "", err + } + length := int32(binary.LittleEndian.Uint32(lenBuf[:])) + payload := make([]byte, length) + if _, err = io.ReadFull(r, payload); err != nil { + return 0, 0, "", err + } + id = int32(binary.LittleEndian.Uint32(payload[0:4])) + typ = int32(binary.LittleEndian.Uint32(payload[4:8])) + body = string(payload[8 : length-2]) + return id, typ, body, nil +} + +func putI32(buf []byte, v int32) []byte { + var b [4]byte + binary.LittleEndian.PutUint32(b[:], uint32(v)) + return append(buf, b[:]...) +} + +func TestDialAndExecute(t *testing.T) { + want := "There are 0 of a max of 20 players online:" + f := startFakeRCON(t, "s3cret", map[string]string{"list": want}) + defer f.stop() + + c, err := rcon.Dial(f.addr(), "s3cret", 2*time.Second) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer c.Close() + + got, err := c.Execute("list") + if err != nil { + t.Fatalf("Execute: %v", err) + } + if got != want { + t.Errorf("Execute(list) = %q, want %q", got, want) + } +} + +func TestDialAuthFailure(t *testing.T) { + f := startFakeRCON(t, "correct-horse", nil) + defer f.stop() + + _, err := rcon.Dial(f.addr(), "wrong-password", 2*time.Second) + if !errors.Is(err, rcon.ErrAuthFailed) { + t.Fatalf("Dial with wrong password: got %v, want ErrAuthFailed", err) + } +} + +func TestDialUnreachable(t *testing.T) { + // Reserved TEST-NET-1 address: connect should fail fast within the timeout. + _, err := rcon.Dial("192.0.2.1:25575", "x", 200*time.Millisecond) + if err == nil { + t.Fatal("Dial to unreachable host: expected error, got nil") + } + if errors.Is(err, rcon.ErrAuthFailed) { + t.Fatalf("Dial to unreachable host: got ErrAuthFailed, want a dial error") + } +} + +func TestExecuteGracefulShutdownSequence(t *testing.T) { + // Mirrors the operator preStop hook: save then stop. + f := startFakeRCON(t, "pw", map[string]string{ + "save-all flush": "Saved the game", + "stop": "Stopping the server", + }) + defer f.stop() + + c, err := rcon.Dial(f.addr(), "pw", 2*time.Second) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer c.Close() + + if out, err := c.Execute("save-all flush"); err != nil || out != "Saved the game" { + t.Fatalf("save-all flush = %q, %v", out, err) + } + if out, err := c.Execute("stop"); err != nil || out != "Stopping the server" { + t.Fatalf("stop = %q, %v", out, err) + } +} diff --git a/internal/store/migrate.go b/internal/store/migrate.go new file mode 100644 index 0000000..4d43f7c --- /dev/null +++ b/internal/store/migrate.go @@ -0,0 +1,136 @@ +// Package store owns the Felis business-layer database (spec §6): the embedded +// schema migrations and the typed access layer. Migrations are applied by +// `felis migrate up` under a Postgres advisory lock so concurrent api/operator +// replicas cannot race each other. +package store + +import ( + "context" + "embed" + "fmt" + "io/fs" + "sort" + "strconv" + "strings" +) + +// AdvisoryLockKey is the fixed pg_advisory_lock key guarding migrations. The +// value is the ASCII bytes of "felis"; any replica running migrations contends +// on the same key. +const AdvisoryLockKey int64 = 0x66656c6973 // "felis" + +//go:embed migrations/*.sql +var migrationsFS embed.FS + +// Migration is a single ordered schema step loaded from the embedded FS. +type Migration struct { + Version int + Name string + SQL string +} + +// Driver is the database-facing seam the migration engine drives. Splitting it +// out lets the ordering/idempotency/lock logic be tested without a live +// Postgres; PostgresDriver is the production implementation. +type Driver interface { + // Lock acquires the migration advisory lock, blocking until held. + Lock(ctx context.Context) error + // Unlock releases the advisory lock. + Unlock(ctx context.Context) error + // EnsureVersionTable creates the schema_migrations bookkeeping table. + EnsureVersionTable(ctx context.Context) error + // AppliedVersions returns the set of versions already applied. + AppliedVersions(ctx context.Context) (map[int]struct{}, error) + // Apply runs one migration and records it, atomically. + Apply(ctx context.Context, m Migration) error +} + +// LoadMigrations parses the embedded migrations into an ascending, gap-tolerant +// but duplicate-free list. +func LoadMigrations() ([]Migration, error) { + entries, err := fs.ReadDir(migrationsFS, "migrations") + if err != nil { + return nil, fmt.Errorf("read embedded migrations: %w", err) + } + var ms []Migration + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") { + continue + } + version, name, err := parseMigrationName(e.Name()) + if err != nil { + return nil, err + } + data, err := migrationsFS.ReadFile("migrations/" + e.Name()) + if err != nil { + return nil, fmt.Errorf("read migration %q: %w", e.Name(), err) + } + if strings.TrimSpace(string(data)) == "" { + return nil, fmt.Errorf("migration %q is empty", e.Name()) + } + ms = append(ms, Migration{Version: version, Name: name, SQL: string(data)}) + } + sort.Slice(ms, func(i, j int) bool { return ms[i].Version < ms[j].Version }) + for i := 1; i < len(ms); i++ { + if ms[i].Version == ms[i-1].Version { + return nil, fmt.Errorf("duplicate migration version %d (%s, %s)", ms[i].Version, ms[i-1].Name, ms[i].Name) + } + } + if len(ms) == 0 { + return nil, fmt.Errorf("no migrations found") + } + return ms, nil +} + +// Up applies every pending migration in ascending order, exactly once, under +// the advisory lock. It is safe to run concurrently from multiple replicas: the +// lock serializes them and AppliedVersions makes the work idempotent. +func Up(ctx context.Context, d Driver, migrations []Migration) (applied []int, err error) { + if err := d.Lock(ctx); err != nil { + return nil, fmt.Errorf("acquire migration lock: %w", err) + } + defer func() { + if uerr := d.Unlock(ctx); uerr != nil && err == nil { + err = fmt.Errorf("release migration lock: %w", uerr) + } + }() + + if err := d.EnsureVersionTable(ctx); err != nil { + return nil, fmt.Errorf("ensure version table: %w", err) + } + done, err := d.AppliedVersions(ctx) + if err != nil { + return nil, fmt.Errorf("read applied versions: %w", err) + } + + ordered := append([]Migration(nil), migrations...) + sort.Slice(ordered, func(i, j int) bool { return ordered[i].Version < ordered[j].Version }) + for _, m := range ordered { + if _, ok := done[m.Version]; ok { + continue + } + if err := d.Apply(ctx, m); err != nil { + return applied, fmt.Errorf("apply migration %04d_%s: %w", m.Version, m.Name, err) + } + applied = append(applied, m.Version) + } + return applied, nil +} + +// parseMigrationName turns "0001_init.sql" into (1, "init"). +func parseMigrationName(filename string) (int, string, error) { + base := strings.TrimSuffix(filename, ".sql") + idx := strings.IndexByte(base, '_') + if idx <= 0 { + return 0, "", fmt.Errorf("migration %q must be named NNNN_name.sql", filename) + } + version, err := strconv.Atoi(base[:idx]) + if err != nil { + return 0, "", fmt.Errorf("migration %q has a non-numeric version: %w", filename, err) + } + name := base[idx+1:] + if name == "" { + return 0, "", fmt.Errorf("migration %q is missing a name", filename) + } + return version, name, nil +} diff --git a/internal/store/migrate_test.go b/internal/store/migrate_test.go new file mode 100644 index 0000000..5e8035e --- /dev/null +++ b/internal/store/migrate_test.go @@ -0,0 +1,157 @@ +package store_test + +import ( + "context" + "errors" + "strings" + "testing" + + "felis.lolicon.best/internal/store" +) + +func TestLoadMigrationsOrderedAndWellFormed(t *testing.T) { + ms, err := store.LoadMigrations() + if err != nil { + t.Fatalf("LoadMigrations: %v", err) + } + if len(ms) == 0 { + t.Fatal("expected at least one migration") + } + if ms[0].Version != 1 || ms[0].Name != "init" { + t.Errorf("first migration = %d_%s, want 0001_init", ms[0].Version, ms[0].Name) + } + for i := 1; i < len(ms); i++ { + if ms[i].Version <= ms[i-1].Version { + t.Errorf("migrations not strictly ascending at %d: %d then %d", i, ms[i-1].Version, ms[i].Version) + } + } + // The init migration must define the core business tables (spec §6). + for _, want := range []string{"CREATE TABLE users", "CREATE TABLE servers", "CREATE TABLE world_backups"} { + if !strings.Contains(ms[0].SQL, want) { + t.Errorf("init migration missing %q", want) + } + } +} + +// recordingDriver captures the migration engine's calls without a database. +type recordingDriver struct { + already map[int]struct{} + applied []int + locked bool + unlocked bool + ensured bool + appliedWhileUnsafe bool // true if Apply ran while not locked or already unlocked + failOn int // version whose Apply should fail (0 = never) +} + +func (d *recordingDriver) Lock(context.Context) error { d.locked = true; return nil } +func (d *recordingDriver) Unlock(context.Context) error { d.unlocked = true; return nil } +func (d *recordingDriver) EnsureVersionTable(context.Context) error { + if !d.locked || d.unlocked { + d.appliedWhileUnsafe = true + } + d.ensured = true + return nil +} +func (d *recordingDriver) AppliedVersions(context.Context) (map[int]struct{}, error) { + if d.already == nil { + return map[int]struct{}{}, nil + } + return d.already, nil +} +func (d *recordingDriver) Apply(_ context.Context, m store.Migration) error { + if !d.locked || d.unlocked { + d.appliedWhileUnsafe = true + } + if d.failOn != 0 && m.Version == d.failOn { + return errors.New("boom") + } + d.applied = append(d.applied, m.Version) + return nil +} + +func TestUpAppliesAllPendingInOrder(t *testing.T) { + d := &recordingDriver{} + ms := []store.Migration{ + {Version: 3, Name: "c", SQL: "x"}, + {Version: 1, Name: "a", SQL: "y"}, + {Version: 2, Name: "b", SQL: "z"}, + } + applied, err := store.Up(context.Background(), d, ms) + if err != nil { + t.Fatalf("Up: %v", err) + } + if got := strings.Trim(strings.Join(intsToStrings(applied), ","), ""); got != "1,2,3" { + t.Errorf("applied = %v, want [1 2 3]", applied) + } + if !d.locked || !d.unlocked || !d.ensured { + t.Errorf("lifecycle flags: locked=%v unlocked=%v ensured=%v", d.locked, d.unlocked, d.ensured) + } + if d.appliedWhileUnsafe { + t.Error("work ran outside the advisory lock") + } +} + +func TestUpSkipsAlreadyApplied(t *testing.T) { + d := &recordingDriver{already: map[int]struct{}{1: {}}} + ms := []store.Migration{ + {Version: 1, Name: "a", SQL: "y"}, + {Version: 2, Name: "b", SQL: "z"}, + } + applied, err := store.Up(context.Background(), d, ms) + if err != nil { + t.Fatalf("Up: %v", err) + } + if len(applied) != 1 || applied[0] != 2 { + t.Errorf("applied = %v, want [2]", applied) + } +} + +func TestUpStopsOnErrorButStillUnlocks(t *testing.T) { + d := &recordingDriver{failOn: 2} + ms := []store.Migration{ + {Version: 1, Name: "a", SQL: "y"}, + {Version: 2, Name: "b", SQL: "z"}, + {Version: 3, Name: "c", SQL: "x"}, + } + applied, err := store.Up(context.Background(), d, ms) + if err == nil { + t.Fatal("expected an error when a migration fails") + } + if len(applied) != 1 || applied[0] != 1 { + t.Errorf("applied = %v, want only [1] before the failure", applied) + } + if !d.unlocked { + t.Error("advisory lock must be released even when a migration fails") + } +} + +func intsToStrings(in []int) []string { + out := make([]string, len(in)) + for i, v := range in { + out[i] = itoa(v) + } + return out +} + +func itoa(v int) string { + if v == 0 { + return "0" + } + neg := v < 0 + if neg { + v = -v + } + var buf [20]byte + i := len(buf) + for v > 0 { + i-- + buf[i] = byte('0' + v%10) + v /= 10 + } + if neg { + i-- + buf[i] = '-' + } + return string(buf[i:]) +} diff --git a/internal/store/migrations/0001_init.sql b/internal/store/migrations/0001_init.sql new file mode 100644 index 0000000..7ec7e1f --- /dev/null +++ b/internal/store/migrations/0001_init.sql @@ -0,0 +1,75 @@ +-- Felis business-layer schema (spec §6). The CRD is the lifecycle +-- source-of-truth; this database owns only what the CRD cannot express: +-- ownership/claim, account links, quotas, image admission, builds, backups, +-- audit. Authoritative CRD fields are never duplicated here. + +CREATE TYPE user_role AS ENUM ('admin','user'); +CREATE TYPE build_status AS ENUM ('pending','building','succeeded','failed','cancelled'); +CREATE TYPE backup_status AS ENUM ('present','expired','deleted'); + +CREATE TABLE users ( + id text PRIMARY KEY, username text UNIQUE NOT NULL, email text, + role user_role NOT NULL DEFAULT 'user', created_at timestamptz NOT NULL DEFAULT now() +); + +-- Identity bridge: web identity <-> MC UUID (claim/owner-only/allowlist rely on it). +CREATE TABLE account_links ( + user_id text NOT NULL REFERENCES users(id), mc_uuid uuid NOT NULL, + verified_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (user_id, mc_uuid), UNIQUE (mc_uuid) +); +CREATE TABLE account_link_codes ( code text PRIMARY KEY, mc_uuid uuid NOT NULL, expires_at timestamptz NOT NULL ); + +CREATE TABLE quotas ( + user_id text PRIMARY KEY REFERENCES users(id), + max_servers int, max_cpu_milli int, max_memory_mb int, max_storage_gb int +); + +-- Business projection: the CRD lives in K8s; this stores only the +-- ownership/activity/warning that the CRD cannot express, plus a fast-query cache. +CREATE TABLE servers ( + name text PRIMARY KEY, -- matches CRD metadata.name + owner_id text REFERENCES users(id), -- NULL until claimed; reaper resets to NULL + claimed_at timestamptz, + last_active_at timestamptz NOT NULL DEFAULT now(), -- max(last human join, created_at) + warned_3d_at timestamptz, warned_1d_at timestamptz, -- reaper warning dedup; cleared on renewal + cached_phase text, -- CRD status projection, non-authoritative + created_at timestamptz NOT NULL DEFAULT now(), deleted_at timestamptz +); +CREATE TABLE server_aliases ( subdomain text PRIMARY KEY, server_name text NOT NULL REFERENCES servers(name) ); +CREATE TABLE server_allowlist ( -- autostartPolicy=allowlist; first join auto-appends + server_name text NOT NULL REFERENCES servers(name), mc_uuid uuid NOT NULL, + added_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (server_name, mc_uuid) +); + +-- Image admission (dynamic, auditable -> DB, not toml). +CREATE TABLE image_whitelist ( + image_ref text PRIMARY KEY, -- registry/foo:1.0 or registry/foo:* + source text NOT NULL DEFAULT 'built', -- built (cluster build) | external (pushed) + build_id text, added_by text NOT NULL, enabled boolean NOT NULL DEFAULT true, + added_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE image_builds ( + id text PRIMARY KEY, image_ref text NOT NULL, status build_status NOT NULL DEFAULT 'pending', + dockerfile text NOT NULL, -- archived for audit + context_ref text, base_image text, -- resolved FROM, audit + requested_by text NOT NULL, job_name text, log_ref text, error text, + created_at timestamptz NOT NULL DEFAULT now(), finished_at timestamptz +); + +-- World backups (reaper output; not FK'd to servers, which may be reset/deleted). +CREATE TABLE world_backups ( + id text PRIMARY KEY, server_name text NOT NULL, former_owner text, + backup_ref text NOT NULL, -- WorldArchiver location (ArchiveRef) + size_bytes bigint, reason text NOT NULL, -- inactive_15d | manual + status backup_status NOT NULL DEFAULT 'present', + created_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL, -- created_at + 3mo + deleted_at timestamptz +); + +CREATE TABLE audit_logs ( + id bigserial PRIMARY KEY, actor text NOT NULL, source text NOT NULL, action text NOT NULL, + server_name text, request_id text, payload jsonb, created_at timestamptz NOT NULL DEFAULT now() +); +CREATE TABLE tokens ( id text PRIMARY KEY, name text NOT NULL, token_hash text NOT NULL, scope jsonb NOT NULL, expires_at timestamptz ); diff --git a/internal/store/migrations/0002_image_submissions.sql b/internal/store/migrations/0002_image_submissions.sql new file mode 100644 index 0000000..856ff84 --- /dev/null +++ b/internal/store/migrations/0002_image_submissions.sql @@ -0,0 +1,42 @@ +-- User-submitted modpack approval lane (a user-directed extension over the §16 +-- build subsystem; see internal/submit for provenance). This is the UNTRUSTED- +-- origin counterpart to the admin build path (POST /images/build): an ordinary +-- user may upload a modpack but cannot start a build directly. Each upload lands +-- here as pending_review; an admin must approve it before anything is built, and +-- the approved submission then routes through the SAME Trivy-gated Kaniko build +-- as an admin build (build subsystem §16). Approval is a human gate layered in +-- FRONT of the automatic scan, never instead of it — a CRITICAL CVE still fails +-- the build and nothing is admitted even after a human approved. +-- +-- Source of truth (spec §1): this row is the Postgres BUSINESS authority for the +-- approval (verdict + reviewer); the build EXECUTION lives in image_builds, +-- linked by build_id once Builder.Submit succeeds. The approval never copies the +-- build's authoritative fields. +-- +-- Trust note: the platform derives BOTH the push target (image_ref) and the +-- build context (context_ref) from the submission id — neither is free-form user +-- input — so an untrusted submitter can never point the build at an arbitrary +-- source or collide with the platform image namespace. There is deliberately no +-- `origin` column: image_submissions is ONLY the user-upload lane (the platform +-- uses the direct build path), and submitted_by already records the origin. + +CREATE TYPE submission_status AS ENUM ('pending_review','approved','rejected'); + +CREATE TABLE image_submissions ( + id text PRIMARY KEY, -- lowercase, namespaces the derived image/context refs + submitted_by text NOT NULL, -- uploading user's id (untrusted origin) + display_name text NOT NULL, -- human-friendly label for the modpack + context_ref text NOT NULL, -- DERIVED pinned build context (not user-supplied) + status submission_status NOT NULL DEFAULT 'pending_review', + image_ref text, -- DERIVED {registry}/user-uploads/{id}:latest, set at approve + build_id text, -- image_builds.id, set only after Builder.Submit succeeds + reviewed_by text, -- admin who approved/rejected + reject_reason text, -- set on rejection + created_at timestamptz NOT NULL DEFAULT now(), + reviewed_at timestamptz +); + +-- The admin review queue scans by status (pending first); the per-user index +-- serves the "my submissions" list. +CREATE INDEX image_submissions_status_idx ON image_submissions (status, created_at); +CREATE INDEX image_submissions_submitted_by_idx ON image_submissions (submitted_by, created_at DESC); diff --git a/internal/store/sqldriver.go b/internal/store/sqldriver.go new file mode 100644 index 0000000..2f8fd05 --- /dev/null +++ b/internal/store/sqldriver.go @@ -0,0 +1,94 @@ +package store + +import ( + "context" + "database/sql" + "fmt" + + _ "github.com/jackc/pgx/v5/stdlib" // register the "pgx" database/sql driver +) + +// PostgresDriver is the production Driver, backed by a database/sql pool using +// the pgx stdlib driver. +type PostgresDriver struct { + db *sql.DB +} + +// Open dials dsn and returns a PostgresDriver. The caller owns Close. +func Open(ctx context.Context, dsn string) (*PostgresDriver, error) { + db, err := sql.Open("pgx", dsn) + if err != nil { + return nil, fmt.Errorf("open postgres: %w", err) + } + if err := db.PingContext(ctx); err != nil { + db.Close() + return nil, fmt.Errorf("ping postgres: %w", err) + } + return &PostgresDriver{db: db}, nil +} + +// DB exposes the underlying pool for the access layer. +func (d *PostgresDriver) DB() *sql.DB { return d.db } + +// Close releases the pool. +func (d *PostgresDriver) Close() error { return d.db.Close() } + +// Lock takes the session-level advisory lock that serializes migrations. +func (d *PostgresDriver) Lock(ctx context.Context) error { + _, err := d.db.ExecContext(ctx, "SELECT pg_advisory_lock($1)", AdvisoryLockKey) + return err +} + +// Unlock releases the advisory lock. +func (d *PostgresDriver) Unlock(ctx context.Context) error { + _, err := d.db.ExecContext(ctx, "SELECT pg_advisory_unlock($1)", AdvisoryLockKey) + return err +} + +// EnsureVersionTable creates the bookkeeping table if absent. +func (d *PostgresDriver) EnsureVersionTable(ctx context.Context) error { + const ddl = `CREATE TABLE IF NOT EXISTS schema_migrations ( + version int PRIMARY KEY, + name text NOT NULL, + applied_at timestamptz NOT NULL DEFAULT now() + )` + _, err := d.db.ExecContext(ctx, ddl) + return err +} + +// AppliedVersions reads the set of recorded versions. +func (d *PostgresDriver) AppliedVersions(ctx context.Context) (map[int]struct{}, error) { + rows, err := d.db.QueryContext(ctx, "SELECT version FROM schema_migrations") + if err != nil { + return nil, err + } + defer rows.Close() + out := map[int]struct{}{} + for rows.Next() { + var v int + if err := rows.Scan(&v); err != nil { + return nil, err + } + out[v] = struct{}{} + } + return out, rows.Err() +} + +// Apply runs the migration body and records it in one transaction, so a failure +// never leaves a half-applied version marked as done. +func (d *PostgresDriver) Apply(ctx context.Context, m Migration) error { + tx, err := d.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck // rollback after a successful commit is a no-op + + if _, err := tx.ExecContext(ctx, m.SQL); err != nil { + return fmt.Errorf("exec body: %w", err) + } + if _, err := tx.ExecContext(ctx, + "INSERT INTO schema_migrations (version, name) VALUES ($1, $2)", m.Version, m.Name); err != nil { + return fmt.Errorf("record version: %w", err) + } + return tx.Commit() +}