diff --git a/cmd/spindle/main.go b/cmd/spindle/main.go index 08621b52..b7194b83 100644 --- a/cmd/spindle/main.go +++ b/cmd/spindle/main.go @@ -113,15 +113,52 @@ func millCommand() *cli.Command { Name: "executor", Usage: "manage executors allowed to join this mill", Commands: []*cli.Command{ + { + Name: "token", + Usage: "manage executor credentials", + Commands: []*cli.Command{ + { + Name: "generate", + Usage: "generate and store an executor token", + Flags: []cli.Flag{ + dbFlag, + &cli.DurationFlag{ + Name: "ttl", + Usage: "token lifetime (e.g. 720h); omit for no expiry", + }, + }, + Action: func(ctx context.Context, cmd *cli.Command) error { + d, err := openDB(ctx, cmd) + if err != nil { + return err + } + token, err := mill.GenerateToken() + if err != nil { + return err + } + var expiresAt *time.Time + if ttl := cmd.Duration("ttl"); ttl > 0 { + exp := time.Now().UTC().Add(ttl) + expiresAt = &exp + } + if err := d.CreateExecutorToken(mill.HashToken(token), expiresAt); err != nil { + return fmt.Errorf("storing executor token: %w", err) + } + fmt.Println(token) + return nil + }, + }, + }, + }, { Name: "add", - Usage: "register an executor and print its token", + Usage: "register an executor with an existing token", ArgsUsage: "", Flags: []cli.Flag{ dbFlag, - &cli.DurationFlag{ - Name: "ttl", - Usage: "token lifetime (e.g. 720h); omit for no expiry", + &cli.StringFlag{ + Name: "token-file", + Usage: "file containing a token created by executor token generate", }, &cli.StringSliceFlag{ Name: "label", @@ -131,26 +168,28 @@ func millCommand() *cli.Command { Action: func(ctx context.Context, cmd *cli.Command) error { name := cmd.Args().First() if name == "" { - return fmt.Errorf("usage: spindle mill executor add ") + return fmt.Errorf("usage: spindle mill executor add --token-file ") } - d, err := openDB(ctx, cmd) + tokenFile := cmd.String("token-file") + if tokenFile == "" { + return fmt.Errorf("--token-file is required") + } + tokenBytes, err := os.ReadFile(tokenFile) if err != nil { - return err + return fmt.Errorf("reading token file: %w", err) } - token, err := mill.GenerateToken() + token := strings.TrimSpace(string(tokenBytes)) + if token == "" { + return fmt.Errorf("token file is empty") + } + d, err := openDB(ctx, cmd) if err != nil { return err } - var expiresAt *time.Time - if ttl := cmd.Duration("ttl"); ttl > 0 { - exp := time.Now().UTC().Add(ttl) - expiresAt = &exp - } labels := cmd.StringSlice("label") - if err := d.AddExecutorToken(name, mill.HashToken(token), expiresAt, labels); err != nil { + if err := d.RegisterExecutor(name, mill.HashToken(token), labels); err != nil { return fmt.Errorf("registering executor %q: %w", name, err) } - fmt.Println(token) return nil }, }, @@ -163,32 +202,32 @@ func millCommand() *cli.Command { if err != nil { return err } - tokens, err := d.ListExecutorTokens() + registrations, err := d.ListExecutorRegistrations() if err != nil { return err } w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0) fmt.Fprintln(w, "NAME\tCREATED\tEXPIRES\tLABELS\tQUARANTINE") - for _, t := range tokens { + for _, registration := range registrations { expires := "never" - if t.ExpiresAt != nil { - expires = t.ExpiresAt.Format(time.RFC3339) - if time.Now().After(*t.ExpiresAt) { + if registration.ExpiresAt != nil { + expires = registration.ExpiresAt.Format(time.RFC3339) + if time.Now().After(*registration.ExpiresAt) { expires += " (expired)" } } - labels := strings.Join(t.Labels, ",") + labels := strings.Join(registration.Labels, ",") if labels == "" { labels = "-" } quarantine := "-" - if t.QuarantineReason != nil { - quarantine = *t.QuarantineReason - if t.QuarantinedAt != nil { - quarantine = *t.QuarantinedAt + ": " + quarantine + if registration.QuarantineReason != nil { + quarantine = *registration.QuarantineReason + if registration.QuarantinedAt != nil { + quarantine = *registration.QuarantinedAt + ": " + quarantine } } - fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", t.Name, t.CreatedAt, expires, labels, quarantine) + fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n", registration.Name, registration.CreatedAt, expires, labels, quarantine) } return w.Flush() }, diff --git a/docker-compose.mill.yml b/docker-compose.mill.yml index 6888a5f0..5b506493 100644 --- a/docker-compose.mill.yml +++ b/docker-compose.mill.yml @@ -99,11 +99,16 @@ services: command: - | set -eu + umask 077 seed() { - spindle mill executor revoke "$$1" >/dev/null 2>&1 || true + token_file="/shared/$$1.mill-token" + if [ ! -s "$$token_file" ]; then + spindle mill executor token generate > "$$token_file.tmp" + mv "$$token_file.tmp" "$$token_file" + fi args="" for l in $$(echo "$$2" | tr ',' ' '); do args="$$args --label $$l"; done - spindle mill executor add "$$1" $$args > "/shared/$$1.mill-token" + spindle mill executor add "$$1" --token-file "$$token_file" $$args } seed executor-a linux,fast seed executor-b linux,slow diff --git a/spindle/db/db.go b/spindle/db/db.go index de8de374..5a929473 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -126,14 +126,19 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { foreign key (pipeline_id) references pipelines(id) on delete cascade ); + create table if not exists mill_executor_tokens ( + token_hash text primary key, + created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + expires_at text + ); + create table if not exists mill_executors ( name text primary key, token_hash text not null unique, - created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), - expires_at text, labels text, quarantine_reason text, - quarantined_at text + quarantined_at text, + foreign key (token_hash) references mill_executor_tokens(token_hash) on delete cascade ); create table if not exists mill_leases ( @@ -322,6 +327,44 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + if err := orm.RunMigration(conn, logger, "mill-executor-credentials", func(tx *sql.Tx) error { + var legacySchema int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('mill_executors') where name = 'expires_at'`, + ).Scan(&legacySchema); err != nil { + return err + } + if legacySchema == 0 { + return nil + } + + _, err := tx.Exec(` + insert into mill_executor_tokens (token_hash, created_at, expires_at) + select token_hash, created_at, expires_at from mill_executors; + + create table mill_executors_new ( + name text primary key, + token_hash text not null unique, + labels text, + quarantine_reason text, + quarantined_at text, + foreign key (token_hash) references mill_executor_tokens(token_hash) on delete cascade + ); + + insert into mill_executors_new ( + name, token_hash, labels, quarantine_reason, quarantined_at + ) + select name, token_hash, labels, quarantine_reason, quarantined_at + from mill_executors; + + drop table mill_executors; + alter table mill_executors_new rename to mill_executors; + `) + return err + }); err != nil { + return err + } + return nil } diff --git a/spindle/db/mill_tokens.go b/spindle/db/mill_tokens.go index 69ba351d..e3fd3422 100644 --- a/spindle/db/mill_tokens.go +++ b/spindle/db/mill_tokens.go @@ -3,13 +3,14 @@ package db import ( "database/sql" "encoding/json" + "fmt" "slices" "strings" "time" ) // the raw token is never stored, only its hash -type ExecutorToken struct { +type ExecutorRegistration struct { Name string CreatedAt string ExpiresAt *time.Time @@ -18,26 +19,87 @@ type ExecutorToken struct { QuarantinedAt *string } -func (d *DB) AddExecutorToken(name, tokenHash string, expiresAt *time.Time, labels []string) error { +func (d *DB) CreateExecutorToken(tokenHash string, expiresAt *time.Time) error { + _, err := d.Exec( + `insert into mill_executor_tokens (token_hash, expires_at) values (?, ?)`, + tokenHash, expiryArg(expiresAt), + ) + return err +} + +func (d *DB) RegisterExecutor(name, tokenHash string, labels []string) error { normalized := normalizeLabels(labels) labelsBytes, err := json.Marshal(normalized) if err != nil { return err } - _, err = d.Exec( - `insert into mill_executors (name, token_hash, expires_at, labels) values (?, ?, ?, ?)`, - name, tokenHash, expiryArg(expiresAt), string(labelsBytes), - ) - return err + + tx, err := d.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + var expires sql.NullString + err = tx.QueryRow( + `select expires_at from mill_executor_tokens where token_hash = ?`, + tokenHash, + ).Scan(&expires) + if err == sql.ErrNoRows { + return fmt.Errorf("unknown executor token") + } + if err != nil { + return err + } + exp, hasExpiry, err := parseExpiry(expires) + if err != nil { + return err + } + if hasExpiry && time.Now().After(exp) { + return fmt.Errorf("executor token expired at %s", exp.Format(time.RFC3339)) + } + + var registeredName string + err = tx.QueryRow( + `select name from mill_executors where token_hash = ?`, + tokenHash, + ).Scan(®isteredName) + if err != nil && err != sql.ErrNoRows { + return err + } + if err == nil && registeredName != name { + return fmt.Errorf("executor token is already registered to %q", registeredName) + } + + res, err := tx.Exec(` + insert into mill_executors (name, token_hash, labels) + values (?, ?, ?) + on conflict(name) do update set labels = excluded.labels + where mill_executors.token_hash = excluded.token_hash + `, name, tokenHash, string(labelsBytes)) + if err != nil { + return err + } + changed, err := res.RowsAffected() + if err != nil { + return err + } + if changed == 0 { + return fmt.Errorf("executor %q is already registered with another token", name) + } + return tx.Commit() } func (d *DB) ResolveExecutorToken(tokenHash string) (string, []string, bool, error) { var name string var expires sql.NullString var labelsRaw, quarantineReason sql.NullString - err := d.QueryRow( - `select name, expires_at, labels, quarantine_reason from mill_executors where token_hash = ?`, tokenHash, - ).Scan(&name, &expires, &labelsRaw, &quarantineReason) + err := d.QueryRow(` + select e.name, t.expires_at, e.labels, e.quarantine_reason + from mill_executors e + join mill_executor_tokens t on t.token_hash = e.token_hash + where e.token_hash = ? + `, tokenHash).Scan(&name, &expires, &labelsRaw, &quarantineReason) if err == sql.ErrNoRows { return "", nil, false, nil } @@ -86,7 +148,10 @@ func (d *DB) ClearExecutorQuarantine(name string) (bool, error) { } func (d *DB) RevokeExecutorToken(name string) (bool, error) { - res, err := d.Exec(`delete from mill_executors where name = ?`, name) + res, err := d.Exec(` + delete from mill_executor_tokens + where token_hash = (select token_hash from mill_executors where name = ?) + `, name) if err != nil { return false, err } @@ -94,18 +159,31 @@ func (d *DB) RevokeExecutorToken(name string) (bool, error) { return n > 0, err } -func (d *DB) ListExecutorTokens() ([]ExecutorToken, error) { - rows, err := d.Query(`select name, created_at, expires_at, labels, quarantine_reason, quarantined_at from mill_executors order by name`) +func (d *DB) ListExecutorRegistrations() ([]ExecutorRegistration, error) { + rows, err := d.Query(` + select e.name, t.created_at, t.expires_at, e.labels, + e.quarantine_reason, e.quarantined_at + from mill_executors e + join mill_executor_tokens t on t.token_hash = e.token_hash + order by e.name + `) if err != nil { return nil, err } defer rows.Close() - var out []ExecutorToken + var out []ExecutorRegistration for rows.Next() { - var t ExecutorToken + var registration ExecutorRegistration var expires, labelsRaw, quarantineReason, quarantinedAt sql.NullString - if err := rows.Scan(&t.Name, &t.CreatedAt, &expires, &labelsRaw, &quarantineReason, &quarantinedAt); err != nil { + if err := rows.Scan( + ®istration.Name, + ®istration.CreatedAt, + &expires, + &labelsRaw, + &quarantineReason, + &quarantinedAt, + ); err != nil { return nil, err } exp, hasExpiry, err := parseExpiry(expires) @@ -113,20 +191,20 @@ func (d *DB) ListExecutorTokens() ([]ExecutorToken, error) { return nil, err } if hasExpiry { - t.ExpiresAt = &exp + registration.ExpiresAt = &exp } if labelsRaw.Valid && labelsRaw.String != "" { - if err := json.Unmarshal([]byte(labelsRaw.String), &t.Labels); err != nil { + if err := json.Unmarshal([]byte(labelsRaw.String), ®istration.Labels); err != nil { return nil, err } } if quarantineReason.Valid { - t.QuarantineReason = &quarantineReason.String + registration.QuarantineReason = &quarantineReason.String } if quarantinedAt.Valid { - t.QuarantinedAt = &quarantinedAt.String + registration.QuarantinedAt = &quarantinedAt.String } - out = append(out, t) + out = append(out, registration) } return out, rows.Err() } diff --git a/spindle/db/mill_tokens_test.go b/spindle/db/mill_tokens_test.go index 948f3c82..8cadb83b 100644 --- a/spindle/db/mill_tokens_test.go +++ b/spindle/db/mill_tokens_test.go @@ -1,106 +1,151 @@ package db import ( + "context" + "database/sql" + "path/filepath" "slices" "testing" "time" ) -func TestAddExecutorTokenRejectsDuplicateName(t *testing.T) { +func createAndRegisterExecutor(t *testing.T, d *DB, name, tokenHash string, expiresAt *time.Time, labels []string) { + t.Helper() + if err := d.CreateExecutorToken(tokenHash, expiresAt); err != nil { + t.Fatalf("CreateExecutorToken: %v", err) + } + if err := d.RegisterExecutor(name, tokenHash, labels); err != nil { + t.Fatalf("RegisterExecutor: %v", err) + } +} + +func TestExecutorTokenMustExistBeforeRegistration(t *testing.T) { d := newTestDB(t) - if err := d.AddExecutorToken("exec-1", "hash-a", nil, nil); err != nil { - t.Fatalf("AddExecutorToken: %v", err) + if err := d.RegisterExecutor("exec-1", "hash-1", nil); err == nil { + t.Fatal("RegisterExecutor accepted an unknown token") } - if err := d.AddExecutorToken("exec-1", "hash-b", nil, nil); err == nil { - t.Fatal("AddExecutorToken re-registered an existing name; a duplicate must be rejected") + if err := d.CreateExecutorToken("hash-1", nil); err != nil { + t.Fatalf("CreateExecutorToken: %v", err) } - - name, _, ok, err := d.ResolveExecutorToken("hash-a") - if err != nil { - t.Fatalf("ResolveExecutorToken(hash-a): %v", err) + if _, _, ok, err := d.ResolveExecutorToken("hash-1"); err != nil || ok { + t.Fatalf("ResolveExecutorToken(unregistered) = (ok=%v, err=%v), want (false, nil)", ok, err) } - if !ok || name != "exec-1" { - t.Fatalf("ResolveExecutorToken(hash-a) = (%q, %v), want (exec-1, true)", name, ok) + if err := d.RegisterExecutor("exec-1", "hash-1", nil); err != nil { + t.Fatalf("RegisterExecutor: %v", err) } - if _, _, ok, _ := d.ResolveExecutorToken("hash-b"); ok { - t.Fatal("rejected duplicate's token resolved; the failed insert leaked a credential") + name, _, ok, err := d.ResolveExecutorToken("hash-1") + if err != nil || !ok || name != "exec-1" { + t.Fatalf("ResolveExecutorToken(registered) = (%q, %v, %v), want (exec-1, true, nil)", name, ok, err) } } -func TestResolveExecutorTokenMissAndHit(t *testing.T) { +func TestRegisterExecutorIsIdempotentAndReconcilesLabels(t *testing.T) { d := newTestDB(t) + createAndRegisterExecutor(t, d, "exec-1", "hash-1", nil, []string{"old"}) - name, _, ok, err := d.ResolveExecutorToken("no-such-hash") - if err != nil { - t.Fatalf("ResolveExecutorToken(miss): %v", err) + if err := d.RegisterExecutor("exec-1", "hash-1", []string{" arm64 ", "kvm", "arm64"}); err != nil { + t.Fatalf("RegisterExecutor(repeat): %v", err) + } + name, labels, ok, err := d.ResolveExecutorToken("hash-1") + if err != nil || !ok || name != "exec-1" { + t.Fatalf("ResolveExecutorToken = (%q, %v, %v), want (exec-1, true, nil)", name, ok, err) } - if ok || name != "" { - t.Fatalf("ResolveExecutorToken(miss) = (%q, %v), want (\"\", false)", name, ok) + if want := []string{"arm64", "kvm"}; !slices.Equal(labels, want) { + t.Fatalf("labels = %v, want %v", labels, want) } +} - if err := d.AddExecutorToken("exec-1", "hash-1", nil, nil); err != nil { - t.Fatalf("AddExecutorToken: %v", err) +func TestRegisterExecutorRejectsCredentialConflicts(t *testing.T) { + d := newTestDB(t) + if err := d.CreateExecutorToken("hash-a", nil); err != nil { + t.Fatalf("CreateExecutorToken(hash-a): %v", err) + } + if err := d.CreateExecutorToken("hash-b", nil); err != nil { + t.Fatalf("CreateExecutorToken(hash-b): %v", err) + } + if err := d.RegisterExecutor("exec-1", "hash-a", nil); err != nil { + t.Fatalf("RegisterExecutor: %v", err) } - name, _, ok, err = d.ResolveExecutorToken("hash-1") - if err != nil { - t.Fatalf("ResolveExecutorToken(hit): %v", err) + if err := d.RegisterExecutor("exec-1", "hash-b", nil); err == nil { + t.Fatal("RegisterExecutor replaced an existing executor's token") } - if !ok || name != "exec-1" { - t.Fatalf("ResolveExecutorToken(hash-1) = (%q, %v), want (exec-1, true)", name, ok) + if err := d.RegisterExecutor("exec-2", "hash-a", nil); err == nil { + t.Fatal("RegisterExecutor assigned one token to two executor names") } + if name, _, ok, err := d.ResolveExecutorToken("hash-a"); err != nil || !ok || name != "exec-1" { + t.Fatalf("ResolveExecutorToken(hash-a) = (%q, %v, %v), want (exec-1, true, nil)", name, ok, err) + } + if _, _, ok, err := d.ResolveExecutorToken("hash-b"); err != nil || ok { + t.Fatalf("ResolveExecutorToken(hash-b) = (ok=%v, err=%v), want (false, nil)", ok, err) + } +} - if _, _, ok, _ := d.ResolveExecutorToken("hash-unregistered"); ok { - t.Fatal("ResolveExecutorToken matched an unregistered hash") +func TestRegisterExecutorRejectsExpiredToken(t *testing.T) { + d := newTestDB(t) + expired := time.Now().Add(-time.Hour) + if err := d.CreateExecutorToken("hash-1", &expired); err != nil { + t.Fatalf("CreateExecutorToken: %v", err) + } + if err := d.RegisterExecutor("exec-1", "hash-1", nil); err == nil { + t.Fatal("RegisterExecutor accepted an expired token") } } -func TestRevokeExecutorToken(t *testing.T) { +func TestResolveExecutorTokenMissAndHit(t *testing.T) { d := newTestDB(t) - if err := d.AddExecutorToken("exec-1", "hash-1", nil, nil); err != nil { - t.Fatalf("AddExecutorToken: %v", err) + name, _, ok, err := d.ResolveExecutorToken("no-such-hash") + if err != nil || ok || name != "" { + t.Fatalf("ResolveExecutorToken(miss) = (%q, %v, %v), want (\"\", false, nil)", name, ok, err) + } + + createAndRegisterExecutor(t, d, "exec-1", "hash-1", nil, nil) + name, _, ok, err = d.ResolveExecutorToken("hash-1") + if err != nil || !ok || name != "exec-1" { + t.Fatalf("ResolveExecutorToken(hit) = (%q, %v, %v), want (exec-1, true, nil)", name, ok, err) } +} + +func TestRevokeExecutorToken(t *testing.T) { + d := newTestDB(t) + createAndRegisterExecutor(t, d, "exec-1", "hash-1", nil, nil) deleted, err := d.RevokeExecutorToken("exec-1") - if err != nil { - t.Fatalf("RevokeExecutorToken: %v", err) + if err != nil || !deleted { + t.Fatalf("RevokeExecutorToken = (%v, %v), want (true, nil)", deleted, err) } - if !deleted { - t.Fatal("RevokeExecutorToken reported no deletion for an existing identity") + if _, _, ok, err := d.ResolveExecutorToken("hash-1"); err != nil || ok { + t.Fatalf("ResolveExecutorToken(revoked) = (ok=%v, err=%v), want (false, nil)", ok, err) } - - if _, _, ok, _ := d.ResolveExecutorToken("hash-1"); ok { - t.Fatal("revoked token still resolves; revocation is not enforced") + var tokens int + if err := d.QueryRow(`select count(*) from mill_executor_tokens where token_hash = ?`, "hash-1").Scan(&tokens); err != nil { + t.Fatalf("count revoked token: %v", err) } - - if deleted, err := d.RevokeExecutorToken("exec-1"); err != nil || deleted { - t.Fatalf("RevokeExecutorToken(already-gone) = (%v, %v), want (false, nil)", deleted, err) + if tokens != 0 { + t.Fatal("RevokeExecutorToken left the credential behind") } - - if deleted, err := d.RevokeExecutorToken("ghost"); err != nil || deleted { - t.Fatalf("RevokeExecutorToken(unknown) = (%v, %v), want (false, nil)", deleted, err) + if deleted, err := d.RevokeExecutorToken("exec-1"); err != nil || deleted { + t.Fatalf("RevokeExecutorToken(already gone) = (%v, %v), want (false, nil)", deleted, err) } } func TestExecutorQuarantineIsVisibleAndReversible(t *testing.T) { d := newTestDB(t) - if err := d.AddExecutorToken("exec-1", "hash-1", nil, nil); err != nil { - t.Fatalf("AddExecutorToken: %v", err) - } + createAndRegisterExecutor(t, d, "exec-1", "hash-1", nil, nil) if err := d.QuarantineExecutor("exec-1", "missed cancel deadline"); err != nil { t.Fatalf("QuarantineExecutor: %v", err) } if _, _, ok, err := d.ResolveExecutorToken("hash-1"); err != nil || ok { t.Fatalf("ResolveExecutorToken(quarantined) = (ok=%v, err=%v), want (false, nil)", ok, err) } - tokens, err := d.ListExecutorTokens() + registrations, err := d.ListExecutorRegistrations() if err != nil { - t.Fatalf("ListExecutorTokens: %v", err) + t.Fatalf("ListExecutorRegistrations: %v", err) } - if len(tokens) != 1 || tokens[0].QuarantineReason == nil || *tokens[0].QuarantineReason != "missed cancel deadline" || tokens[0].QuarantinedAt == nil { - t.Fatalf("quarantined token not surfaced: %+v", tokens) + if len(registrations) != 1 || registrations[0].QuarantineReason == nil || *registrations[0].QuarantineReason != "missed cancel deadline" || registrations[0].QuarantinedAt == nil { + t.Fatalf("quarantined executor not surfaced: %+v", registrations) } if cleared, err := d.ClearExecutorQuarantine("exec-1"); err != nil || !cleared { t.Fatalf("ClearExecutorQuarantine = (%v, %v), want (true, nil)", cleared, err) @@ -108,51 +153,39 @@ func TestExecutorQuarantineIsVisibleAndReversible(t *testing.T) { if _, _, ok, err := d.ResolveExecutorToken("hash-1"); err != nil || !ok { t.Fatalf("ResolveExecutorToken(cleared) = (ok=%v, err=%v), want (true, nil)", ok, err) } - if cleared, err := d.ClearExecutorQuarantine("missing"); err != nil || cleared { - t.Fatalf("ClearExecutorQuarantine(missing) = (%v, %v), want (false, nil)", cleared, err) - } } -func TestListExecutorTokens(t *testing.T) { +func TestListExecutorRegistrations(t *testing.T) { d := newTestDB(t) - tokens, err := d.ListExecutorTokens() - if err != nil { - t.Fatalf("ListExecutorTokens(empty): %v", err) + registrations, err := d.ListExecutorRegistrations() + if err != nil || len(registrations) != 0 { + t.Fatalf("ListExecutorRegistrations(empty) = (%d rows, %v), want (0, nil)", len(registrations), err) } - if len(tokens) != 0 { - t.Fatalf("ListExecutorTokens on empty table = %d rows, want 0", len(tokens)) - } - - // out of alphabetical order, the query has to sort them for _, name := range []string{"charlie", "alice", "bob"} { - if err := d.AddExecutorToken(name, "hash-"+name, nil, nil); err != nil { - t.Fatalf("AddExecutorToken(%s): %v", name, err) - } + createAndRegisterExecutor(t, d, name, "hash-"+name, nil, nil) } - - tokens, err = d.ListExecutorTokens() + registrations, err = d.ListExecutorRegistrations() if err != nil { - t.Fatalf("ListExecutorTokens: %v", err) + t.Fatalf("ListExecutorRegistrations: %v", err) } want := []string{"alice", "bob", "charlie"} - if len(tokens) != len(want) { - t.Fatalf("ListExecutorTokens = %d rows, want %d", len(tokens), len(want)) + if len(registrations) != len(want) { + t.Fatalf("ListExecutorRegistrations = %d rows, want %d", len(registrations), len(want)) } for i := range want { - if tokens[i].Name != want[i] { - t.Fatalf("ListExecutorTokens[%d].Name = %q, want %q (ordered by name)", i, tokens[i].Name, want[i]) + if registrations[i].Name != want[i] { + t.Fatalf("ListExecutorRegistrations[%d].Name = %q, want %q", i, registrations[i].Name, want[i]) } } } -// expired tokens must fail closed func TestResolveExecutorTokenExpiry(t *testing.T) { cases := []struct { - name string - seqno time.Duration - noExpiry bool - wantOK bool + name string + offset time.Duration + noExpiry bool + wantValid bool }{ {"future expiry resolves", time.Hour, false, true}, {"past expiry fails closed", -time.Hour, false, false}, @@ -161,28 +194,24 @@ func TestResolveExecutorTokenExpiry(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { d := newTestDB(t) - var expires *time.Time if !tc.noExpiry { - exp := time.Now().Add(tc.seqno) + exp := time.Now().Add(tc.offset) expires = &exp } - if err := d.AddExecutorToken("exec-1", "hash-1", expires, nil); err != nil { - t.Fatalf("AddExecutorToken: %v", err) + if err := d.CreateExecutorToken("hash-1", expires); err != nil { + t.Fatalf("CreateExecutorToken: %v", err) } - - name, _, ok, err := d.ResolveExecutorToken("hash-1") - if err != nil { - t.Fatalf("ResolveExecutorToken: %v", err) + if tc.wantValid { + if err := d.RegisterExecutor("exec-1", "hash-1", nil); err != nil { + t.Fatalf("RegisterExecutor: %v", err) + } + } else if _, err := d.Exec(`insert into mill_executors (name, token_hash) values (?, ?)`, "exec-1", "hash-1"); err != nil { + t.Fatalf("insert expired registration: %v", err) } - if ok != tc.wantOK { - t.Fatalf("ResolveExecutorToken ok = %v, want %v", ok, tc.wantOK) - } - if tc.wantOK && name != "exec-1" { - t.Fatalf("ResolveExecutorToken name = %q, want exec-1", name) - } - if !tc.wantOK && name != "" { - t.Fatalf("ResolveExecutorToken name = %q, want \"\" when failing closed", name) + name, _, ok, err := d.ResolveExecutorToken("hash-1") + if err != nil || ok != tc.wantValid { + t.Fatalf("ResolveExecutorToken = (%q, %v, %v), want valid=%v", name, ok, err, tc.wantValid) } }) } @@ -190,83 +219,97 @@ func TestResolveExecutorTokenExpiry(t *testing.T) { func TestResolveExecutorTokenRejectsMalformedExpiry(t *testing.T) { d := newTestDB(t) - if _, err := d.Exec( - `insert into mill_executors (name, token_hash, expires_at) values (?, ?, ?)`, - "exec-1", "hash-1", "not-a-timestamp", - ); err != nil { + if _, err := d.Exec(`insert into mill_executor_tokens (token_hash, expires_at) values (?, ?)`, "hash-1", "not-a-timestamp"); err != nil { t.Fatalf("insert malformed token: %v", err) } - - name, _, ok, err := d.ResolveExecutorToken("hash-1") - if err == nil { - t.Fatal("ResolveExecutorToken accepted a malformed non-NULL expiry") + if _, err := d.Exec(`insert into mill_executors (name, token_hash) values (?, ?)`, "exec-1", "hash-1"); err != nil { + t.Fatalf("insert registration: %v", err) } - if ok || name != "" { + name, _, ok, err := d.ResolveExecutorToken("hash-1") + if err == nil || ok || name != "" { t.Fatalf("ResolveExecutorToken = (%q, %v, %v), want (\"\", false, error)", name, ok, err) } } -// expiry storage is RFC3339 second precision UTC so -// round trips compare to the second -func TestListExecutorTokensSurfacesExpiry(t *testing.T) { +func TestListExecutorRegistrationsSurfacesExpiry(t *testing.T) { d := newTestDB(t) - exp := time.Now().Add(24 * time.Hour) - if err := d.AddExecutorToken("expiring", "hash-exp", &exp, nil); err != nil { - t.Fatalf("AddExecutorToken(expiring): %v", err) - } - if err := d.AddExecutorToken("forever", "hash-forever", nil, nil); err != nil { - t.Fatalf("AddExecutorToken(forever): %v", err) - } + createAndRegisterExecutor(t, d, "expiring", "hash-exp", &exp, nil) + createAndRegisterExecutor(t, d, "forever", "hash-forever", nil, nil) - tokens, err := d.ListExecutorTokens() + registrations, err := d.ListExecutorRegistrations() if err != nil { - t.Fatalf("ListExecutorTokens: %v", err) + t.Fatalf("ListExecutorRegistrations: %v", err) } - - got := make(map[string]*time.Time, len(tokens)) - for _, tok := range tokens { - got[tok.Name] = tok.ExpiresAt + got := make(map[string]*time.Time, len(registrations)) + for _, registration := range registrations { + got[registration.Name] = registration.ExpiresAt } - - e, present := got["forever"] - if !present { - t.Fatal("ListExecutorTokens omitted the non-expiring identity") + if got["forever"] != nil { + t.Fatalf("forever.ExpiresAt = %v, want nil", got["forever"]) } - if e != nil { - t.Fatalf("forever.ExpiresAt = %v, want nil (never expires)", e) + if got["expiring"] == nil || got["expiring"].Unix() != exp.Unix() { + t.Fatalf("expiring.ExpiresAt = %v, want unix %d", got["expiring"], exp.Unix()) } +} - e, present = got["expiring"] - if !present { - t.Fatal("ListExecutorTokens omitted the expiring identity") - } - if e == nil { - t.Fatal("expiring.ExpiresAt = nil, want the stored expiry") +func TestLegacyExecutorTokensMigrateWithoutChangingIdentityState(t *testing.T) { + path := filepath.Join(t.TempDir(), "spindle.db") + legacy, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatalf("open legacy db: %v", err) + } + expires := time.Now().UTC().Add(24 * time.Hour).Truncate(time.Second).Format(time.RFC3339) + if _, err := legacy.Exec(` + create table mill_executors ( + name text primary key, + token_hash text not null unique, + created_at text not null, + expires_at text, + labels text, + quarantine_reason text, + quarantined_at text + ); + create table mill_executor_cursors ( + node_id text not null, + epoch text not null, + acked_seqno integer not null, + primary key (node_id, epoch) + ); + insert into mill_executors ( + name, token_hash, created_at, expires_at, labels, quarantine_reason, quarantined_at + ) values ('exec-1', 'hash-1', '2026-01-02T03:04:05Z', ?, '["arm64"]', null, null); + insert into mill_executor_cursors (node_id, epoch, acked_seqno) + values ('exec-1', 'epoch-1', 42); + `, expires); err != nil { + legacy.Close() + t.Fatalf("seed legacy db: %v", err) + } + if err := legacy.Close(); err != nil { + t.Fatalf("close legacy db: %v", err) + } + + d, err := Make(context.Background(), path) + if err != nil { + t.Fatalf("Make migrated db: %v", err) } - if e.Unix() != exp.Unix() { - t.Fatalf("expiring.ExpiresAt = %d (unix), want %d", e.Unix(), exp.Unix()) + defer d.Close() + name, labels, ok, err := d.ResolveExecutorToken("hash-1") + if err != nil || !ok || name != "exec-1" || !slices.Equal(labels, []string{"arm64"}) { + t.Fatalf("ResolveExecutorToken(migrated) = (%q, %v, %v, %v)", name, labels, ok, err) } -} - -func TestExecutorTokenLabels(t *testing.T) { - d := newTestDB(t) - - labels := []string{" foo ", "bar", " foo", ""} - if err := d.AddExecutorToken("exec-1", "hash-1", nil, labels); err != nil { - t.Fatalf("AddExecutorToken: %v", err) + registrations, err := d.ListExecutorRegistrations() + if err != nil || len(registrations) != 1 { + t.Fatalf("ListExecutorRegistrations(migrated) = (%+v, %v)", registrations, err) } - - name, resolvedLabels, ok, err := d.ResolveExecutorToken("hash-1") - if err != nil { - t.Fatalf("ResolveExecutorToken: %v", err) + if registrations[0].CreatedAt != "2026-01-02T03:04:05Z" || registrations[0].ExpiresAt == nil || registrations[0].ExpiresAt.Format(time.RFC3339) != expires { + t.Fatalf("migrated credential metadata = %+v, want created and expiry preserved", registrations[0]) } - if !ok || name != "exec-1" { - t.Fatalf("ResolveExecutorToken: ok=%v, name=%q, want true, exec-1", ok, name) + var acked uint64 + if err := d.QueryRow(`select acked_seqno from mill_executor_cursors where node_id = 'exec-1' and epoch = 'epoch-1'`).Scan(&acked); err != nil { + t.Fatalf("read migrated cursor: %v", err) } - - wantLabels := []string{"bar", "foo"} - if !slices.Equal(resolvedLabels, wantLabels) { - t.Fatalf("resolved labels = %v, want %v", resolvedLabels, wantLabels) + if acked != 42 { + t.Fatalf("migrated cursor = %d, want 42", acked) } } diff --git a/spindle/mill/auth_test.go b/spindle/mill/auth_test.go index 6876f791..c78f851c 100644 --- a/spindle/mill/auth_test.go +++ b/spindle/mill/auth_test.go @@ -28,6 +28,16 @@ func nopEncoder() scriptedEncoder { return scriptedEncoder(func(*millproto.Message) error { return nil }) } +func registerTestExecutor(t *testing.T, d *db.DB, name, tokenHash string, labels []string) { + t.Helper() + if err := d.CreateExecutorToken(tokenHash, nil); err != nil { + t.Fatalf("CreateExecutorToken: %v", err) + } + if err := d.RegisterExecutor(name, tokenHash, labels); err != nil { + t.Fatalf("RegisterExecutor: %v", err) + } +} + func TestHashToken(t *testing.T) { const raw = "super-secret-executor-token" @@ -318,9 +328,7 @@ func setupTestServer(t *testing.T, authorizedLabels []string) (*Mill, *db.DB, *h m.Attach(bdb, &n) const secret = "test-secret" - if err := bdb.AddExecutorToken("dev-node", HashToken(secret), nil, authorizedLabels); err != nil { - t.Fatalf("AddExecutorToken: %v", err) - } + registerTestExecutor(t, bdb, "dev-node", HashToken(secret), authorizedLabels) server := httptest.NewServer(http.HandlerFunc(m.HandleExecutorConn)) t.Cleanup(server.Close) diff --git a/spindle/mill/integration_test.go b/spindle/mill/integration_test.go index 409e0f31..0aad2795 100644 --- a/spindle/mill/integration_test.go +++ b/spindle/mill/integration_test.go @@ -35,9 +35,7 @@ func TestEndToEndDummyJob(t *testing.T) { bn := notifier.New() mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second}) mill.Attach(bdb, &bn) - if err := bdb.AddExecutorToken("exec-1", HashToken("test-token"), nil, nil); err != nil { - t.Fatalf("register executor token: %v", err) - } + registerTestExecutor(t, bdb, "exec-1", HashToken("test-token"), nil) srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn)) defer srv.Close() @@ -144,9 +142,7 @@ func TestExecutorConfiguredLabelsAreStoredOnSession(t *testing.T) { bn := notifier.New() mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second}) mill.Attach(bdb, &bn) - if err := bdb.AddExecutorToken("exec-labels", HashToken("test-token"), nil, []string{"linux", "arm64", "gpu"}); err != nil { - t.Fatalf("register executor token: %v", err) - } + registerTestExecutor(t, bdb, "exec-labels", HashToken("test-token"), []string{"linux", "arm64", "gpu"}) srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn)) defer srv.Close() @@ -193,12 +189,8 @@ func TestEndToEndDummyJobUsesRequiredLabelsAcrossExecutors(t *testing.T) { bn := notifier.New() mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second}) mill.Attach(bdb, &bn) - if err := bdb.AddExecutorToken("exec-x86", HashToken("token-x86"), nil, []string{"linux/amd64", "kvm"}); err != nil { - t.Fatalf("register x86 executor token: %v", err) - } - if err := bdb.AddExecutorToken("exec-arm", HashToken("token-arm"), nil, []string{"linux/arm64", "kvm"}); err != nil { - t.Fatalf("register arm executor token: %v", err) - } + registerTestExecutor(t, bdb, "exec-x86", HashToken("token-x86"), []string{"linux/amd64", "kvm"}) + registerTestExecutor(t, bdb, "exec-arm", HashToken("token-arm"), []string{"linux/arm64", "kvm"}) srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn)) defer srv.Close() diff --git a/spindle/mill/mill_test.go b/spindle/mill/mill_test.go index 1cf88d02..8a29a88b 100644 --- a/spindle/mill/mill_test.go +++ b/spindle/mill/mill_test.go @@ -875,9 +875,7 @@ func TestBoundedBidding(t *testing.T) { func TestProtocolStrikesQuarantineExecutor(t *testing.T) { m, bdb := restoreTestMill(t, Config{ReconnectGrace: time.Minute, QuarantineStrikes: 2}) - if err := bdb.AddExecutorToken("node-1", HashToken("tok-1"), nil, nil); err != nil { - t.Fatalf("AddExecutorToken: %v", err) - } + registerTestExecutor(t, bdb, "node-1", HashToken("tok-1"), nil) sess := newSession("node-1", "inc-1", nil, nopEncoder(), discardLogger()) resolve := func() bool { @@ -904,9 +902,9 @@ func TestProtocolStrikesQuarantineExecutor(t *testing.T) { if resolve() { t.Fatal("expected quarantine after consecutive protocol deaths") } - tokens, err := bdb.ListExecutorTokens() - if err != nil || len(tokens) != 1 || tokens[0].QuarantineReason == nil { - t.Fatalf("quarantine not visible in list: %+v, %v", tokens, err) + registrations, err := bdb.ListExecutorRegistrations() + if err != nil || len(registrations) != 1 || registrations[0].QuarantineReason == nil { + t.Fatalf("quarantine not visible in list: %+v, %v", registrations, err) } }