Monorepo for Tangled
spindle/mill: authenticate executors with per-executor tokens
Registers executors in the db and resolves identity from the token hash
in the Authorization header rather than the client-supplied Hello. Adds
`spindle mill executor add/list/revoke` and seeds the dev SharedSecret as
a bootstrap identity.
+834
-120
+120
cmd/spindle/main.go
+120
cmd/spindle/main.go
···
2
2
3
3
import (
4
4
"context"
5
+
"fmt"
5
6
"log/slog"
6
7
"os"
8
+
"text/tabwriter"
9
+
"time"
7
10
8
11
"github.com/urfave/cli/v3"
9
12
tlog "tangled.org/core/log"
10
13
"tangled.org/core/spindle"
14
+
"tangled.org/core/spindle/db"
15
+
"tangled.org/core/spindle/mill"
11
16
)
12
17
13
18
func main() {
···
16
21
Usage: "spindle continuous integration runner",
17
22
Commands: []*cli.Command{
18
23
Command(),
24
+
millCommand(),
19
25
},
20
26
DefaultCommand: "run",
21
27
}
···
41
47
},
42
48
}
43
49
}
50
+
51
+
// millCommand groups mill-host administration. Executors allowed to join the
52
+
// mill are managed under `mill executor`.
53
+
func millCommand() *cli.Command {
54
+
dbFlag := &cli.StringFlag{
55
+
Name: "db",
56
+
Usage: "path to the spindle sqlite db",
57
+
Value: "spindle.db",
58
+
Sources: cli.EnvVars("SPINDLE_SERVER_DB_PATH"),
59
+
}
60
+
openDB := func(ctx context.Context, cmd *cli.Command) (*db.DB, error) {
61
+
return db.Make(ctx, cmd.String("db"))
62
+
}
63
+
return &cli.Command{
64
+
Name: "mill",
65
+
Usage: "mill host administration",
66
+
Commands: []*cli.Command{
67
+
{
68
+
Name: "executor",
69
+
Usage: "manage executors allowed to join this mill",
70
+
Commands: []*cli.Command{
71
+
{
72
+
Name: "add",
73
+
Usage: "register an executor and print its token",
74
+
ArgsUsage: "<name>",
75
+
Flags: []cli.Flag{
76
+
dbFlag,
77
+
&cli.DurationFlag{
78
+
Name: "ttl",
79
+
Usage: "token lifetime (e.g. 720h); omit for no expiry",
80
+
},
81
+
},
82
+
Action: func(ctx context.Context, cmd *cli.Command) error {
83
+
name := cmd.Args().First()
84
+
if name == "" {
85
+
return fmt.Errorf("usage: spindle mill executor add <name>")
86
+
}
87
+
d, err := openDB(ctx, cmd)
88
+
if err != nil {
89
+
return err
90
+
}
91
+
token, err := mill.GenerateToken()
92
+
if err != nil {
93
+
return err
94
+
}
95
+
var expiresAt *time.Time
96
+
if ttl := cmd.Duration("ttl"); ttl > 0 {
97
+
exp := time.Now().UTC().Add(ttl)
98
+
expiresAt = &exp
99
+
}
100
+
if err := d.AddExecutorToken(name, mill.HashToken(token), expiresAt); err != nil {
101
+
return fmt.Errorf("registering executor %q: %w", name, err)
102
+
}
103
+
fmt.Println(token)
104
+
return nil
105
+
},
106
+
},
107
+
{
108
+
Name: "list",
109
+
Usage: "list registered executors",
110
+
Flags: []cli.Flag{dbFlag},
111
+
Action: func(ctx context.Context, cmd *cli.Command) error {
112
+
d, err := openDB(ctx, cmd)
113
+
if err != nil {
114
+
return err
115
+
}
116
+
tokens, err := d.ListExecutorTokens()
117
+
if err != nil {
118
+
return err
119
+
}
120
+
w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
121
+
fmt.Fprintln(w, "NAME\tCREATED\tEXPIRES")
122
+
for _, t := range tokens {
123
+
expires := "never"
124
+
if t.ExpiresAt != nil {
125
+
expires = t.ExpiresAt.Format(time.RFC3339)
126
+
if time.Now().After(*t.ExpiresAt) {
127
+
expires += " (expired)"
128
+
}
129
+
}
130
+
fmt.Fprintf(w, "%s\t%s\t%s\n", t.Name, t.CreatedAt, expires)
131
+
}
132
+
return w.Flush()
133
+
},
134
+
},
135
+
{
136
+
Name: "revoke",
137
+
Usage: "revoke an executor's token",
138
+
ArgsUsage: "<name>",
139
+
Flags: []cli.Flag{dbFlag},
140
+
Action: func(ctx context.Context, cmd *cli.Command) error {
141
+
name := cmd.Args().First()
142
+
if name == "" {
143
+
return fmt.Errorf("usage: spindle mill executor revoke <name>")
144
+
}
145
+
d, err := openDB(ctx, cmd)
146
+
if err != nil {
147
+
return err
148
+
}
149
+
ok, err := d.RevokeExecutorToken(name)
150
+
if err != nil {
151
+
return err
152
+
}
153
+
if !ok {
154
+
return fmt.Errorf("no such executor identity %q", name)
155
+
}
156
+
return nil
157
+
},
158
+
},
159
+
},
160
+
},
161
+
},
162
+
}
163
+
}
+51
-52
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
+51
-52
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
···
2
2
// This file is @generated by prost-build.
3
3
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
4
4
pub struct Hello {
5
-
#[prost(uint32, tag="1")]
5
+
#[prost(uint32, tag = "1")]
6
6
pub protocol_version: u32,
7
-
#[prost(string, tag="2")]
7
+
#[prost(string, tag = "2")]
8
8
pub agent_version: ::prost::alloc::string::String,
9
-
#[prost(string, tag="3")]
9
+
#[prost(string, tag = "3")]
10
10
pub boot_id: ::prost::alloc::string::String,
11
-
#[prost(string, tag="4")]
11
+
#[prost(string, tag = "4")]
12
12
pub nix_version: ::prost::alloc::string::String,
13
13
}
14
14
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
15
15
pub struct Init {
16
-
#[prost(string, tag="1")]
16
+
#[prost(string, tag = "1")]
17
17
pub job_id: ::prost::alloc::string::String,
18
-
#[prost(string, repeated, tag="2")]
18
+
#[prost(string, repeated, tag = "2")]
19
19
pub cache_trusted_public_keys: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
20
-
#[prost(uint32, tag="3")]
20
+
#[prost(uint32, tag = "3")]
21
21
pub cache_read_proxy_port: u32,
22
-
#[prost(uint32, tag="4")]
22
+
#[prost(uint32, tag = "4")]
23
23
pub cache_upload_proxy_port: u32,
24
-
#[prost(uint32, tag="5")]
24
+
#[prost(uint32, tag = "5")]
25
25
pub dns_proxy_port: u32,
26
26
}
27
27
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
28
28
pub struct ExecStart {
29
-
#[prost(string, repeated, tag="1")]
29
+
#[prost(string, repeated, tag = "1")]
30
30
pub argv: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
31
-
#[prost(string, repeated, tag="2")]
31
+
#[prost(string, repeated, tag = "2")]
32
32
pub env: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
33
-
#[prost(string, tag="3")]
33
+
#[prost(string, tag = "3")]
34
34
pub cwd: ::prost::alloc::string::String,
35
-
#[prost(string, tag="4")]
35
+
#[prost(string, tag = "4")]
36
36
pub user: ::prost::alloc::string::String,
37
-
#[prost(uint32, tag="5")]
37
+
#[prost(uint32, tag = "5")]
38
38
pub timeout_seconds: u32,
39
39
}
40
40
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
41
41
pub struct ExecStdout {
42
-
#[prost(string, tag="1")]
42
+
#[prost(string, tag = "1")]
43
43
pub data: ::prost::alloc::string::String,
44
44
}
45
45
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
46
46
pub struct ExecStderr {
47
-
#[prost(string, tag="1")]
47
+
#[prost(string, tag = "1")]
48
48
pub data: ::prost::alloc::string::String,
49
49
}
50
50
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
51
51
pub struct ExecExit {
52
-
#[prost(int32, tag="1")]
52
+
#[prost(int32, tag = "1")]
53
53
pub exit_code: i32,
54
-
#[prost(string, tag="2")]
54
+
#[prost(string, tag = "2")]
55
55
pub error: ::prost::alloc::string::String,
56
56
/// set when the guest killed the step on its own timeout timer, so the host
57
57
/// can classify it as a timeout rather than inferring failure from exit_code.
58
-
#[prost(bool, tag="3")]
58
+
#[prost(bool, tag = "3")]
59
59
pub timed_out: bool,
60
60
}
61
61
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
62
62
pub struct ActivateConfig {
63
-
#[prost(string, tag="1")]
63
+
#[prost(string, tag = "1")]
64
64
pub config_key: ::prost::alloc::string::String,
65
-
#[prost(string, tag="2")]
65
+
#[prost(string, tag = "2")]
66
66
pub base_config_hash: ::prost::alloc::string::String,
67
-
#[prost(string, tag="3")]
67
+
#[prost(string, tag = "3")]
68
68
pub user_config: ::prost::alloc::string::String,
69
-
#[prost(string, tag="4")]
69
+
#[prost(string, tag = "4")]
70
70
pub toplevel: ::prost::alloc::string::String,
71
-
#[prost(uint32, tag="5")]
71
+
#[prost(uint32, tag = "5")]
72
72
pub timeout_seconds: u32,
73
73
}
74
74
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
75
75
pub struct ActivateConfigResult {
76
-
#[prost(string, tag="1")]
76
+
#[prost(string, tag = "1")]
77
77
pub config_key: ::prost::alloc::string::String,
78
-
#[prost(string, tag="2")]
78
+
#[prost(string, tag = "2")]
79
79
pub toplevel: ::prost::alloc::string::String,
80
-
#[prost(string, tag="3")]
80
+
#[prost(string, tag = "3")]
81
81
pub error: ::prost::alloc::string::String,
82
82
}
83
83
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
84
84
pub struct BuiltPaths {
85
-
#[prost(string, repeated, tag="1")]
85
+
#[prost(string, repeated, tag = "1")]
86
86
pub paths: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
87
-
#[prost(string, tag="2")]
87
+
#[prost(string, tag = "2")]
88
88
pub reason: ::prost::alloc::string::String,
89
89
}
90
90
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
91
91
pub struct CacheDrain {
92
-
#[prost(uint32, tag="1")]
92
+
#[prost(uint32, tag = "1")]
93
93
pub timeout_seconds: u32,
94
94
}
95
95
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
96
96
pub struct CacheDrainResult {
97
-
#[prost(string, tag="1")]
97
+
#[prost(string, tag = "1")]
98
98
pub error: ::prost::alloc::string::String,
99
-
#[prost(uint32, tag="2")]
99
+
#[prost(uint32, tag = "2")]
100
100
pub cache_queued: u32,
101
-
#[prost(uint32, tag="3")]
101
+
#[prost(uint32, tag = "3")]
102
102
pub cache_active: u32,
103
-
#[prost(uint32, tag="4")]
103
+
#[prost(uint32, tag = "4")]
104
104
pub cache_uploaded: u32,
105
-
#[prost(uint32, tag="5")]
105
+
#[prost(uint32, tag = "5")]
106
106
pub cache_failed: u32,
107
107
}
108
108
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
109
-
pub struct Poweroff {
110
-
}
109
+
pub struct Poweroff {}
111
110
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
112
111
pub struct PoweroffResult {
113
-
#[prost(string, tag="1")]
112
+
#[prost(string, tag = "1")]
114
113
pub error: ::prost::alloc::string::String,
115
114
}
116
115
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
117
116
pub struct Message {
118
-
#[prost(string, tag="1")]
117
+
#[prost(string, tag = "1")]
119
118
pub id: ::prost::alloc::string::String,
120
-
#[prost(message, optional, tag="2")]
119
+
#[prost(message, optional, tag = "2")]
121
120
pub hello: ::core::option::Option<Hello>,
122
-
#[prost(message, optional, tag="3")]
121
+
#[prost(message, optional, tag = "3")]
123
122
pub init: ::core::option::Option<Init>,
124
-
#[prost(message, optional, tag="4")]
123
+
#[prost(message, optional, tag = "4")]
125
124
pub exec_start: ::core::option::Option<ExecStart>,
126
-
#[prost(message, optional, tag="5")]
125
+
#[prost(message, optional, tag = "5")]
127
126
pub exec_stdout: ::core::option::Option<ExecStdout>,
128
-
#[prost(message, optional, tag="6")]
127
+
#[prost(message, optional, tag = "6")]
129
128
pub exec_stderr: ::core::option::Option<ExecStderr>,
130
-
#[prost(message, optional, tag="7")]
129
+
#[prost(message, optional, tag = "7")]
131
130
pub exec_exit: ::core::option::Option<ExecExit>,
132
-
#[prost(message, optional, tag="8")]
131
+
#[prost(message, optional, tag = "8")]
133
132
pub activate_config: ::core::option::Option<ActivateConfig>,
134
-
#[prost(message, optional, tag="9")]
133
+
#[prost(message, optional, tag = "9")]
135
134
pub activate_config_result: ::core::option::Option<ActivateConfigResult>,
136
-
#[prost(message, optional, tag="10")]
135
+
#[prost(message, optional, tag = "10")]
137
136
pub built_paths: ::core::option::Option<BuiltPaths>,
138
-
#[prost(message, optional, tag="11")]
137
+
#[prost(message, optional, tag = "11")]
139
138
pub cache_drain: ::core::option::Option<CacheDrain>,
140
-
#[prost(message, optional, tag="12")]
139
+
#[prost(message, optional, tag = "12")]
141
140
pub cache_drain_result: ::core::option::Option<CacheDrainResult>,
142
-
#[prost(message, optional, tag="13")]
141
+
#[prost(message, optional, tag = "13")]
143
142
pub poweroff: ::core::option::Option<Poweroff>,
144
-
#[prost(message, optional, tag="14")]
143
+
#[prost(message, optional, tag = "14")]
145
144
pub poweroff_result: ::core::option::Option<PoweroffResult>,
146
145
}
147
146
// @@protoc_insertion_point(module)
+3
spindle/config/config.go
+3
spindle/config/config.go
···
134
134
if c.Mill.URL == "" {
135
135
return fmt.Errorf("SPINDLE_ROLE=executor requires SPINDLE_MILL_URL (the mill to dial)")
136
136
}
137
+
if c.Mill.SharedSecret == "" {
138
+
return fmt.Errorf("SPINDLE_ROLE=executor requires SPINDLE_MILL_SHARED_SECRET (its executor token)")
139
+
}
137
140
default:
138
141
return fmt.Errorf("unknown SPINDLE_ROLE %q (want standalone, mill, or executor)", c.Role)
139
142
}
+7
spindle/db/db.go
+7
spindle/db/db.go
···
123
123
foreign key (pipeline_id) references pipelines(id) on delete cascade
124
124
);
125
125
126
+
create table if not exists mill_executors (
127
+
name text primary key,
128
+
token_hash text not null,
129
+
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
130
+
expires_at text
131
+
);
132
+
126
133
create table if not exists migrations (
127
134
id integer primary key autoincrement,
128
135
name text unique
+111
spindle/db/mill_tokens.go
+111
spindle/db/mill_tokens.go
···
1
+
package db
2
+
3
+
import (
4
+
"database/sql"
5
+
"time"
6
+
)
7
+
8
+
// ExecutorToken is a registered mill executor identity. The raw token is never
9
+
// stored; token_hash is a hash of it (see mill.HashToken). ExpiresAt is nil for
10
+
// a token that never expires.
11
+
type ExecutorToken struct {
12
+
Name string
13
+
CreatedAt string
14
+
ExpiresAt *time.Time
15
+
}
16
+
17
+
// AddExecutorToken registers a new executor identity. expiresAt is optional (nil
18
+
// means the token never expires). It fails if the name is already taken.
19
+
func (d *DB) AddExecutorToken(name, tokenHash string, expiresAt *time.Time) error {
20
+
_, err := d.Exec(
21
+
`insert into mill_executors (name, token_hash, expires_at) values (?, ?, ?)`,
22
+
name, tokenHash, expiryArg(expiresAt),
23
+
)
24
+
return err
25
+
}
26
+
27
+
// UpsertExecutorToken registers or replaces an executor identity's token with no
28
+
// expiry. Used for the dev bootstrap seed; production registers per-executor
29
+
// tokens via AddExecutorToken.
30
+
func (d *DB) UpsertExecutorToken(name, tokenHash string) error {
31
+
_, err := d.Exec(
32
+
`insert into mill_executors (name, token_hash, expires_at) values (?, ?, null)
33
+
on conflict(name) do update set token_hash = excluded.token_hash, expires_at = null`,
34
+
name, tokenHash,
35
+
)
36
+
return err
37
+
}
38
+
39
+
// ResolveExecutorToken returns the executor name a token hash is registered
40
+
// under. ok is false when no identity matches or the token has expired.
41
+
func (d *DB) ResolveExecutorToken(tokenHash string) (string, bool, error) {
42
+
var name string
43
+
var expires sql.NullString
44
+
err := d.QueryRow(
45
+
`select name, expires_at from mill_executors where token_hash = ?`, tokenHash,
46
+
).Scan(&name, &expires)
47
+
if err == sql.ErrNoRows {
48
+
return "", false, nil
49
+
}
50
+
if err != nil {
51
+
return "", false, err
52
+
}
53
+
if exp, ok := parseExpiry(expires); ok && time.Now().After(exp) {
54
+
return "", false, nil
55
+
}
56
+
return name, true, nil
57
+
}
58
+
59
+
// RevokeExecutorToken removes an executor identity, returning whether a row was
60
+
// deleted.
61
+
func (d *DB) RevokeExecutorToken(name string) (bool, error) {
62
+
res, err := d.Exec(`delete from mill_executors where name = ?`, name)
63
+
if err != nil {
64
+
return false, err
65
+
}
66
+
n, err := res.RowsAffected()
67
+
return n > 0, err
68
+
}
69
+
70
+
// ListExecutorTokens returns the registered executor identities ordered by name.
71
+
func (d *DB) ListExecutorTokens() ([]ExecutorToken, error) {
72
+
rows, err := d.Query(`select name, created_at, expires_at from mill_executors order by name`)
73
+
if err != nil {
74
+
return nil, err
75
+
}
76
+
defer rows.Close()
77
+
78
+
var out []ExecutorToken
79
+
for rows.Next() {
80
+
var t ExecutorToken
81
+
var expires sql.NullString
82
+
if err := rows.Scan(&t.Name, &t.CreatedAt, &expires); err != nil {
83
+
return nil, err
84
+
}
85
+
if exp, ok := parseExpiry(expires); ok {
86
+
t.ExpiresAt = &exp
87
+
}
88
+
out = append(out, t)
89
+
}
90
+
return out, rows.Err()
91
+
}
92
+
93
+
// expiryArg renders an optional expiry as a sql argument (nil -> NULL).
94
+
func expiryArg(t *time.Time) any {
95
+
if t == nil {
96
+
return nil
97
+
}
98
+
return t.UTC().Format(time.RFC3339)
99
+
}
100
+
101
+
// parseExpiry decodes a stored expiry column; ok is false for NULL or unparseable.
102
+
func parseExpiry(s sql.NullString) (time.Time, bool) {
103
+
if !s.Valid || s.String == "" {
104
+
return time.Time{}, false
105
+
}
106
+
t, err := time.Parse(time.RFC3339, s.String)
107
+
if err != nil {
108
+
return time.Time{}, false
109
+
}
110
+
return t, true
111
+
}
+263
spindle/db/mill_tokens_test.go
+263
spindle/db/mill_tokens_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"testing"
5
+
"time"
6
+
)
7
+
8
+
// AddExecutorToken must fail closed on a name collision: a second registration
9
+
// under a live identity's name cannot silently take it over, and the failed
10
+
// insert must not leave the newcomer's token resolvable.
11
+
func TestAddExecutorTokenRejectsDuplicateName(t *testing.T) {
12
+
d := newTestDB(t)
13
+
14
+
if err := d.AddExecutorToken("exec-1", "hash-a", nil); err != nil {
15
+
t.Fatalf("AddExecutorToken: %v", err)
16
+
}
17
+
if err := d.AddExecutorToken("exec-1", "hash-b", nil); err == nil {
18
+
t.Fatal("AddExecutorToken re-registered an existing name; a duplicate must be rejected")
19
+
}
20
+
21
+
// the original identity is untouched...
22
+
name, ok, err := d.ResolveExecutorToken("hash-a")
23
+
if err != nil {
24
+
t.Fatalf("ResolveExecutorToken(hash-a): %v", err)
25
+
}
26
+
if !ok || name != "exec-1" {
27
+
t.Fatalf("ResolveExecutorToken(hash-a) = (%q, %v), want (exec-1, true)", name, ok)
28
+
}
29
+
// ...and the rejected duplicate's token never became a credential.
30
+
if _, ok, _ := d.ResolveExecutorToken("hash-b"); ok {
31
+
t.Fatal("rejected duplicate's token resolved; the failed insert leaked a credential")
32
+
}
33
+
}
34
+
35
+
// ResolveExecutorToken is the auth lookup: an unknown hash must fail closed
36
+
// (ok=false), a known hash must resolve to the exact identity it was
37
+
// registered under.
38
+
func TestResolveExecutorTokenMissAndHit(t *testing.T) {
39
+
d := newTestDB(t)
40
+
41
+
// miss on an empty table: unknown token does not authenticate.
42
+
name, ok, err := d.ResolveExecutorToken("no-such-hash")
43
+
if err != nil {
44
+
t.Fatalf("ResolveExecutorToken(miss): %v", err)
45
+
}
46
+
if ok || name != "" {
47
+
t.Fatalf("ResolveExecutorToken(miss) = (%q, %v), want (\"\", false)", name, ok)
48
+
}
49
+
50
+
if err := d.AddExecutorToken("exec-1", "hash-1", nil); err != nil {
51
+
t.Fatalf("AddExecutorToken: %v", err)
52
+
}
53
+
54
+
// hit: resolves to the registered identity.
55
+
name, ok, err = d.ResolveExecutorToken("hash-1")
56
+
if err != nil {
57
+
t.Fatalf("ResolveExecutorToken(hit): %v", err)
58
+
}
59
+
if !ok || name != "exec-1" {
60
+
t.Fatalf("ResolveExecutorToken(hash-1) = (%q, %v), want (exec-1, true)", name, ok)
61
+
}
62
+
63
+
// a hash no identity was registered under still misses.
64
+
if _, ok, _ := d.ResolveExecutorToken("hash-unregistered"); ok {
65
+
t.Fatal("ResolveExecutorToken matched an unregistered hash")
66
+
}
67
+
}
68
+
69
+
// UpsertExecutorToken rotates an identity's credential: after an upsert the new
70
+
// hash authenticates as the same name and the rotated-out hash no longer does.
71
+
// An upsert of a fresh name inserts it.
72
+
func TestUpsertExecutorTokenRotatesHash(t *testing.T) {
73
+
d := newTestDB(t)
74
+
75
+
if err := d.AddExecutorToken("exec-1", "old-hash", nil); err != nil {
76
+
t.Fatalf("AddExecutorToken: %v", err)
77
+
}
78
+
if err := d.UpsertExecutorToken("exec-1", "new-hash"); err != nil {
79
+
t.Fatalf("UpsertExecutorToken(rotate): %v", err)
80
+
}
81
+
82
+
name, ok, err := d.ResolveExecutorToken("new-hash")
83
+
if err != nil {
84
+
t.Fatalf("ResolveExecutorToken(new-hash): %v", err)
85
+
}
86
+
if !ok || name != "exec-1" {
87
+
t.Fatalf("ResolveExecutorToken(new-hash) = (%q, %v), want (exec-1, true)", name, ok)
88
+
}
89
+
if _, ok, _ := d.ResolveExecutorToken("old-hash"); ok {
90
+
t.Fatal("rotated-out token still resolves; UpsertExecutorToken did not replace the hash")
91
+
}
92
+
93
+
// upsert of a name that does not exist yet inserts a new identity.
94
+
if err := d.UpsertExecutorToken("exec-2", "hash-2"); err != nil {
95
+
t.Fatalf("UpsertExecutorToken(insert): %v", err)
96
+
}
97
+
name, ok, err = d.ResolveExecutorToken("hash-2")
98
+
if err != nil {
99
+
t.Fatalf("ResolveExecutorToken(hash-2): %v", err)
100
+
}
101
+
if !ok || name != "exec-2" {
102
+
t.Fatalf("ResolveExecutorToken(hash-2) = (%q, %v), want (exec-2, true)", name, ok)
103
+
}
104
+
}
105
+
106
+
// RevokeExecutorToken removes an identity so its token stops authenticating,
107
+
// and reports deleted=false when the name is not present.
108
+
func TestRevokeExecutorToken(t *testing.T) {
109
+
d := newTestDB(t)
110
+
111
+
if err := d.AddExecutorToken("exec-1", "hash-1", nil); err != nil {
112
+
t.Fatalf("AddExecutorToken: %v", err)
113
+
}
114
+
115
+
deleted, err := d.RevokeExecutorToken("exec-1")
116
+
if err != nil {
117
+
t.Fatalf("RevokeExecutorToken: %v", err)
118
+
}
119
+
if !deleted {
120
+
t.Fatal("RevokeExecutorToken reported no deletion for an existing identity")
121
+
}
122
+
123
+
// the revoked token must no longer authenticate.
124
+
if _, ok, _ := d.ResolveExecutorToken("hash-1"); ok {
125
+
t.Fatal("revoked token still resolves; revocation is not enforced")
126
+
}
127
+
128
+
// revoking the now-absent name reports nothing deleted.
129
+
if deleted, err := d.RevokeExecutorToken("exec-1"); err != nil || deleted {
130
+
t.Fatalf("RevokeExecutorToken(already-gone) = (%v, %v), want (false, nil)", deleted, err)
131
+
}
132
+
133
+
// revoking a name that never existed reports nothing deleted.
134
+
if deleted, err := d.RevokeExecutorToken("ghost"); err != nil || deleted {
135
+
t.Fatalf("RevokeExecutorToken(unknown) = (%v, %v), want (false, nil)", deleted, err)
136
+
}
137
+
}
138
+
139
+
// ListExecutorTokens returns every registered identity ordered by name.
140
+
func TestListExecutorTokens(t *testing.T) {
141
+
d := newTestDB(t)
142
+
143
+
tokens, err := d.ListExecutorTokens()
144
+
if err != nil {
145
+
t.Fatalf("ListExecutorTokens(empty): %v", err)
146
+
}
147
+
if len(tokens) != 0 {
148
+
t.Fatalf("ListExecutorTokens on empty table = %d rows, want 0", len(tokens))
149
+
}
150
+
151
+
// insert out of alphabetical order to prove the ordering is the query's.
152
+
for _, name := range []string{"charlie", "alice", "bob"} {
153
+
if err := d.AddExecutorToken(name, "hash-"+name, nil); err != nil {
154
+
t.Fatalf("AddExecutorToken(%s): %v", name, err)
155
+
}
156
+
}
157
+
158
+
tokens, err = d.ListExecutorTokens()
159
+
if err != nil {
160
+
t.Fatalf("ListExecutorTokens: %v", err)
161
+
}
162
+
want := []string{"alice", "bob", "charlie"}
163
+
if len(tokens) != len(want) {
164
+
t.Fatalf("ListExecutorTokens = %d rows, want %d", len(tokens), len(want))
165
+
}
166
+
for i := range want {
167
+
if tokens[i].Name != want[i] {
168
+
t.Fatalf("ListExecutorTokens[%d].Name = %q, want %q (ordered by name)", i, tokens[i].Name, want[i])
169
+
}
170
+
}
171
+
}
172
+
173
+
// ResolveExecutorToken enforces token expiry fail-closed: a token whose
174
+
// expires_at lies in the past must NOT authenticate (ok=false, no error),
175
+
// while a future or absent expiry authenticates as normal. An expired
176
+
// credential that still resolves is an auth defect, so this pins that branch.
177
+
func TestResolveExecutorTokenExpiry(t *testing.T) {
178
+
cases := []struct {
179
+
name string
180
+
offset time.Duration
181
+
noExpiry bool
182
+
wantOK bool
183
+
}{
184
+
{"future expiry resolves", time.Hour, false, true},
185
+
{"past expiry fails closed", -time.Hour, false, false},
186
+
{"nil expiry never expires", 0, true, true},
187
+
}
188
+
for _, tc := range cases {
189
+
t.Run(tc.name, func(t *testing.T) {
190
+
d := newTestDB(t)
191
+
192
+
var expires *time.Time
193
+
if !tc.noExpiry {
194
+
exp := time.Now().Add(tc.offset)
195
+
expires = &exp
196
+
}
197
+
if err := d.AddExecutorToken("exec-1", "hash-1", expires); err != nil {
198
+
t.Fatalf("AddExecutorToken: %v", err)
199
+
}
200
+
201
+
name, ok, err := d.ResolveExecutorToken("hash-1")
202
+
if err != nil {
203
+
t.Fatalf("ResolveExecutorToken: %v", err)
204
+
}
205
+
if ok != tc.wantOK {
206
+
t.Fatalf("ResolveExecutorToken ok = %v, want %v", ok, tc.wantOK)
207
+
}
208
+
if tc.wantOK && name != "exec-1" {
209
+
t.Fatalf("ResolveExecutorToken name = %q, want exec-1", name)
210
+
}
211
+
if !tc.wantOK && name != "" {
212
+
t.Fatalf("ResolveExecutorToken name = %q, want \"\" when failing closed", name)
213
+
}
214
+
})
215
+
}
216
+
}
217
+
218
+
// ListExecutorTokens surfaces each identity's expiry: nil for a non-expiring
219
+
// token and the stored instant for an expiring one. Storage is RFC3339
220
+
// second-precision UTC, so the round-tripped instant must equal the original
221
+
// to the second.
222
+
func TestListExecutorTokensSurfacesExpiry(t *testing.T) {
223
+
d := newTestDB(t)
224
+
225
+
exp := time.Now().Add(24 * time.Hour)
226
+
if err := d.AddExecutorToken("expiring", "hash-exp", &exp); err != nil {
227
+
t.Fatalf("AddExecutorToken(expiring): %v", err)
228
+
}
229
+
if err := d.AddExecutorToken("forever", "hash-forever", nil); err != nil {
230
+
t.Fatalf("AddExecutorToken(forever): %v", err)
231
+
}
232
+
233
+
tokens, err := d.ListExecutorTokens()
234
+
if err != nil {
235
+
t.Fatalf("ListExecutorTokens: %v", err)
236
+
}
237
+
238
+
got := make(map[string]*time.Time, len(tokens))
239
+
for _, tok := range tokens {
240
+
got[tok.Name] = tok.ExpiresAt
241
+
}
242
+
243
+
// the non-expiring token reports a nil expiry.
244
+
e, present := got["forever"]
245
+
if !present {
246
+
t.Fatal("ListExecutorTokens omitted the non-expiring identity")
247
+
}
248
+
if e != nil {
249
+
t.Fatalf("forever.ExpiresAt = %v, want nil (never expires)", e)
250
+
}
251
+
252
+
// the expiring token reports its stored instant, to the second.
253
+
e, present = got["expiring"]
254
+
if !present {
255
+
t.Fatal("ListExecutorTokens omitted the expiring identity")
256
+
}
257
+
if e == nil {
258
+
t.Fatal("expiring.ExpiresAt = nil, want the stored expiry")
259
+
}
260
+
if e.Unix() != exp.Unix() {
261
+
t.Fatalf("expiring.ExpiresAt = %d (unix), want %d", e.Unix(), exp.Unix())
262
+
}
263
+
}
+39
spindle/mill/auth_test.go
+39
spindle/mill/auth_test.go
···
25
25
return scriptedEncoder(func(*millproto.Message) error { return nil })
26
26
}
27
27
28
+
// HashToken is the storage/lookup transform for executor credentials. It must be
29
+
// deterministic (so a token minted once keeps authenticating), collision-free
30
+
// across distinct tokens, and it must never echo the raw token (so a DB leak of
31
+
// the hash never yields a usable credential).
32
+
func TestHashToken(t *testing.T) {
33
+
const raw = "super-secret-executor-token"
34
+
35
+
if HashToken(raw) != HashToken(raw) {
36
+
t.Fatal("HashToken is not deterministic; the same token would stop authenticating")
37
+
}
38
+
if HashToken("token-a") == HashToken("token-b") {
39
+
t.Fatal("HashToken collided two distinct tokens")
40
+
}
41
+
if HashToken(raw) == raw {
42
+
t.Fatal("HashToken returned the raw token; a hash leak would expose a usable credential")
43
+
}
44
+
}
45
+
46
+
// GenerateToken must hand out distinct, non-empty tokens on every call; a
47
+
// generator that repeats or emits empties would let one executor's credential
48
+
// collide with another's.
49
+
func TestGenerateTokenDistinct(t *testing.T) {
50
+
const n = 100
51
+
seen := make(map[string]struct{}, n)
52
+
for i := range n {
53
+
tok, err := GenerateToken()
54
+
if err != nil {
55
+
t.Fatalf("GenerateToken: %v", err)
56
+
}
57
+
if tok == "" {
58
+
t.Fatalf("GenerateToken returned an empty token on call %d", i)
59
+
}
60
+
if _, dup := seen[tok]; dup {
61
+
t.Fatalf("GenerateToken repeated a token after %d calls: %q", i, tok)
62
+
}
63
+
seen[tok] = struct{}{}
64
+
}
65
+
}
66
+
28
67
// attachSession is the anti-hijack gate: a valid token must not be able to
29
68
// displace a live executor that already holds that node identity. Only once the
30
69
// incumbent is disconnected (in its reconnect grace window) may a reconnect
-11
spindle/mill/executor/executor.go
-11
spindle/mill/executor/executor.go
···
138
138
139
139
hello := &millproto.Message{Hello: &millv1.Hello{
140
140
ProtocolVersion: millproto.ProtocolVersion,
141
-
NodeId: e.nodeID,
142
-
Engines: e.engineNames(),
143
141
Arch: runtime.GOARCH,
144
142
Labels: e.labels,
145
-
LastOffset: e.relay.lastOffset(),
146
143
}}
147
144
if err := enc.Encode(hello); err != nil {
148
145
return fmt.Errorf("send hello: %w", err)
···
500
497
e.draining = true
501
498
e.mu.Unlock()
502
499
e.pushSnapshot()
503
-
}
504
-
505
-
func (e *Executor) engineNames() []string {
506
-
names := make([]string, 0, len(e.engines))
507
-
for name := range e.engines {
508
-
names = append(names, name)
509
-
}
510
-
return names
511
500
}
512
501
513
502
func ttlDuration(secs uint32) time.Duration {
+2
-2
spindle/mill/executor/reserved.go
+2
-2
spindle/mill/executor/reserved.go
···
17
17
// This is the only change to the execution path on an executor.
18
18
type reservedEngine struct {
19
19
models.Engine
20
-
slot engine.WorkflowSlot
21
-
once sync.Once
20
+
slot engine.WorkflowSlot
21
+
once sync.Once
22
22
}
23
23
24
24
// newReservedEngine returns a wrapper around inner that hands back slot exactly
+33
-8
spindle/mill/handler.go
+33
-8
spindle/mill/handler.go
···
2
2
3
3
import (
4
4
"net/http"
5
+
"strings"
5
6
6
7
"github.com/gorilla/websocket"
7
8
···
18
19
// secret in the Authorization header, checked before the upgrade so a bad token
19
20
// never opens a socket.
20
21
func (m *Mill) HandleExecutorConn(w http.ResponseWriter, r *http.Request) {
21
-
if m.cfg.SharedSecret != "" {
22
-
if r.Header.Get("Authorization") != "Bearer "+m.cfg.SharedSecret {
23
-
http.Error(w, "unauthorized", http.StatusUnauthorized)
24
-
return
25
-
}
22
+
name, ok := m.authenticate(r)
23
+
if !ok {
24
+
http.Error(w, "unauthorized", http.StatusUnauthorized)
25
+
return
26
26
}
27
27
28
28
conn, err := upgrader.Upgrade(w, r, nil)
···
52
52
return
53
53
}
54
54
55
-
sess := newSession(h.GetNodeId(), enc, m.l)
55
+
// identity comes from the authenticated token, never the client-supplied
56
+
// Hello node id, so a valid token can't impersonate another executor.
57
+
sess := newSession(name, enc, m.l)
58
+
sess.labels = h.GetLabels()
56
59
resume, ok := m.attachSession(sess)
57
60
if !ok {
58
-
m.l.Warn("rejecting duplicate live executor session", "node", sess.nodeID)
61
+
m.l.Warn("rejecting duplicate live executor session", "node", name)
59
62
return
60
63
}
61
-
m.l.Info("executor connected", "node", sess.nodeID, "engines", h.GetEngines(), "arch", h.GetArch(), "resume", resume)
64
+
m.l.Info("executor connected", "node", sess.nodeID, "arch", h.GetArch(), "labels", h.GetLabels(), "resume", resume)
62
65
63
66
if err := sess.send(&millproto.Message{Resume: &millv1.Resume{AckOffset: resume}}); err != nil {
64
67
m.l.Error("fleet send resume failed", "err", err)
···
70
73
sess.readLoop(m, dec)
71
74
m.detachSession(sess)
72
75
}
76
+
77
+
// authenticate resolves the executor identity from the pre-shared token in the
78
+
// Authorization header before the upgrade, so a bad token never opens a socket.
79
+
// The token is looked up by hash; an unknown or missing token is rejected
80
+
// (fail closed).
81
+
func (m *Mill) authenticate(r *http.Request) (string, bool) {
82
+
const prefix = "Bearer "
83
+
h := r.Header.Get("Authorization")
84
+
if !strings.HasPrefix(h, prefix) {
85
+
return "", false
86
+
}
87
+
token := strings.TrimPrefix(h, prefix)
88
+
if token == "" || m.db == nil {
89
+
return "", false
90
+
}
91
+
name, ok, err := m.db.ResolveExecutorToken(HashToken(token))
92
+
if err != nil {
93
+
m.l.Error("executor token lookup failed", "err", err)
94
+
return "", false
95
+
}
96
+
return name, ok
97
+
}
+155
spindle/mill/integration_test.go
+155
spindle/mill/integration_test.go
···
39
39
bn := notifier.New()
40
40
mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second})
41
41
mill.Attach(bdb, &bn)
42
+
if err := bdb.AddExecutorToken("exec-1", HashToken("test-token"), nil); err != nil {
43
+
t.Fatalf("register executor token: %v", err)
44
+
}
42
45
43
46
srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn))
44
47
defer srv.Close()
···
56
59
cfg.Server.Hostname = "exec-1"
57
60
cfg.Mill.URL = wsURL
58
61
cfg.Mill.Seats = 2
62
+
cfg.Mill.SharedSecret = "test-token"
59
63
60
64
engines := map[string]models.Engine{"dummy": dummy.New(l)}
61
65
exec := executor.New(cfg, engines, edb, &en, l)
···
90
94
if !waitForStatus(t, bdb, wid, "running") {
91
95
t.Fatal("mill never saw relayed running status")
92
96
}
97
+
}
98
+
99
+
func TestExecutorConfiguredLabelsAreStoredOnSession(t *testing.T) {
100
+
ctx, cancel := context.WithCancel(context.Background())
101
+
defer cancel()
102
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
103
+
104
+
millDir := t.TempDir()
105
+
bdb, err := db.Make(ctx, filepath.Join(millDir, "mill.db"))
106
+
if err != nil {
107
+
t.Fatalf("mill db: %v", err)
108
+
}
109
+
bn := notifier.New()
110
+
mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second})
111
+
mill.Attach(bdb, &bn)
112
+
if err := bdb.AddExecutorToken("exec-labels", HashToken("test-token"), nil); err != nil {
113
+
t.Fatalf("register executor token: %v", err)
114
+
}
115
+
116
+
srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn))
117
+
defer srv.Close()
118
+
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
119
+
120
+
execDir := t.TempDir()
121
+
edb, err := db.Make(ctx, filepath.Join(execDir, "exec.db"))
122
+
if err != nil {
123
+
t.Fatalf("exec db: %v", err)
124
+
}
125
+
en := notifier.New()
126
+
cfg := &config.Config{}
127
+
cfg.Server.LogDir = execDir
128
+
cfg.Server.Hostname = "exec-labels"
129
+
cfg.Mill.URL = wsURL
130
+
cfg.Mill.Seats = 2
131
+
cfg.Mill.SharedSecret = "test-token"
132
+
cfg.Mill.Labels = []string{"linux", "arm64", "gpu"}
133
+
134
+
engines := map[string]models.Engine{"dummy": dummy.New(l)}
135
+
exec := executor.New(cfg, engines, edb, &en, l)
136
+
go exec.Connect(ctx)
137
+
138
+
if !waitForSessionLabels(t, mill, "exec-labels", []string{"linux", "arm64", "gpu"}) {
139
+
t.Fatal("mill session never stored executor labels from hello")
140
+
}
141
+
}
142
+
143
+
func TestEndToEndDummyJobUsesRequiredLabelsAcrossExecutors(t *testing.T) {
144
+
ctx, cancel := context.WithCancel(context.Background())
145
+
defer cancel()
146
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
147
+
148
+
millDir := t.TempDir()
149
+
bdb, err := db.Make(ctx, filepath.Join(millDir, "mill.db"))
150
+
if err != nil {
151
+
t.Fatalf("mill db: %v", err)
152
+
}
153
+
bn := notifier.New()
154
+
mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second})
155
+
mill.Attach(bdb, &bn)
156
+
if err := bdb.AddExecutorToken("exec-x86", HashToken("token-x86"), nil); err != nil {
157
+
t.Fatalf("register x86 executor token: %v", err)
158
+
}
159
+
if err := bdb.AddExecutorToken("exec-arm", HashToken("token-arm"), nil); err != nil {
160
+
t.Fatalf("register arm executor token: %v", err)
161
+
}
162
+
163
+
srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn))
164
+
defer srv.Close()
165
+
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
166
+
167
+
startExecutor := func(name, token string, labels []string) {
168
+
t.Helper()
169
+
execDir := t.TempDir()
170
+
edb, err := db.Make(ctx, filepath.Join(execDir, "exec.db"))
171
+
if err != nil {
172
+
t.Fatalf("%s exec db: %v", name, err)
173
+
}
174
+
en := notifier.New()
175
+
cfg := &config.Config{}
176
+
cfg.Server.LogDir = execDir
177
+
cfg.Server.Hostname = name
178
+
cfg.Mill.URL = wsURL
179
+
cfg.Mill.Seats = 1
180
+
cfg.Mill.SharedSecret = token
181
+
cfg.Mill.Labels = labels
182
+
183
+
engines := map[string]models.Engine{"dummy": dummy.New(l)}
184
+
exec := executor.New(cfg, engines, edb, &en, l)
185
+
go exec.Connect(ctx)
186
+
}
187
+
startExecutor("exec-x86", "token-x86", []string{"linux/amd64", "kvm"})
188
+
startExecutor("exec-arm", "token-arm", []string{"linux/arm64", "kvm"})
189
+
190
+
if !waitForSessionLabels(t, mill, "exec-x86", []string{"linux/amd64", "kvm"}) {
191
+
t.Fatal("x86 executor did not connect with labels")
192
+
}
193
+
if !waitForSessionLabels(t, mill, "exec-arm", []string{"linux/arm64", "kvm"}) {
194
+
t.Fatal("arm executor did not connect with labels")
195
+
}
196
+
197
+
be := NewEngine("dummy", mill)
198
+
twf := tangled.Pipeline_Workflow{
199
+
Name: "build-arm",
200
+
RunsOn: []string{"linux/arm64"},
201
+
Raw: "steps:\n - name: hello\n command: echo hi\n",
202
+
}
203
+
wf, err := be.InitWorkflow(twf, tangled.Pipeline{})
204
+
if err != nil {
205
+
t.Fatalf("InitWorkflow: %v", err)
206
+
}
207
+
wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "knot.test", Rkey: "rkey-arm"}, Name: "build-arm"}
208
+
209
+
placeCtx, placeCancel := context.WithTimeout(ctx, 10*time.Second)
210
+
defer placeCancel()
211
+
212
+
slot, err := mill.place(placeCtx, "dummy", wid, wf)
213
+
if err != nil {
214
+
t.Fatalf("place: %v", err)
215
+
}
216
+
defer slot.Release()
217
+
218
+
lease := wf.Data.(*millWorkflowState).Lease
219
+
if lease == nil {
220
+
t.Fatal("place did not attach lease to workflow state")
221
+
}
222
+
if lease.nodeID != "exec-arm" {
223
+
t.Fatalf("placed on %q, want exec-arm", lease.nodeID)
224
+
}
225
+
226
+
if err := mill.commitAndWait(placeCtx, wf, nil); err != nil {
227
+
t.Fatalf("commitAndWait: %v, want success", err)
228
+
}
229
+
}
230
+
231
+
func waitForSessionLabels(t *testing.T, m *Mill, nodeID string, want []string) bool {
232
+
t.Helper()
233
+
deadline := time.Now().Add(3 * time.Second)
234
+
for time.Now().Before(deadline) {
235
+
m.mu.Lock()
236
+
sess := m.sessions[nodeID]
237
+
var got []string
238
+
if sess != nil {
239
+
got = append([]string(nil), sess.labels...)
240
+
}
241
+
m.mu.Unlock()
242
+
if sameStringMultiset(got, want) {
243
+
return true
244
+
}
245
+
time.Sleep(10 * time.Millisecond)
246
+
}
247
+
return false
93
248
}
94
249
95
250
func waitForStatus(t *testing.T, d *db.DB, wid models.WorkflowId, want string) bool {
+9
spindle/mill/mill.go
+9
spindle/mill/mill.go
···
90
90
m.mu.Unlock()
91
91
}
92
92
93
+
// dev bootstrap: shared-secret identity, single-token fleet. production
94
+
// should register per-executor tokens instead.
95
+
func (m *Mill) SeedBootstrapToken() error {
96
+
if m.cfg.SharedSecret == "" || m.db == nil {
97
+
return nil
98
+
}
99
+
return m.db.UpsertExecutorToken(bootstrapTokenName, HashToken(m.cfg.SharedSecret))
100
+
}
101
+
93
102
func (m *Mill) nextLeaseID() string {
94
103
m.mu.Lock()
95
104
m.leaseSeq++
+6
-38
spindle/mill/proto/gen/mill.pb.go
+6
-38
spindle/mill/proto/gen/mill.pb.go
···
76
76
type Hello struct {
77
77
state protoimpl.MessageState `protogen:"open.v1"`
78
78
ProtocolVersion uint32 `protobuf:"varint,1,opt,name=protocol_version,json=protocolVersion,proto3" json:"protocol_version,omitempty"`
79
-
// stable across reconnects (e.g. hostname / DID); the mill keys sessions on
80
-
// this so a brief blip reattaches the same node rather than creating a new one.
81
-
NodeId string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
82
-
// engine names this node can run ("microvm", "nixery").
83
-
Engines []string `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty"`
84
79
// GOARCH of the node, retained as an informational trait for logs.
85
-
Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"`
86
-
// the highest relay offset the executor believes it has sent; a resume hint.
87
-
LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
80
+
Arch string `protobuf:"bytes,2,opt,name=arch,proto3" json:"arch,omitempty"`
88
81
// opaque operator-defined labels used for runs_on matching.
89
-
Labels []string `protobuf:"bytes,6,rep,name=labels,proto3" json:"labels,omitempty"`
82
+
Labels []string `protobuf:"bytes,3,rep,name=labels,proto3" json:"labels,omitempty"`
90
83
unknownFields protoimpl.UnknownFields
91
84
sizeCache protoimpl.SizeCache
92
85
}
···
128
121
return 0
129
122
}
130
123
131
-
func (x *Hello) GetNodeId() string {
132
-
if x != nil {
133
-
return x.NodeId
134
-
}
135
-
return ""
136
-
}
137
-
138
-
func (x *Hello) GetEngines() []string {
139
-
if x != nil {
140
-
return x.Engines
141
-
}
142
-
return nil
143
-
}
144
-
145
124
func (x *Hello) GetArch() string {
146
125
if x != nil {
147
126
return x.Arch
148
127
}
149
128
return ""
150
-
}
151
-
152
-
func (x *Hello) GetLastOffset() uint64 {
153
-
if x != nil {
154
-
return x.LastOffset
155
-
}
156
-
return 0
157
129
}
158
130
159
131
func (x *Hello) GetLabels() []string {
···
1175
1147
1176
1148
const file_spindle_mill_v1_mill_proto_rawDesc = "" +
1177
1149
"\n" +
1178
-
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xbb\x01\n" +
1150
+
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"^\n" +
1179
1151
"\x05Hello\x12)\n" +
1180
-
"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12 \n" +
1181
-
"\anode_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06nodeId\x12\x18\n" +
1182
-
"\aengines\x18\x03 \x03(\tR\aengines\x12\x12\n" +
1183
-
"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1f\n" +
1184
-
"\vlast_offset\x18\x05 \x01(\x04R\n" +
1185
-
"lastOffset\x12\x16\n" +
1186
-
"\x06labels\x18\x06 \x03(\tR\x06labels\"'\n" +
1152
+
"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12\x12\n" +
1153
+
"\x04arch\x18\x02 \x01(\tR\x04arch\x12\x16\n" +
1154
+
"\x06labels\x18\x03 \x03(\tR\x06labels\"'\n" +
1187
1155
"\x06Resume\x12\x1d\n" +
1188
1156
"\n" +
1189
1157
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xae\x01\n" +
+2
-9
spindle/mill/proto/spindle/mill/v1/mill.proto
+2
-9
spindle/mill/proto/spindle/mill/v1/mill.proto
···
10
10
// carries the static traits of the node plus a resume hint.
11
11
message Hello {
12
12
uint32 protocol_version = 1;
13
-
// stable across reconnects (e.g. hostname / DID); the mill keys sessions on
14
-
// this so a brief blip reattaches the same node rather than creating a new one.
15
-
string node_id = 2 [(buf.validate.field).string.min_len = 1];
16
-
// engine names this node can run ("microvm", "nixery").
17
-
repeated string engines = 3;
18
13
// GOARCH of the node, retained as an informational trait for logs.
19
-
string arch = 4;
20
-
// the highest relay offset the executor believes it has sent; a resume hint.
21
-
uint64 last_offset = 5;
14
+
string arch = 2;
22
15
// opaque operator-defined labels used for runs_on matching.
23
-
repeated string labels = 6;
16
+
repeated string labels = 3;
24
17
}
25
18
26
19
// Resume is the mill's reply to Hello. The executor replays every buffered
+30
spindle/mill/token.go
+30
spindle/mill/token.go
···
1
+
package mill
2
+
3
+
import (
4
+
"crypto/rand"
5
+
"crypto/sha256"
6
+
"encoding/base64"
7
+
"encoding/hex"
8
+
)
9
+
10
+
// bootstrapTokenName is the identity the mill seeds cfg.SharedSecret under so a
11
+
// single-secret dev fleet works without minting. Production registers per-executor
12
+
// identities with `spindle mill executor add`.
13
+
const bootstrapTokenName = "dev-bootstrap"
14
+
15
+
// GenerateToken returns a fresh, high-entropy executor token.
16
+
func GenerateToken() (string, error) {
17
+
var b [32]byte
18
+
if _, err := rand.Read(b[:]); err != nil {
19
+
return "", err
20
+
}
21
+
return base64.RawURLEncoding.EncodeToString(b[:]), nil
22
+
}
23
+
24
+
// HashToken hashes a raw token for storage/lookup. The mill only ever persists
25
+
// and compares hashes, so a DB leak never exposes a usable token, and lookup by
26
+
// hash avoids a secret-dependent comparison.
27
+
func HashToken(token string) string {
28
+
sum := sha256.Sum256([]byte(token))
29
+
return hex.EncodeToString(sum[:])
30
+
}
+3
spindle/server.go
+3
spindle/server.go
···
414
414
// but the mill's db/notifier are created inside New.
415
415
m.Attach(s.DB(), s.Notifier())
416
416
s.mill = m
417
+
if err := m.SeedBootstrapToken(); err != nil {
418
+
return fmt.Errorf("seeding mill bootstrap token: %w", err)
419
+
}
417
420
}
418
421
if cfg.Role == config.RoleExecutor {
419
422
s.exec = executor.New(cfg, engines, s.DB(), s.Notifier(), log.SubLogger(logger, "executor"))