diff --git a/coordinator/cmd/coordinator/setup_cmd.go b/coordinator/cmd/coordinator/setup_cmd.go index 05eb7bb..aabd5c0 100644 --- a/coordinator/cmd/coordinator/setup_cmd.go +++ b/coordinator/cmd/coordinator/setup_cmd.go @@ -16,8 +16,8 @@ import ( func runSetup(args []string) error { flags := flag.NewFlagSet("setup", flag.ContinueOnError) flags.Usage = func() { - fmt.Fprintf(flags.Output(), "usage: coordinator setup [options]\n") - fmt.Fprintf(flags.Output(), "Provisions the coordinator database, schema, and local .env settings.\n\n") + _, _ = fmt.Fprintf(flags.Output(), "usage: coordinator setup [options]\n") + _, _ = fmt.Fprintf(flags.Output(), "Provisions the coordinator database, schema, and local .env settings.\n\n") flags.PrintDefaults() } var ( @@ -68,7 +68,7 @@ func runSetup(args []string) error { log.Error("setup failed", "err", err) return err } - fmt.Fprint(os.Stdout, summary) + _, _ = fmt.Fprint(os.Stdout, summary) log.Info("setup complete") return nil } diff --git a/coordinator/internal/agent/auth.go b/coordinator/internal/agent/auth.go index 029827b..546ca1f 100644 --- a/coordinator/internal/agent/auth.go +++ b/coordinator/internal/agent/auth.go @@ -2,6 +2,7 @@ package agent import ( "bytes" + "context" "encoding/json" "fmt" "io" @@ -69,7 +70,7 @@ func (p *WorkerKeyToken) exchangeLocked() error { if err != nil { return err } - request, err := http.NewRequest(http.MethodPost, p.userserviceURL+"/worker-tokens/exchange", bytes.NewReader(payload)) + request, err := http.NewRequestWithContext(context.Background(), http.MethodPost, p.userserviceURL+"/worker-tokens/exchange", bytes.NewReader(payload)) if err != nil { return err } @@ -79,7 +80,7 @@ func (p *WorkerKeyToken) exchangeLocked() error { if err != nil { return fmt.Errorf("worker key exchange request failed") } - defer response.Body.Close() + defer func() { _ = response.Body.Close() }() if response.StatusCode != http.StatusOK { return fmt.Errorf("worker key exchange rejected with status %d", response.StatusCode) } diff --git a/coordinator/internal/agent/auth_test.go b/coordinator/internal/agent/auth_test.go index 845e942..7519375 100644 --- a/coordinator/internal/agent/auth_test.go +++ b/coordinator/internal/agent/auth_test.go @@ -11,8 +11,7 @@ import ( func TestWorkerKeyTokenExchangesAndCaches(t *testing.T) { var exchanges atomic.Int64 - var server *httptest.Server - server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/worker-tokens/exchange" { http.NotFound(w, r) return @@ -53,8 +52,7 @@ func TestWorkerKeyTokenExchangesAndCaches(t *testing.T) { } func TestWorkerKeyTokenRejectsBadKey(t *testing.T) { - var server *httptest.Server - server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusUnauthorized) })) defer server.Close() @@ -76,8 +74,7 @@ func TestNewTokenProviderSelectsStrategy(t *testing.T) { func TestClientRefreshesTokenOnceOn401(t *testing.T) { var attempts atomic.Int64 - var server *httptest.Server - server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { attempts.Add(1) if attempts.Load() == 1 { w.WriteHeader(http.StatusUnauthorized) diff --git a/coordinator/internal/agent/client.go b/coordinator/internal/agent/client.go index 439c428..1087f84 100644 --- a/coordinator/internal/agent/client.go +++ b/coordinator/internal/agent/client.go @@ -2,6 +2,7 @@ package agent import ( "bytes" + "context" "crypto/sha256" "encoding/hex" "encoding/json" @@ -215,7 +216,7 @@ func (c *Client) Download(uri, destination string) (string, error) { if resolved.Scheme != "http" && resolved.Scheme != "https" { return "", fmt.Errorf("input URI must be an HTTP(S) URL") } - request, err := http.NewRequest(http.MethodGet, resolved.String(), nil) + request, err := http.NewRequestWithContext(context.Background(), http.MethodGet, resolved.String(), nil) if err != nil { return "", err } @@ -230,17 +231,18 @@ func (c *Client) Download(uri, destination string) (string, error) { if err != nil { return "", &TransientError{msg: "input download failed"} } - defer response.Body.Close() + defer func() { _ = response.Body.Close() }() if response.StatusCode == http.StatusUnauthorized && c.refreshAndRetry() { return c.Download(uri, destination) } if response.StatusCode != http.StatusOK { return "", &CoordinatorError{msg: fmt.Sprintf("input download rejected with status %d", response.StatusCode)} } - if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil { + if err := os.MkdirAll(filepath.Dir(destination), 0o750); err != nil { return "", err } - target, err := os.OpenFile(destination, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o644) + // #nosec G304 -- destination is the worker's own attempt directory file. + target, err := os.OpenFile(destination, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) if err != nil { return "", err } @@ -248,7 +250,7 @@ func (c *Client) Download(uri, destination string) (string, error) { _, copyErr := io.Copy(io.MultiWriter(target, digest), response.Body) closeErr := target.Close() if copyErr != nil { - os.Remove(destination) + _ = os.Remove(destination) return "", &TransientError{msg: "input download interrupted"} } if closeErr != nil { @@ -259,29 +261,30 @@ func (c *Client) Download(uri, destination string) (string, error) { // Upload streams a partial artifact and verifies the returned metadata. func (c *Client) Upload(task *Task, workerID string, path, contentType string) (*Uploaded, error) { + // #nosec G304 -- the upload path is this worker's own artifact file. file, err := os.Open(path) if err != nil { return nil, err } info, err := file.Stat() if err != nil { - file.Close() + _ = file.Close() return nil, err } digest := sha256.New() if _, err := io.Copy(digest, file); err != nil { - file.Close() + _ = file.Close() return nil, err } if _, err := file.Seek(0, io.SeekStart); err != nil { - file.Close() + _ = file.Close() return nil, err } localSHA := hex.EncodeToString(digest.Sum(nil)) uploadURL := c.baseURL + "/tasks/" + url.PathEscape(task.TaskID) + "/artifacts/" + url.PathEscape(filepath.Base(path)) - request, err := http.NewRequest(http.MethodPut, uploadURL, file) + request, err := http.NewRequestWithContext(context.Background(), http.MethodPut, uploadURL, file) if err != nil { - file.Close() + _ = file.Close() return nil, err } request.ContentLength = info.Size() @@ -290,18 +293,18 @@ func (c *Client) Upload(task *Task, workerID string, path, contentType string) ( request.Header.Set("X-Task-Attempt", strconv.Itoa(task.Attempt)) headers, err := c.authHeaders() if err != nil { - file.Close() + _ = file.Close() return nil, err } for name, value := range headers { request.Header.Set(name, value) } response, err := c.apiClient.Do(request) - file.Close() + _ = file.Close() if err != nil { return nil, &TransientError{msg: "artifact upload failed"} } - defer response.Body.Close() + defer func() { _ = response.Body.Close() }() raw, err := io.ReadAll(io.LimitReader(response.Body, 1<<20)) if err != nil { return nil, &TransientError{msg: "artifact upload interrupted"} @@ -334,7 +337,7 @@ func (c *Client) requestJSON(method, path string, payload any) (int, map[string] if err != nil { return 0, nil, err } - request, err := http.NewRequest(method, c.baseURL+path, bytes.NewReader(body)) + request, err := http.NewRequestWithContext(context.Background(), method, c.baseURL+path, bytes.NewReader(body)) if err != nil { return 0, nil, err } @@ -350,7 +353,7 @@ func (c *Client) requestJSON(method, path string, payload any) (int, map[string] if err != nil { return 0, nil, &TransientError{msg: "coordinator request failed"} } - defer response.Body.Close() + defer func() { _ = response.Body.Close() }() raw, err := io.ReadAll(io.LimitReader(response.Body, 1<<20)) if err != nil { return 0, nil, &TransientError{msg: "coordinator request interrupted"} diff --git a/coordinator/internal/agent/client_test.go b/coordinator/internal/agent/client_test.go index 5f7fee3..f78ec31 100644 --- a/coordinator/internal/agent/client_test.go +++ b/coordinator/internal/agent/client_test.go @@ -3,6 +3,7 @@ package agent import ( "crypto/sha256" "encoding/json" + "errors" "fmt" "net/http" "net/http/httptest" @@ -20,7 +21,8 @@ func newTestClient(t *testing.T, server *httptest.Server) *Client { func TestClientRegisterClaimHeartbeat(t *testing.T) { var registered, claimed, heartbeated bool - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var server *httptest.Server //nolint:staticcheck // the handler closure references server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Header.Get("Authorization") != "Bearer test-token" { http.Error(w, "missing token", http.StatusUnauthorized) return @@ -76,7 +78,8 @@ func TestClientRegisterClaimHeartbeat(t *testing.T) { } func TestClientClaimEmptyAndConflict(t *testing.T) { - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var server *httptest.Server //nolint:staticcheck // the handler closure references server + server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/tasks/claim": w.WriteHeader(http.StatusNoContent) @@ -103,14 +106,14 @@ func TestClientClaimEmptyAndConflict(t *testing.T) { } if _, err := client.Heartbeat(claimed, "worker"); err == nil { t.Error("expected conflict error") - } else if _, ok := err.(*ConflictError); !ok { + } else if !errors.As(err, &conflictError) { t.Errorf("error type = %T", err) } } func TestClientUploadSubmitFail(t *testing.T) { var uploadedPath string - var server *httptest.Server + var server *httptest.Server //nolint:staticcheck // the handler closure references server server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch { case r.Method == http.MethodPut && strings.HasPrefix(r.URL.Path, "/tasks/11111111-1111-4111-8111-111111111111/artifacts/"): @@ -162,14 +165,12 @@ func TestClientUploadSubmitFail(t *testing.T) { func TestClientDownloadVerifiesChecksumAndStripsAuthOnRedirect(t *testing.T) { var redirectedAuth string - var bucket *httptest.Server - bucket = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + bucket := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { redirectedAuth = r.Header.Get("Authorization") _, _ = w.Write([]byte("input bytes")) })) defer bucket.Close() - var server *httptest.Server - server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path == "/tasks/11111111-1111-4111-8111-111111111111/input" { http.Redirect(w, r, bucket.URL+"/presigned", http.StatusFound) return diff --git a/coordinator/internal/agent/daemon.go b/coordinator/internal/agent/daemon.go index c63a40c..3af610f 100644 --- a/coordinator/internal/agent/daemon.go +++ b/coordinator/internal/agent/daemon.go @@ -1,10 +1,8 @@ package agent import ( - "crypto/sha256" - "encoding/hex" + "errors" "fmt" - "io" "log/slog" "os" "path/filepath" @@ -21,6 +19,9 @@ type Outcome struct { // Daemon is the agent state machine: register, claim, execute via the Python // task runner, upload, and submit — mirroring the Python worker's lifecycle. +// conflictError is the errors.As target for lease conflicts. +var conflictError *ConflictError + type Daemon struct { config *Config client *Client @@ -159,14 +160,14 @@ func (d *Daemon) runOnce() (Outcome, error) { } started := time.Now() taskDir := filepath.Join(d.config.WorkDir, task.TaskID, fmt.Sprint(task.Attempt)) - if err := os.MkdirAll(taskDir, 0o755); err != nil { + if err := os.MkdirAll(taskDir, 0o750); err != nil { return Outcome{Claimed: true}, err } heartbeat := newLeaseHeartbeat(task, workerID, d.client, d.config.Heartbeat) completed := false err = heartbeat.Start() if err != nil { - if _, ok := err.(*ConflictError); ok { + if errors.As(err, &conflictError) { d.log.Warn("lease lost", "task_id", task.TaskID) return Outcome{Claimed: true}, nil } @@ -178,7 +179,7 @@ func (d *Daemon) runOnce() (Outcome, error) { // attempt directories are retained under the work directory. failure := d.executeTask(task, workerID, taskDir, started, heartbeat) if failure != nil { - if _, ok := failure.(*ConflictError); ok { + if errors.As(failure, &conflictError) { d.log.Warn("lease lost", "task_id", task.TaskID) return Outcome{Claimed: true}, nil } @@ -235,17 +236,17 @@ func (d *Daemon) executeTask(task *Task, workerID, taskDir string, started time. func (d *Daemon) reportFailure(task *Task, workerID string, failure error) { var code string - switch failure.(type) { - case *CoordinatorError: + var coordinatorErr *CoordinatorError + if errors.As(failure, &coordinatorErr) { code = "ValueError" - default: + } else { code = "TaskRunnerFailed" } retryable := IsRetryableError(failure) message := SanitizeErrorMessage(failure.Error(), d.config.WorkDir) d.log.Warn("task failed", "task_id", task.TaskID, "error_code", code, "retryable", retryable) if err := d.client.Fail(task, workerID, code, message, retryable); err != nil { - if _, ok := err.(*ConflictError); ok { + if errors.As(err, &conflictError) { d.log.Warn("lease lost while reporting failure", "task_id", task.TaskID) return } @@ -333,17 +334,3 @@ func (h *leaseHeartbeat) nextDelay() time.Duration { func roundSeconds(seconds float64) float64 { return float64(int64(seconds*1000)) / 1000 } - -// File-digest helper used by tests. -func sha256File(path string) (string, error) { - file, err := os.Open(path) - if err != nil { - return "", err - } - defer file.Close() - digest := sha256.New() - if _, err := io.Copy(digest, file); err != nil { - return "", err - } - return hex.EncodeToString(digest.Sum(nil)), nil -} diff --git a/coordinator/internal/agent/sanitize.go b/coordinator/internal/agent/sanitize.go index 7972b4a..a789902 100644 --- a/coordinator/internal/agent/sanitize.go +++ b/coordinator/internal/agent/sanitize.go @@ -1,6 +1,7 @@ package agent import ( + "errors" "fmt" "os" "regexp" @@ -36,10 +37,9 @@ func IsRetryableError(err error) bool { if err == nil { return false } - switch err.(type) { - case *CoordinatorError: - return false - case *os.PathError: + var coordinatorErr *CoordinatorError + var pathErr *os.PathError + if errors.As(err, &coordinatorErr) || errors.As(err, &pathErr) { return false } return true diff --git a/coordinator/internal/agent/taskrunner.go b/coordinator/internal/agent/taskrunner.go index e362be6..770aa0b 100644 --- a/coordinator/internal/agent/taskrunner.go +++ b/coordinator/internal/agent/taskrunner.go @@ -2,7 +2,9 @@ package agent import ( "bytes" + "context" "encoding/json" + "errors" "fmt" "os" "os/exec" @@ -23,7 +25,7 @@ func NewTaskRunner(command []string) *TaskRunner { // directory, the Python entry computes and seals the partial, and the written // manifest is parsed back. stderr is captured for failure reporting. func (r *TaskRunner) Run(task *Task, taskDir string, manifestPath string, extraEnv []string) (*TaskRunnerManifest, error) { - if err := os.MkdirAll(taskDir, 0o755); err != nil { + if err := os.MkdirAll(taskDir, 0o750); err != nil { return nil, err } payload := map[string]any{ @@ -51,17 +53,20 @@ func (r *TaskRunner) Run(task *Task, taskDir string, manifestPath string, extraE "--task-dir", taskDir, "--output", manifestPath, ) - command := exec.Command(r.command[0], args...) + // #nosec G204 -- the command comes from the operator-configured TASK_RUNNER. + command := exec.CommandContext(context.Background(), r.command[0], args...) command.Dir = taskDir command.Env = append(os.Environ(), extraEnv...) var stderr bytes.Buffer command.Stderr = &stderr if err := command.Run(); err != nil { - if exitErr, ok := err.(*exec.ExitError); ok { + var exitErr *exec.ExitError + if errors.As(err, &exitErr) { return nil, runnerExitError(exitErr.ExitCode(), stderr.String()) } return nil, fmt.Errorf("task runner could not be started: %w", err) } + // #nosec G304 -- the manifest path is inside the worker's own task directory. raw, err := os.ReadFile(manifestPath) if err != nil { return nil, fmt.Errorf("task runner produced no result manifest") diff --git a/coordinator/internal/setup/setup.go b/coordinator/internal/setup/setup.go index b7bef55..27bd1d4 100644 --- a/coordinator/internal/setup/setup.go +++ b/coordinator/internal/setup/setup.go @@ -58,7 +58,7 @@ func Run(ctx context.Context, options Options) (string, error) { } report := func(format string, args ...any) { - fmt.Fprintf(options.Out, format+"\n", args...) + _, _ = fmt.Fprintf(options.Out, format+"\n", args...) } report("SciMesh coordinator setup") @@ -195,6 +195,7 @@ func writeEnvFile(options Options, secret, storageDir string) error { "COORDINATOR_STORAGE_DIR=" + storageDir, "", // trailing newline }, "\n") + // #nosec G703 -- the env file path is operator-supplied (--env-file / ENV_FILE). if err := os.WriteFile(path, []byte(content), 0o600); err != nil { return fmt.Errorf("write %s: %w", path, err) } @@ -224,7 +225,7 @@ func SanitizeDatabaseURL(raw string) string { // prompt asks a question and returns the trimmed answer ("" on EOF). func prompt(options Options, question, fallback string) string { - fmt.Fprintf(options.Out, "%s [%s]: ", question, fallback) + _, _ = fmt.Fprintf(options.Out, "%s [%s]: ", question, fallback) reader := bufio.NewReader(options.In) line, err := reader.ReadString('\n') if err != nil && !errors.Is(err, io.EOF) { diff --git a/coordinator/internal/transport/http/ui.go b/coordinator/internal/transport/http/ui.go index 5367877..5adcd21 100644 --- a/coordinator/internal/transport/http/ui.go +++ b/coordinator/internal/transport/http/ui.go @@ -328,6 +328,7 @@ func newJobPayload(catalog *workloads.Catalog) template.JS { if err != nil { return template.JS("null") } + // #nosec G203 -- the payload is marshaled JSON from the embedded workload catalog, injected as script data. return template.JS(encoded) } diff --git a/coordinator/internal/transport/http/ui_docs_internal_test.go b/coordinator/internal/transport/http/ui_docs_internal_test.go index 2ee02d7..318bf1e 100644 --- a/coordinator/internal/transport/http/ui_docs_internal_test.go +++ b/coordinator/internal/transport/http/ui_docs_internal_test.go @@ -1,6 +1,7 @@ package http import ( + "context" "io" "log/slog" "net/http" @@ -34,13 +35,13 @@ func TestUIDocsServesIndexAndNestedFiles(t *testing.T) { server := docsTestServer(t, root) index := httptest.NewRecorder() - server.handleUIDocs(index, httptest.NewRequest(http.MethodGet, "/ui/docs/", nil)) + server.handleUIDocs(index, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/", nil)) if index.Code != http.StatusOK || !strings.Contains(index.Body.String(), "

Home

") { t.Fatalf("index = %d %q", index.Code, index.Body.String()) } page := httptest.NewRecorder() - server.handleUIDocs(page, httptest.NewRequest(http.MethodGet, "/ui/docs/api/page.html", nil)) + server.handleUIDocs(page, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/api/page.html", nil)) if page.Code != http.StatusOK || !strings.Contains(page.Body.String(), "

API page

") { t.Fatalf("nested page = %d %q", page.Code, page.Body.String()) } @@ -54,7 +55,7 @@ func TestUIDocsRejectsPathTraversal(t *testing.T) { } server := docsTestServer(t, root) - request := httptest.NewRequest(http.MethodGet, "/ui/docs/../secret.txt", nil) + request := httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/../secret.txt", nil) request.URL.Path = "/ui/docs/../secret.txt" recorder := httptest.NewRecorder() server.handleUIDocs(recorder, request) @@ -66,14 +67,14 @@ func TestUIDocsRejectsPathTraversal(t *testing.T) { func TestUIDocsShowsBuildHintWhenDisabledOrMissing(t *testing.T) { disabled := docsTestServer(t, "") recorder := httptest.NewRecorder() - disabled.handleUIDocs(recorder, httptest.NewRequest(http.MethodGet, "/ui/docs/", nil)) + disabled.handleUIDocs(recorder, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/", nil)) if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "Documentation is not available") { t.Fatalf("disabled docs = %d %q", recorder.Code, recorder.Body.String()) } missing := docsTestServer(t, filepath.Join(t.TempDir(), "does-not-exist")) recorder = httptest.NewRecorder() - missing.handleUIDocs(recorder, httptest.NewRequest(http.MethodGet, "/ui/docs/", nil)) + missing.handleUIDocs(recorder, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs/", nil)) if recorder.Code != http.StatusOK || !strings.Contains(recorder.Body.String(), "Documentation is not available") { t.Fatalf("missing docs = %d %q", recorder.Code, recorder.Body.String()) } @@ -82,7 +83,7 @@ func TestUIDocsShowsBuildHintWhenDisabledOrMissing(t *testing.T) { func TestUIDocsIndexRedirectsToTrailingSlash(t *testing.T) { server := docsTestServer(t, t.TempDir()) recorder := httptest.NewRecorder() - server.handleUIDocsIndex(recorder, httptest.NewRequest(http.MethodGet, "/ui/docs", nil)) + server.handleUIDocsIndex(recorder, httptest.NewRequestWithContext(context.Background(), http.MethodGet, "/ui/docs", nil)) if recorder.Code != http.StatusPermanentRedirect || recorder.Header().Get("Location") != "/ui/docs/" { t.Fatalf("redirect = %d %q", recorder.Code, recorder.Header().Get("Location")) }