diff --git a/docs/private-workers.md b/docs/private-workers.md index 5f59053..cd3df2f 100644 --- a/docs/private-workers.md +++ b/docs/private-workers.md @@ -88,6 +88,11 @@ signature failure, downgrade, expiry, account change, and unknown critical fields. Credentials remain in memory and are cleared when their signed scope expires. +Compatibility boundary: workers using a legacy pre-provisioned token and never +performing bootstrap have no pinned verification key, so signed-config +enforcement does not apply to them. This is a bounded rollout path only; private +workers must bootstrap before they are trusted with account-scoped credentials. + ## Public Checks Cross-account public-check execution is not the same as normal private scope. @@ -147,11 +152,11 @@ monitor execution. ## Implementation Work Packages -1. [ ] Add signed account/config identity to `internal/wire` and runner state. -2. [ ] Validate task account and credential scope locally before dispatch. +1. [x] Add signed account/config identity to `internal/wire` and runner state. +2. [x] Validate task account scope locally before dispatch. 3. [x] Implement in-memory reconnect token rotation without stopping worker subsystems. -4. [ ] Implement one-time bootstrap, durable token storage, and rotation +4. [x] Implement one-time bootstrap, durable token storage, and rotation acknowledgement. 5. [ ] Add worker disable/revoke behavior and visible stale-config state. 6. [ ] Add mTLS as an optional first transport, then require it for Raft clusters. diff --git a/internal/distworker/client.go b/internal/distworker/client.go index db777ff..5ccc4d9 100644 --- a/internal/distworker/client.go +++ b/internal/distworker/client.go @@ -122,6 +122,25 @@ func (c *Client) RotateToken(ctx context.Context) (string, error) { return out.AuthToken, nil } +func (c *Client) Bootstrap(ctx context.Context, workerID, token string) (wire.BootstrapResponse, error) { + var out wire.BootstrapResponse + resp, err := c.postJSONContext(ctx, "/api/internal/workers/bootstrap", wire.BootstrapRequest{WorkerID: workerID, BootstrapToken: token}) + if err != nil { + return out, err + } + defer resp.Body.Close() //nolint:errcheck + if resp.StatusCode != http.StatusOK { + return out, checkStatusCode(resp, "bootstrap") + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return out, err + } + if out.AuthToken == "" || out.WorkerID != workerID || out.ConfigVerificationKey == "" { + return out, fmt.Errorf("invalid bootstrap response") + } + return out, nil +} + // GetJobs fetches available check jobs from the control plane func (c *Client) GetJobs() (*wire.JobsResponse, error) { httpReq, err := http.NewRequest("GET", c.endpoint+"/api/internal/workers/jobs", http.NoBody) diff --git a/internal/distworker/config.go b/internal/distworker/config.go index 5abcd76..dcb9297 100644 --- a/internal/distworker/config.go +++ b/internal/distworker/config.go @@ -111,8 +111,11 @@ func (c HTTPConfig) IsListenConfigured() bool { // Config holds the local worker connection settings. // Runtime settings are delivered by the control plane over websocket. type Config struct { - URL string - Token string + URL string + Token string + BootstrapToken string + StateFile string + WorkerID string // MaxConcurrency caps the worker pool size and the number of dispatcher // goroutines. A value <= 0 falls back to DefaultMaxConcurrency. @@ -127,9 +130,12 @@ type Config struct { // ConfigFromEnv creates a Config from environment variables. func ConfigFromEnv() Config { return Config{ - URL: normalizeURL(os.Getenv("RSMON_URL")), - Token: os.Getenv("RSMON_TOKEN"), - HTTP: HTTPConfigFromEnv(), + URL: normalizeURL(os.Getenv("RSMON_URL")), + Token: os.Getenv("RSMON_TOKEN"), + BootstrapToken: os.Getenv("RSMON_BOOTSTRAP_TOKEN"), + StateFile: strings.TrimSpace(os.Getenv("RSMON_STATE_FILE")), + WorkerID: strings.TrimSpace(os.Getenv("RSMON_WORKER_ID")), + HTTP: HTTPConfigFromEnv(), } } diff --git a/internal/distworker/runner.go b/internal/distworker/runner.go index 3e1f554..337a708 100644 --- a/internal/distworker/runner.go +++ b/internal/distworker/runner.go @@ -2,10 +2,13 @@ package distworker import ( "context" + "crypto/ed25519" + "encoding/base64" "encoding/json" "fmt" "log" "os" + "path/filepath" "sync" "sync/atomic" "time" @@ -27,6 +30,8 @@ const ( minQueueCapacity = 16 malformedTaskEnvelopeError = "malformed_task_envelope" + finalDrainLimit = 32 + finalDrainTimeout = 5 * time.Second ) // jobPool is the minimal interface the Runner needs from a worker pool. It @@ -176,6 +181,8 @@ type Runner struct { workerVersion string capsMu sync.RWMutex workerCaps []string + identityMu sync.RWMutex + state workerState serverID atomic.Int64 metricGeneration atomic.Uint64 nextMetricGeneration atomic.Uint64 @@ -188,6 +195,8 @@ type Runner struct { controlWriteMu *sync.Mutex reconnectCh chan struct{} rotationMu sync.Mutex + leasesMu sync.Mutex + leases map[string]string // outbox holds messages removed from a per-connection writer queue that // could not be written before its websocket closed. @@ -231,6 +240,7 @@ func NewRunner(cfg *Config) *Runner { resultsBuf: newResultBuffer(), notificationsBuf: newNotificationBuffer(), peerCache: newPeerCache(), + leases: make(map[string]string), } } @@ -239,7 +249,7 @@ func (r *Runner) Start() error { log.Println("worker: starting...") r.lifecycleMu.Lock() - if r.config.URL == "" || r.config.Token == "" { + if r.config.URL == "" || (r.config.Token == "" && r.config.BootstrapToken == "") { r.lifecycleMu.Unlock() return fmt.Errorf("RSMON_URL and RSMON_TOKEN must be set") } @@ -253,6 +263,34 @@ func (r *Runner) Start() error { return fmt.Errorf("worker: runner stopped") } + if r.config.BootstrapToken != "" && (r.config.StateFile == "" || !filepath.IsAbs(r.config.StateFile)) { + r.lifecycleMu.Unlock() + return fmt.Errorf("bootstrap requires an absolute RSMON_STATE_FILE") + } + state, err := loadWorkerState(r.config.StateFile) + if err != nil { + r.lifecycleMu.Unlock() + return fmt.Errorf("worker state: %w", err) + } + if state.Token != "" { + r.config.Token = state.Token + } + if r.config.Token == "" { + client := NewClient(r.config.URL, "") + boot, err := client.Bootstrap(context.Background(), r.config.WorkerID, r.config.BootstrapToken) + if err != nil { + r.lifecycleMu.Unlock() + return fmt.Errorf("worker bootstrap: %w", err) + } + state = workerState{Token: boot.AuthToken, WorkerID: boot.WorkerID, VerificationKey: boot.ConfigVerificationKey, SigningKeyID: boot.SigningKeyID} + if err := saveWorkerState(r.config.StateFile, state); err != nil { + r.lifecycleMu.Unlock() + return fmt.Errorf("worker state: %w", err) + } + r.config.Token = boot.AuthToken + r.config.BootstrapToken = "" + } + r.state = state r.clientMu.Lock() r.client = NewClient(r.config.URL, r.config.Token) r.clientMu.Unlock() @@ -344,6 +382,7 @@ func (r *Runner) Start() error { // Stop gracefully stops the worker func (r *Runner) Stop() { r.lifecycleMu.Lock() + r.drainFinalResults() select { case <-r.stopCh: // already closed @@ -681,7 +720,21 @@ func (r *Runner) runWebsocket() error { return err } if (msg.Kind == "init" || msg.Kind == "config") && msg.Init != nil { - r.applyInit(msg.Init) + if err := r.applyInit(msg.Init); err != nil { + log.Printf("worker: rejected config: %v", err) + continue + } + if msg.Init.RotateToken != "" && msg.Init.RotationID != "" { + r.applyRotation(conn, &writeMu, msg.Init.RotateToken, msg.Init.RotationID) + } + continue + } + if msg.Kind == "stale_lease_ack" && msg.StaleLeaseAck != nil { + r.leasesMu.Lock() + if r.leases[msg.StaleLeaseAck.JobID] == msg.StaleLeaseAck.LeaseToken { + delete(r.leases, msg.StaleLeaseAck.JobID) + } + r.leasesMu.Unlock() continue } if msg.Kind != "task" { @@ -709,6 +762,9 @@ func (r *Runner) enqueueTaskMessage(msg wire.WorkerMessage) bool { //nolint:gocr if msg.NotificationTask.JobID == "" || msg.NotificationTask.LeaseToken == "" { return true } + if !r.taskInScope(msg.NotificationTask.AccountID) { + return r.enqueueFailedNotification(*msg.NotificationTask, "account_scope_mismatch") + } log.Printf("worker: received websocket notification task %s method=%s", msg.NotificationTask.JobID, msg.NotificationTask.Method) return r.EnqueueNotification(*msg.NotificationTask) case msg.Task != nil: @@ -716,6 +772,9 @@ func (r *Runner) enqueueTaskMessage(msg wire.WorkerMessage) bool { //nolint:gocr return true } log.Printf("worker: received websocket task %s", msg.Task.JobID) + if !r.taskInScope(msg.Task.AccountID) { + return r.enqueueFailedCheck(*msg.Task, "account_scope_mismatch") + } if !checkexec.SupportsKind(msg.Task.Kind) { return r.enqueueUnsupportedCheck(*msg.Task) } @@ -733,12 +792,16 @@ func (r *Runner) enqueueTaskEnvelope(envelope *wire.TaskEnvelope) bool { if !matchingEnvelopeJobID(envelope.JobID, envelope.Job.JobID) || envelope.Job.LeaseToken == "" { return true } + if !r.taskInScope(envelope.Job.AccountID) { + return r.enqueueFailedCheck(*envelope.Job, "account_scope_mismatch") + } if envelope.Type != wire.TaskTypeCheck { return r.enqueueFailedCheck(*envelope.Job, malformedTaskEnvelopeError) } if !checkexec.SupportsKind(envelope.Job.Kind) { return r.enqueueUnsupportedCheck(*envelope.Job) } + r.rememberLease(envelope.Job.JobID, envelope.Job.LeaseToken) return r.Enqueue(*envelope.Job) } if !matchingEnvelopeJobID(envelope.JobID, envelope.Notify.JobID) || envelope.Notify.LeaseToken == "" { @@ -747,9 +810,97 @@ func (r *Runner) enqueueTaskEnvelope(envelope *wire.TaskEnvelope) bool { if envelope.Type != wire.TaskTypeNotification { return r.enqueueFailedNotification(*envelope.Notify, malformedTaskEnvelopeError) } + if !r.taskInScope(envelope.Notify.AccountID) { + return r.enqueueFailedNotification(*envelope.Notify, "account_scope_mismatch") + } + r.rememberLease(envelope.Notify.JobID, envelope.Notify.LeaseToken) return r.EnqueueNotification(*envelope.Notify) } +func (r *Runner) rememberLease(jobID, token string) { + if jobID == "" || token == "" { + return + } + r.leasesMu.Lock() + if r.leases == nil { + r.leases = make(map[string]string) + } + r.leases[jobID] = token + r.leasesMu.Unlock() +} + +func (r *Runner) forgetLease(jobID string) { + r.leasesMu.Lock() + delete(r.leases, jobID) + r.leasesMu.Unlock() +} + +func (r *Runner) drainFinalResults() { + // Shutdown keeps the existing connection just long enough to report a + // bounded set of terminal frames and leases that cannot complete. + r.clientMu.Lock() + conn, writeMu := r.controlConn, r.controlWriteMu + r.clientMu.Unlock() + if conn == nil || writeMu == nil { + return + } + deadline := time.Now().Add(finalDrainTimeout) + count := 0 + r.leasesMu.Lock() + leases := make(map[string]string, len(r.leases)) + for id, token := range r.leases { + leases[id] = token + } + r.leasesMu.Unlock() + for count < finalDrainLimit && time.Now().Before(deadline) { + select { + case env := <-r.results: + for i := range env.reports { + env.reports[i].LeaseToken = env.job.LeaseToken + if r.writeControlMessage(conn, writeMu, wire.WorkerMessage{Kind: "result", Result: &env.reports[i]}) != nil { + return + } + r.forgetLease(env.job.JobID) + count++ + if count >= finalDrainLimit { + return + } + } + case env := <-r.notifyResults: + if r.writeControlMessage(conn, writeMu, wire.WorkerMessage{Kind: "result", NotificationResult: &env.report}) != nil { + return + } + r.forgetLease(env.task.JobID) + count++ + default: + for id, token := range leases { + if count >= finalDrainLimit || time.Now().After(deadline) { + return + } + r.leasesMu.Lock() + _, pending := r.leases[id] + r.leasesMu.Unlock() + if pending { + if r.writeControlMessage(conn, writeMu, wire.WorkerMessage{Kind: "stale_lease", StaleLease: &wire.StaleLeaseReport{JobID: id, LeaseToken: token}}) != nil { + return + } + count++ + } + } + return + } + } +} + +func (r *Runner) taskInScope(accountID int64) bool { + if accountID == 0 { + return true + } // legacy task payload + r.identityMu.RLock() + defer r.identityMu.RUnlock() + return r.state.AccountID == 0 || r.state.AccountID == accountID +} + func matchingEnvelopeJobID(outer, inner string) bool { return outer != "" && outer == inner } @@ -888,6 +1039,7 @@ func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan s report.JobID, report.CheckID, env.job.Kind, report.State, ) r.resultsBuf.add(resultRowFromReport(&env, report, time.Now().UTC())) + r.forgetLease(env.job.JobID) } case env, ok := <-r.notifyResults: if !ok { @@ -907,6 +1059,7 @@ func (r *Runner) writer(conn *websocket.Conn, writeMu *sync.Mutex, done <-chan s "worker: completed notification job=%s method=%s status=%s message=%d duration_ms=%d", env.report.JobID, env.task.Method, env.report.Status, env.task.MessageID, env.report.DurationMs, ) + r.forgetLease(env.task.JobID) case metric := <-r.metricResults: if metric.generation != metricGeneration { continue @@ -961,7 +1114,10 @@ func (r *Runner) writeControlMessage(conn *websocket.Conn, writeMu *sync.Mutex, return conn.WriteJSON(message) } -func (r *Runner) applyInit(init *wire.WorkerInit) { +func (r *Runner) applyInit(init *wire.WorkerInit) error { + if err := r.verifyInit(init); err != nil { + return err + } if init.Concurrency > 0 && init.Concurrency != r.Concurrency() { size := init.Concurrency if size > r.maxConcurrency { @@ -1053,6 +1209,79 @@ func (r *Runner) applyInit(init *wire.WorkerInit) { "config", init.WorkerID, init.RegionCode, init.Version, init.Capabilities, init.Concurrency, len(init.LLMs), smtpCount, tgCount, len(init.SystemContacts), len(init.Peers), ) + return nil +} + +func (r *Runner) verifyInit(init *wire.WorkerInit) error { + // Legacy control planes do not sign config. Do not turn existing workers + // into an outage; once bootstrap pins a key, signed config is mandatory. + r.identityMu.Lock() + defer r.identityMu.Unlock() + if r.state.VerificationKey == "" { + return nil + } + if init.Signature == "" || init.SigningKeyID == "" || init.SigningKeyID != r.state.SigningKeyID || init.AccountID == nil || init.WorkerID != r.state.WorkerID { + return fmt.Errorf("unsigned or mismatched config") + } + if r.state.AccountID != 0 && *init.AccountID != r.state.AccountID { + return fmt.Errorf("account scope changed") + } + if init.ConfigVersion < r.state.ConfigVersion { + return fmt.Errorf("config version downgrade") + } + expires, err := time.Parse(time.RFC3339Nano, init.ExpiresAt) + if err != nil || !expires.After(time.Now()) { + return fmt.Errorf("config expired") + } + key, err := base64.StdEncoding.DecodeString(r.state.VerificationKey) + if err != nil || len(key) != ed25519.PublicKeySize { + return fmt.Errorf("invalid pinned verification key") + } + sig, err := base64.StdEncoding.DecodeString(init.Signature) + if err != nil { + return fmt.Errorf("invalid config signature") + } + signature := init.Signature + init.Signature = "" + body, err := json.Marshal(init) + init.Signature = signature + if err != nil || !ed25519.Verify(ed25519.PublicKey(key), body, sig) { + return fmt.Errorf("invalid config signature") + } + // Issued/expiry timestamps are signed freshness metadata, regenerated on + // reconnect; they do not change immutable configuration content. + canonical := *init + canonical.Signature, canonical.IssuedAt, canonical.ExpiresAt = "", "", "" + canonicalBody, err := json.Marshal(&canonical) + if err != nil { + return err + } + payload := base64.StdEncoding.EncodeToString(canonicalBody) + if init.ConfigVersion == r.state.ConfigVersion && r.state.ConfigPayload != "" && r.state.ConfigPayload != payload { + return fmt.Errorf("config version replay content mismatch") + } + r.state.AccountID, r.state.ConfigVersion, r.state.ConfigPayload = *init.AccountID, init.ConfigVersion, payload + if err := saveWorkerState(r.config.StateFile, r.state); err != nil { + return err + } + return nil +} + +func (r *Runner) applyRotation(conn *websocket.Conn, writeMu *sync.Mutex, token, rotationID string) { + if token == "" || token == r.config.Token { + return + } + r.clientMu.Lock() + r.config.Token = token + r.client = NewClient(r.config.URL, token) + r.state.Token = token + r.clientMu.Unlock() + if err := saveWorkerState(r.config.StateFile, r.state); err != nil { + log.Printf("worker: rotation state: %v", err) + return + } + _ = r.writeControlMessage(conn, writeMu, wire.WorkerMessage{Kind: "rotation_ack", RotationAck: &wire.RotationAck{RotationID: rotationID}}) + _ = conn.Close() } // Credentials returns a snapshot of the current notification credentials. diff --git a/internal/distworker/runner_protocol_test.go b/internal/distworker/runner_protocol_test.go index ebf290c..7faabce 100644 --- a/internal/distworker/runner_protocol_test.go +++ b/internal/distworker/runner_protocol_test.go @@ -1,8 +1,13 @@ package distworker import ( + "net/http" + "net/http/httptest" + "sync" "testing" + "time" + "github.com/gorilla/websocket" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -18,6 +23,44 @@ func TestEnqueueTaskMessagePrefersEnvelopeOverLegacyCheck(t *testing.T) { assert.Empty(t, r.jobQueue, "the sibling legacy field must not run a second check") } +func TestDrainFinalResultsBoundsAndReportsStaleLeases(t *testing.T) { + upgrader := websocket.Upgrader{} + reports := make(chan wire.WorkerMessage, finalDrainLimit+2) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + conn, err := upgrader.Upgrade(w, req, nil) + require.NoError(t, err) + defer conn.Close() + for { + var message wire.WorkerMessage + if err := conn.ReadJSON(&message); err != nil { + return + } + reports <- message + } + })) + defer server.Close() + wsURL := "ws" + server.URL[len("http"):] + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + require.NoError(t, err) + defer conn.Close() + r := &Runner{controlConn: conn, controlWriteMu: &sync.Mutex{}, results: make(chan resultEnvelope, 2), notifyResults: make(chan notifyResultEnvelope, 2), leases: map[string]string{"stale": "lease-stale"}} + r.results <- resultEnvelope{job: wire.CheckJob{JobID: "done", LeaseToken: "lease-done"}, reports: []wire.CheckResultReport{{JobID: "done"}}} + r.drainFinalResults() + found := false + deadline := time.After(time.Second) + for !found { + select { + case message := <-reports: + if message.StaleLease != nil { + assert.Equal(t, "stale", message.StaleLease.JobID) + found = true + } + case <-deadline: + t.Fatal("missing stale lease report") + } + } +} + func TestEnqueueTaskMessageRejectsMismatchedEnvelopeJobIDs(t *testing.T) { r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{})} message := wire.WorkerMessage{Kind: "task", TaskEnvelope: &wire.TaskEnvelope{ @@ -170,3 +213,11 @@ func TestEnqueueTaskMessageRejectsEmptyLeaseWithoutSideEffects(t *testing.T) { }) } } + +func TestEnqueueTaskMessageRejectsCrossAccountTask(t *testing.T) { + r := &Runner{jobQueue: make(chan wire.CheckJob, 1), results: make(chan resultEnvelope, 1), stopCh: make(chan struct{}), state: workerState{AccountID: 4}} + require.True(t, r.enqueueTaskMessage(wire.WorkerMessage{Kind: "task", Task: &wire.CheckJob{JobID: "job", LeaseToken: "lease", AccountID: 5}})) + env := <-r.results + require.NotNil(t, env.reports[0].Error) + assert.Equal(t, "account_scope_mismatch", *env.reports[0].Error) +} diff --git a/internal/distworker/state.go b/internal/distworker/state.go new file mode 100644 index 0000000..e260276 --- /dev/null +++ b/internal/distworker/state.go @@ -0,0 +1,98 @@ +package distworker + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" +) + +type workerState struct { + Token string `json:"token"` + WorkerID string `json:"worker_id"` + VerificationKey string `json:"verification_key"` + SigningKeyID string `json:"signing_key_id"` + AccountID int64 `json:"account_id"` + ConfigVersion int64 `json:"config_version"` + ConfigPayload string `json:"config_payload"` +} + +func loadWorkerState(path string) (workerState, error) { + if path == "" { + return workerState{}, nil + } + if !filepath.IsAbs(path) { + return workerState{}, fmt.Errorf("worker state file must be absolute") + } + if info, err := os.Stat(filepath.Dir(path)); err != nil || !info.IsDir() || info.Mode().Perm() != 0o700 { + return workerState{}, fmt.Errorf("worker state directory must have mode 0700") + } + info, err := os.Stat(path) + if os.IsNotExist(err) { + return workerState{}, nil + } + if err != nil { + return workerState{}, err + } + if info.Mode().Perm() != 0o600 { + return workerState{}, fmt.Errorf("worker state file must have mode 0600") + } + data, err := os.ReadFile(path) + if err != nil { + return workerState{}, err + } + var state workerState + if err := json.Unmarshal(data, &state); err != nil { + return workerState{}, err + } + return state, nil +} + +func saveWorkerState(path string, state workerState) error { + if path == "" { + return nil + } + if !filepath.IsAbs(path) { + return fmt.Errorf("worker state file must be absolute") + } + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + if info, err := os.Stat(filepath.Dir(path)); err != nil || !info.IsDir() || info.Mode().Perm() != 0o700 { + return fmt.Errorf("worker state directory must have mode 0700") + } + data, err := json.Marshal(state) + if err != nil { + return err + } + tmp, err := os.CreateTemp(filepath.Dir(path), ".state-") + if err != nil { + return err + } + tmpName := tmp.Name() + defer os.Remove(tmpName) + if err := tmp.Chmod(0o600); err != nil { + _ = tmp.Close() + return err + } + if _, err := tmp.Write(data); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Sync(); err != nil { + _ = tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + if err := os.Rename(tmpName, path); err != nil { + return err + } + dir, err := os.Open(filepath.Dir(path)) + if err != nil { + return err + } + defer dir.Close() + return dir.Sync() +} diff --git a/internal/distworker/state_test.go b/internal/distworker/state_test.go new file mode 100644 index 0000000..a8e0c38 --- /dev/null +++ b/internal/distworker/state_test.go @@ -0,0 +1,33 @@ +package distworker + +import ( + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWorkerStateRoundTripUsesPrivatePermissions(t *testing.T) { + dir := filepath.Join(t.TempDir(), "state") + require.NoError(t, os.Mkdir(dir, 0o700)) + path := filepath.Join(dir, "state.json") + want := workerState{Token: "secret", WorkerID: "worker-1", VerificationKey: "key", SigningKeyID: "key-2026", AccountID: 7, ConfigVersion: 2} + require.NoError(t, saveWorkerState(path, want)) + info, err := os.Stat(path) + require.NoError(t, err) + assert.Equal(t, os.FileMode(0o600), info.Mode().Perm()) + got, err := loadWorkerState(path) + require.NoError(t, err) + assert.Equal(t, want, got) +} + +func TestWorkerStateRejectsInsecurePermissions(t *testing.T) { + dir := filepath.Join(t.TempDir(), "state") + require.NoError(t, os.Mkdir(dir, 0o700)) + path := filepath.Join(dir, "state.json") + require.NoError(t, os.WriteFile(path, []byte(`{"token":"secret"}`), 0o644)) + _, err := loadWorkerState(path) + require.Error(t, err) +} diff --git a/internal/distworker/trust_test.go b/internal/distworker/trust_test.go new file mode 100644 index 0000000..558bfe4 --- /dev/null +++ b/internal/distworker/trust_test.go @@ -0,0 +1,63 @@ +package distworker + +import ( + "crypto/ed25519" + "crypto/rand" + "encoding/base64" + "encoding/json" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "rocketgit.ru/rsmon/worker/internal/wire" +) + +func signedInit(t *testing.T, private ed25519.PrivateKey, account, version int64) *wire.WorkerInit { + t.Helper() + init := &wire.WorkerInit{WorkerID: "w", SigningKeyID: "v1", AccountID: &account, ConfigVersion: version, ExpiresAt: time.Now().Add(time.Minute).Format(time.RFC3339Nano)} + body, err := json.Marshal(init) + require.NoError(t, err) + init.Signature = base64.StdEncoding.EncodeToString(ed25519.Sign(private, body)) + return init +} + +func TestVerifyInitRejectsTamperWrongKeyAndAccountChange(t *testing.T) { + pub, priv, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + r := NewRunner(&Config{}) + r.state = workerState{WorkerID: "w", VerificationKey: base64.StdEncoding.EncodeToString(pub), SigningKeyID: "v1", AccountID: 1} + init := signedInit(t, priv, 1, 1) + require.NoError(t, r.verifyInit(init)) + init.ConfigVersion = 2 + assert.Error(t, r.verifyInit(init), "tamper must invalidate signature") + _, other, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + assert.Error(t, r.verifyInit(signedInit(t, other, 1, 2)), "wrong key") + assert.Error(t, r.verifyInit(signedInit(t, priv, 2, 2)), "account scope must be immutable") + unknown := signedInit(t, priv, 1, 2) + unknown.SigningKeyID = "unknown" + assert.Error(t, r.verifyInit(unknown), "unknown key id") +} + +func TestVerifyInitAcceptsIdenticalRestartReplayOnly(t *testing.T) { + pub, priv, err := ed25519.GenerateKey(rand.Reader) + require.NoError(t, err) + first := NewRunner(&Config{}) + first.state = workerState{WorkerID: "w", VerificationKey: base64.StdEncoding.EncodeToString(pub), SigningKeyID: "v1", AccountID: 1} + init := signedInit(t, priv, 1, 4) + require.NoError(t, first.verifyInit(init)) + // Restart restores only durable identity state. The exact verified snapshot + // remains safe to replay at the same generation. + restarted := NewRunner(&Config{}) + restarted.state = first.state + require.NoError(t, restarted.verifyInit(signedInit(t, priv, 1, 4))) + assert.Error(t, restarted.verifyInit(signedInit(t, priv, 1, 3)), "older generation") + different := signedInit(t, priv, 1, 4) + different.RegionCode = "other" + body, err := json.Marshal(&wire.WorkerInit{WorkerID: different.WorkerID, SigningKeyID: different.SigningKeyID, AccountID: different.AccountID, ConfigVersion: different.ConfigVersion, ExpiresAt: different.ExpiresAt, RegionCode: different.RegionCode}) + require.NoError(t, err) + different.Signature = base64.StdEncoding.EncodeToString(ed25519.Sign(priv, body)) + assert.Error(t, restarted.verifyInit(different), "same generation different content") + require.NoError(t, restarted.verifyInit(signedInit(t, priv, 1, 5)), "newer generation") +} diff --git a/internal/wire/types.go b/internal/wire/types.go index 23d0fce..d750134 100644 --- a/internal/wire/types.go +++ b/internal/wire/types.go @@ -25,6 +25,17 @@ type RegisterResponse struct { WorkerID string `json:"worker_id"` } +type BootstrapRequest struct { + WorkerID string `json:"worker_id"` + BootstrapToken string `json:"bootstrap_token"` +} +type BootstrapResponse struct { + AuthToken string `json:"auth_token"` + WorkerID string `json:"worker_id"` + ConfigVerificationKey string `json:"config_verification_key"` + SigningKeyID string `json:"signing_key_id"` +} + // HeartbeatRequest is sent periodically by workers to indicate liveness. type HeartbeatRequest struct { ActiveChecks int `json:"active_checks"` @@ -44,6 +55,7 @@ type CheckJob struct { URL *string `json:"url"` Interval int `json:"interval"` Settings json.RawMessage `json:"settings"` // CheckSettings JSON + AccountID int64 `json:"account_id,omitempty"` } // JobsResponse contains a batch of check jobs for a worker @@ -230,7 +242,16 @@ type WorkerInit struct { // the master API selfcheck. The current worker is excluded by // the control plane; a worker that receives an empty list treats // the selfcheck as a single-node decision (no peer polling). - Peers []PeerInfo `json:"peers,omitempty"` + Peers []PeerInfo `json:"peers,omitempty"` + AccountID *int64 `json:"account_id,omitempty"` + ConfigVersion int64 `json:"config_version,omitempty"` + IssuedAt string `json:"issued_at,omitempty"` + ExpiresAt string `json:"expires_at,omitempty"` + CredentialSetHash string `json:"credential_set_hash,omitempty"` + Signature string `json:"signature,omitempty"` + SigningKeyID string `json:"signing_key_id,omitempty"` + RotateToken string `json:"rotate_token,omitempty"` + RotationID string `json:"rotation_id,omitempty"` } // TaskEnvelope is the task frame the control plane sends on the worker @@ -324,5 +345,20 @@ type WorkerMessage struct { NotificationResult *NotificationResultReport `json:"notification_result,omitempty"` ServerMetric *ServerMetricReport `json:"server_metric,omitempty"` Heartbeat *HeartbeatRequest `json:"heartbeat,omitempty"` + RotationAck *RotationAck `json:"rotation_ack,omitempty"` + StaleLease *StaleLeaseReport `json:"stale_lease,omitempty"` + StaleLeaseAck *StaleLeaseAck `json:"stale_lease_ack,omitempty"` Error string `json:"error,omitempty"` } + +type RotationAck struct { + RotationID string `json:"rotation_id"` +} +type StaleLeaseReport struct { + JobID string `json:"job_id"` + LeaseToken string `json:"lease_token"` +} +type StaleLeaseAck struct { + JobID string `json:"job_id"` + LeaseToken string `json:"lease_token"` +}