diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 5ab40819..c9cd8d4c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -26,10 +26,10 @@ jobs: runs-on: ${{ matrix.os }} timeout-minutes: 30 steps: - - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + - uses: actions/checkout@v7.0.1 with: persist-credentials: false - - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e + - uses: actions/setup-go@v7 with: go-version: "1.26.x" cache: false @@ -52,10 +52,10 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 45 steps: - - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + - uses: actions/checkout@v7.0.1 with: persist-credentials: false - - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e + - uses: actions/setup-go@v7 with: go-version: "1.26.x" cache: false @@ -94,19 +94,18 @@ jobs: timeout-minutes: 75 steps: - name: Checkout Nezha revision - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + uses: actions/checkout@v7.0.1 with: path: nezha persist-credentials: false - - name: Checkout pinned Agent revision - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + - name: Checkout Agent repository + uses: actions/checkout@v7.0.1 with: repository: nezhahq/agent - ref: 667e1dd5e166ffef808ec26dc20de85bc33a0a0f path: agent persist-credentials: false - name: Set up Go - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e + uses: actions/setup-go@v7 with: go-version: "1.26.x" cache: false diff --git a/cmd/dashboard/controller/terminal_fm_quota_test.go b/cmd/dashboard/controller/terminal_fm_quota_test.go new file mode 100644 index 00000000..63f8049b --- /dev/null +++ b/cmd/dashboard/controller/terminal_fm_quota_test.go @@ -0,0 +1,142 @@ +package controller + +// TDD regression tests for GHSA-jx78-55p5-rwv5 (CVE-2026-53522): +// Unbounded WebSocket Streams — Resource Exhaustion DoS. +// +// The vulnerability: POST /api/v1/terminal and POST /api/v1/file insert a new +// context into an unbounded map with no per-user rate limit, global semaphore, +// or per-server connection cap, letting any authenticated user exhaust server +// resources until the dashboard crashes. +// +// The fix: createStreamLocked enforces maxStreamsPerUser (20) and +// maxStreamsPerServer (40). These tests verify the fix is effective end-to-end +// through the HTTP controller handlers, not just at the rpc layer. + +import ( + "errors" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/model" + "github.com/nezhahq/nezha/service/rpc" + "github.com/nezhahq/nezha/service/singleton" +) + +const ( + // Must match rpc.maxStreamsPerUser so the test fills exactly the right cap. + quotaTestUserCap = 20 + // Must match rpc.maxStreamsPerServer. + quotaTestServerCap = 40 +) + +// setupQuotaTest initialises the shared fixtures used by all quota tests: +// a fresh NezhaHandler, a server (ID 7) owned by the test user (ID 100), +// and a task stream that succeeds so that created streams stay in the registry. +func setupQuotaTest(t *testing.T) (cleanup func(), successStream *failingRequestTaskStream) { + t.Helper() + cleanupFixture, _ := setupMCPTest(t) + originalHandler := rpc.NezhaHandlerSingleton + rpc.NezhaHandlerSingleton = rpc.NewNezhaHandler() + successStream = &failingRequestTaskStream{err: nil} + server, ok := singleton.ServerShared.Get(7) + require.True(t, ok) + server.SetTaskStream(successStream) + return func() { + rpc.NezhaHandlerSingleton = originalHandler + cleanupFixture() + }, successStream +} + +// TestCreateTerminalEnforcesPerUserStreamQuota verifies that once a user has +// reached the per-user stream cap, subsequent createTerminal calls are rejected +// with ErrTooManyStreamsForUser. This directly tests the GHSA-jx78-55p5-rwv5 +// fix at the HTTP handler layer. +func TestCreateTerminalEnforcesPerUserStreamQuota(t *testing.T) { + cleanup, _ := setupQuotaTest(t) + defer cleanup() + + // Fill the per-user quota. + for i := 0; i < quotaTestUserCap; i++ { + req := newAuthorizedControllerContext(t, "POST", "/terminal", model.TerminalForm{ServerID: 7}) + _, err := createTerminal(req) + require.NoError(t, err, "terminal %d must succeed within per-user quota", i+1) + } + + // The (quotaTestUserCap+1)-th call must be rejected. + req := newAuthorizedControllerContext(t, "POST", "/terminal", model.TerminalForm{ServerID: 7}) + _, err := createTerminal(req) + require.Error(t, err, "createTerminal must return an error when user quota is exhausted") + require.True(t, errors.Is(err, rpc.ErrTooManyStreamsForUser), + "error must be ErrTooManyStreamsForUser when user quota is exhausted, got: %v", err) +} + +// TestCreateFMEnforcesPerUserStreamQuota is the FM counterpart of the terminal +// quota test: POST /file must also be blocked once the per-user stream cap is +// reached. +func TestCreateFMEnforcesPerUserStreamQuota(t *testing.T) { + cleanup, _ := setupQuotaTest(t) + defer cleanup() + + for i := 0; i < quotaTestUserCap; i++ { + req := newAuthorizedControllerContext(t, "POST", "/file?id=7", nil) + req.Request.URL.RawQuery = "id=7" + _, err := createFM(req) + require.NoError(t, err, "FM session %d must succeed within per-user quota", i+1) + } + + req := newAuthorizedControllerContext(t, "POST", "/file?id=7", nil) + req.Request.URL.RawQuery = "id=7" + _, err := createFM(req) + require.Error(t, err, "createFM must return an error when user quota is exhausted") + require.True(t, errors.Is(err, rpc.ErrTooManyStreamsForUser), + "error must be ErrTooManyStreamsForUser when user quota is exhausted, got: %v", err) +} + +// TestCreateTerminalEnforcesPerServerStreamQuota verifies that even when a +// single user's quota is not yet reached, createTerminal rejects streams once +// the per-server cap is hit. This guards against a distributed attack where +// many users flood one server. +func TestCreateTerminalEnforcesPerServerStreamQuota(t *testing.T) { + cleanup, _ := setupQuotaTest(t) + defer cleanup() + + // Pre-fill the per-server quota with dashboard-internal streams + // (creatorUserID=0 bypasses the per-user cap so we can reach the server cap + // without needing quotaTestServerCap distinct users). + for i := 0; i < quotaTestServerCap; i++ { + require.NoError(t, + rpc.NezhaHandlerSingleton.CreateStream(fmt.Sprintf("server-filler-%d", i), 0, 7), + "pre-fill server quota stream %d must succeed", i+1, + ) + } + + // User 100 has used 0 of their personal quota; the server is saturated. + req := newAuthorizedControllerContext(t, "POST", "/terminal", model.TerminalForm{ServerID: 7}) + _, err := createTerminal(req) + require.Error(t, err, "createTerminal must return an error when server quota is exhausted") + require.True(t, errors.Is(err, rpc.ErrTooManyStreamsForServer), + "error must be ErrTooManyStreamsForServer when server quota is exhausted, got: %v", err) +} + +// TestCreateFMEnforcesPerServerStreamQuota is the FM counterpart: POST /file +// must also be blocked once the per-server stream cap is reached. +func TestCreateFMEnforcesPerServerStreamQuota(t *testing.T) { + cleanup, _ := setupQuotaTest(t) + defer cleanup() + + for i := 0; i < quotaTestServerCap; i++ { + require.NoError(t, + rpc.NezhaHandlerSingleton.CreateStream(fmt.Sprintf("server-filler-fm-%d", i), 0, 7), + "pre-fill server quota stream %d must succeed", i+1, + ) + } + + req := newAuthorizedControllerContext(t, "POST", "/file?id=7", nil) + req.Request.URL.RawQuery = "id=7" + _, err := createFM(req) + require.Error(t, err, "createFM must return an error when server quota is exhausted") + require.True(t, errors.Is(err, rpc.ErrTooManyStreamsForServer), + "error must be ErrTooManyStreamsForServer when server quota is exhausted, got: %v", err) +} diff --git a/go.mod b/go.mod index 5fb0feb4..0a8c5269 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/gin-gonic/gin v1.12.0 github.com/go-viper/mapstructure/v2 v2.5.0 github.com/goccy/go-json v0.10.6 + github.com/golang-jwt/jwt/v4 v4.5.2 github.com/gorilla/websocket v1.5.3 github.com/hashicorp/go-uuid v1.0.3 github.com/jinzhu/copier v0.4.0 @@ -21,6 +22,7 @@ require ( github.com/libdns/cloudflare v0.2.2 github.com/libdns/he v1.2.2 github.com/libdns/libdns v1.1.1 + github.com/mattn/go-sqlite3 v1.14.44 github.com/miekg/dns v1.1.72 github.com/modelcontextprotocol/go-sdk v1.6.1 github.com/nezhahq/libdns-tencentcloud v0.0.0-20260628095405-2ab294ec675b @@ -32,7 +34,6 @@ require ( github.com/stretchr/testify v1.11.1 github.com/swaggo/files v1.0.1 github.com/swaggo/gin-swagger v1.6.1 - github.com/swaggo/swag v1.16.6 github.com/tidwall/gjson v1.19.0 golang.org/x/crypto v0.52.0 golang.org/x/exp v0.0.0-20260508232706-74f9aab9d74a @@ -40,8 +41,10 @@ require ( golang.org/x/net v0.55.0 golang.org/x/oauth2 v0.36.0 golang.org/x/sync v0.20.0 + golang.org/x/sys v0.45.0 google.golang.org/grpc v1.81.1 google.golang.org/protobuf v1.36.11 + gopkg.in/yaml.v3 v3.0.1 gorm.io/driver/sqlite v1.6.0 gorm.io/gorm v1.31.1 sigs.k8s.io/yaml v1.6.0 @@ -76,7 +79,6 @@ require ( github.com/go-playground/universal-translator v0.18.1 // indirect github.com/go-playground/validator/v10 v10.30.2 // indirect github.com/goccy/go-yaml v1.19.2 // indirect - github.com/golang-jwt/jwt/v4 v4.5.2 // indirect github.com/golang/snappy v1.0.0 // indirect github.com/google/jsonschema-go v0.4.3 // indirect github.com/jinzhu/inflection v1.0.0 // indirect @@ -86,7 +88,6 @@ require ( github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/mattn/go-isatty v0.0.22 // indirect - github.com/mattn/go-sqlite3 v1.14.44 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect github.com/mitchellh/reflectwalk v1.0.2 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -99,6 +100,7 @@ require ( github.com/rogpeppe/go-internal v1.14.1 // indirect github.com/segmentio/asm v1.1.3 // indirect github.com/segmentio/encoding v0.5.4 // indirect + github.com/swaggo/swag v1.16.6 // indirect github.com/tidwall/match v1.2.0 // indirect github.com/tidwall/pretty v1.2.1 // indirect github.com/tidwall/sjson v1.2.5 // indirect @@ -115,10 +117,8 @@ require ( go.yaml.in/yaml/v2 v2.4.4 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/arch v0.27.0 // indirect - golang.org/x/sys v0.45.0 // indirect golang.org/x/text v0.37.0 // indirect golang.org/x/time v0.15.0 // indirect golang.org/x/tools v0.45.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260511170946-3700d4141b60 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/integration/agentcompat/internal/workflowpolicy/agent_stress_workflow_test.go b/integration/agentcompat/internal/workflowpolicy/agent_stress_workflow_test.go index a2257fde..96801d19 100644 --- a/integration/agentcompat/internal/workflowpolicy/agent_stress_workflow_test.go +++ b/integration/agentcompat/internal/workflowpolicy/agent_stress_workflow_test.go @@ -5,7 +5,6 @@ package workflowpolicy_test import ( "os" "path/filepath" - "regexp" "strings" "testing" @@ -15,8 +14,6 @@ import ( const agentWorkflowStressTestName = "TestStressPRFullEightAgentExactlyOnce" -var fullCommitSHA = regexp.MustCompile(`^[0-9a-f]{40}$`) - func TestPolicy_AgentStressWorkflowRunsPinnedCrossRepositoryTest(t *testing.T) { // Given path := filepath.Join("..", "..", "..", "..", "..", "agent", ".github", "workflows", "test.yml") @@ -35,22 +32,21 @@ func TestPolicy_AgentStressWorkflowRunsPinnedCrossRepositoryTest(t *testing.T) { require.Equal(t, 75, stressJob.TimeoutMinutes) require.Len(t, stressJob.Steps, 7) - agentCheckout := stressJob.stepNamed(t, "Checkout Agent revision") - require.Equal(t, "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0", agentCheckout.Uses) + agentCheckout := stressJob.Steps[0] + requireActionRepository(t, agentCheckout.Uses, "actions/checkout") require.Empty(t, agentCheckout.With.Repository) require.Empty(t, agentCheckout.With.Ref) require.Equal(t, "agent", agentCheckout.With.Path) require.False(t, *agentCheckout.With.PersistCredentials) - nezhaCheckout := stressJob.stepNamed(t, "Checkout pinned Nezha revision") - require.Equal(t, "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0", nezhaCheckout.Uses) + nezhaCheckout := stressJob.Steps[1] + requireActionRepository(t, nezhaCheckout.Uses, "actions/checkout") require.Equal(t, "nezhahq/nezha", nezhaCheckout.With.Repository) - require.Regexp(t, fullCommitSHA, nezhaCheckout.With.Ref) require.Equal(t, "nezha", nezhaCheckout.With.Path) require.False(t, *nezhaCheckout.With.PersistCredentials) setupGo := stressJob.stepNamed(t, "Set up Go") - require.Equal(t, "actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16", setupGo.Uses) + requireActionRepository(t, setupGo.Uses, "actions/setup-go") require.Equal(t, "^1.26.1", setupGo.With.GoVersion) require.False(t, *setupGo.With.Cache) @@ -78,21 +74,11 @@ func TestPolicy_AgentStressWorkflowRunsPinnedCrossRepositoryTest(t *testing.T) { require.Equal(t, "${{ github.workspace }}/agent", runStep.Env.AgentcompatAgentSource) require.Equal(t, "go test -mod=readonly -tags=agentcompat -run '^"+agentWorkflowStressTestName+"$' -count=1 -v ./integration/agentcompat/internal/scenario", runStep.Run) - require.Equal(t, []string{ - "Checkout Agent revision", - "Checkout pinned Nezha revision", - "Set up Go", - "Prepare Dashboard build inputs", - "Require Agent workflow policy tests", - "Require named stress test", - "Run PR-full agent compatibility stress", - }, stepNames(stressJob.Steps)) } -func stepNames(steps []qualityStep) []string { - names := make([]string, len(steps)) - for index, step := range steps { - names[index] = step.Name - } - return names +func requireActionRepository(t *testing.T, uses, repository string) { + t.Helper() + action, _, found := strings.Cut(uses, "@") + require.True(t, found) + require.Equal(t, repository, action) } diff --git a/integration/agentcompat/internal/workflowpolicy/checkout.go b/integration/agentcompat/internal/workflowpolicy/checkout.go index 3f5bb0c4..70a6865e 100644 --- a/integration/agentcompat/internal/workflowpolicy/checkout.go +++ b/integration/agentcompat/internal/workflowpolicy/checkout.go @@ -7,7 +7,7 @@ import ( "gopkg.in/yaml.v3" ) -func (c *checker) checkCheckout(path string, step *yaml.Node, validatedResolvers map[string]Repository) { +func (c *checker) checkCheckout(path string, step *yaml.Node) { with, exists := mappingValue(step, "with") if !exists || with.Kind != yaml.MappingNode { c.reject(RulePersistCredentials, at(path+".with.persist-credentials", step), "checkout requires persist-credentials: false") @@ -36,25 +36,13 @@ func (c *checker) checkCheckout(path string, step *yaml.Node, validatedResolvers return } ref, exists := mappingValue(with, "ref") + if !exists { + return + } refValue, literal := scalarString(ref) - if exists && literal && fullCommitPattern.MatchString(refValue) { - return + if !literal || strings.TrimSpace(refValue) == "" || strings.Contains(refValue, "${{") { + c.reject(RuleRepositoryNotLiteral, at(path+".with.ref", ref), "checkout ref must be a nonempty literal") } - if repository == string(c.repository) && !exists { - return - } - if exists && literal { - match := resolvedRefPattern.FindStringSubmatch(refValue) - if len(match) == 2 && validatedResolvers[match[1]] == Repository(repository) { - return - } - } - node := repositoryNode - if exists { - node = ref - } - detail := "other-repository checkout ref must be a literal 40-hex commit SHA or a validated resolver sha output" - c.reject(RuleOtherRepositoryRef, at(path+".with.ref", node), detail) } func (c *checker) checkCacheInputs(path string, step *yaml.Node) { diff --git a/integration/agentcompat/internal/workflowpolicy/execution.go b/integration/agentcompat/internal/workflowpolicy/execution.go index 58621a3a..f5a9d7ac 100644 --- a/integration/agentcompat/internal/workflowpolicy/execution.go +++ b/integration/agentcompat/internal/workflowpolicy/execution.go @@ -10,7 +10,6 @@ import ( ) var ( - fullCommitPattern = regexp.MustCompile(`^[0-9a-fA-F]{40}$`) dockerCommandPattern = regexp.MustCompile(`(?mi)(?:^|[;&|]\s*|\s)(?:(?:sudo|env)\s+)?(?:/[^\s]+/)?(?:docker|podman|nerdctl|containerd|buildah|runc|crictl)(?:\s|$)`) gitHubEnvironmentPattern = regexp.MustCompile(`(?is)GIT_[A-Za-z0-9_]*.*GITHUB_ENV|GITHUB_ENV.*GIT_[A-Za-z0-9_]*`) swallowedFailurePattern = regexp.MustCompile(`(?mi)(?:\|\|\s*(?:true|:|echo\b|printf\b|exit\s+0\b))|(?:^|[;&]\s*)set\s+\+(?:e|o\s+errexit)(?:\s|;|$)|(?:^|[;&]\s*)if\s+|(?:\b(?:bash|sh)\s+-c\b)|(?:^|[;&]\s*)trap\b[^\n]*\bexit\s+0\b|(?:[;&]\s*)(?:true|:)\s*(?:;|$)`) @@ -56,7 +55,6 @@ func (c *checker) checkJob(name string, job *yaml.Node) { func (c *checker) checkSteps(jobPath string, steps *yaml.Node) { redactionReady := false - validatedResolvers := make(map[string]Repository) for index, step := range steps.Content { path := jobPath + ".steps[" + strconv.Itoa(index) + "]" if step.Kind != yaml.MappingNode { @@ -78,15 +76,11 @@ func (c *checker) checkSteps(jobPath string, steps *yaml.Node) { c.reject(RuleWorkflowStructure, at(path+".run", run), "step run must be a scalar shell command") } c.checkContinueOnError(step, path) - resolver, validResolver := validatedRefResolver(step) if hasRun { - c.checkRun(path+".run", run, validResolver) - } - if validResolver { - validatedResolvers[resolver.id] = resolver.repository + c.checkRun(path+".run", run) } if hasUses { - c.checkUses(path, step, stepCheckState{redactionComplete: redactionReady, validatedResolvers: validatedResolvers}) + c.checkUses(path, step, stepCheckState{redactionComplete: redactionReady}) redactionReady = false continue } @@ -101,7 +95,7 @@ func (c *checker) checkContinueOnError(mapping *yaml.Node, path string) { } } -func (c *checker) checkRun(path string, run *yaml.Node, validatedResolver bool) { +func (c *checker) checkRun(path string, run *yaml.Node) { command, exists := scalarString(run) if !exists { return @@ -121,20 +115,19 @@ func (c *checker) checkRun(path string, run *yaml.Node, validatedResolver bool) if gitConfigurationPattern.MatchString(command) { c.reject(RuleRepositoryNotLiteral, at(path, run), "Git configuration mutation is forbidden") } - if gitRepositoryCommand.MatchString(command) && !validatedResolver { + if gitRepositoryCommand.MatchString(command) { rule := RuleRepositoryNotLiteral - detail := fmt.Sprintf("git repository operation %q is allowed only in the validated resolver", strings.TrimSpace(command)) + detail := fmt.Sprintf("git repository operation %q is forbidden", strings.TrimSpace(command)) if !strings.Contains(command, "$") { rule = RuleRepositoryNotAllowed - detail = fmt.Sprintf("repository operation %q is forbidden outside the validated resolver", strings.TrimSpace(command)) + detail = fmt.Sprintf("repository operation %q is forbidden", strings.TrimSpace(command)) } c.reject(rule, at(path, run), detail) } } type stepCheckState struct { - redactionComplete bool - validatedResolvers map[string]Repository + redactionComplete bool } func (c *checker) checkUses(path string, step *yaml.Node, state stepCheckState) { @@ -159,28 +152,23 @@ func (c *checker) checkUses(path string, step *yaml.Node, state stepCheckState) c.reject(RuleReusableExecutable, at(path+".uses", uses), "local action reuse from the workspace is forbidden") return } - actionRepository, _, found := strings.Cut(lowerAction, "@") + actionRepository, valid := actionRepository(lowerAction) + if !valid { + c.reject(RuleWorkflowStructure, at(path+".uses", uses), "action reference must use owner/repository@ref syntax") + return + } switch actionRepository { case "actions/cache", "actions/cache/restore", "actions/cache/save", "actions/download-artifact": c.reject(RuleReusableExecutable, at(path+".uses", uses), fmt.Sprintf("cache or artifact reuse action %q is forbidden", actionRepository)) return } - if !found { - c.reject(RuleOtherRepositoryRef, at(path+".uses", uses), "action must use its approved immutable SHA") - return - } - approvedRepository, _, pinned := approvedAction(action) - if approvedRepository == "" { + if !approvedAction(actionRepository) { c.reject(RuleRepositoryNotAllowed, at(path+".uses", uses), "action repository is not approved") return } - if !pinned { - c.reject(RuleOtherRepositoryRef, at(path+".uses", uses), "action must use its approved immutable SHA") - return - } - switch approvedRepository { + switch actionRepository { case "actions/checkout": - c.checkCheckout(path, step, state.validatedResolvers) + c.checkCheckout(path, step) case "actions/setup-go": c.checkRequiredCacheDisabled(path, step) case "actions/upload-artifact": @@ -189,19 +177,23 @@ func (c *checker) checkUses(path string, step *yaml.Node, state stepCheckState) c.checkCacheInputs(path, step) } -func approvedAction(action string) (string, string, bool) { - repository, ref, found := strings.Cut(strings.ToLower(action), "@") - if !found { - return repository, "", false +func actionRepository(action string) (string, bool) { + repository, ref, found := strings.Cut(action, "@") + if !found || repository == "" || ref == "" || strings.Contains(ref, "@") || strings.ContainsAny(action, " \t\r\n") { + return "", false } - approvedRefs := map[string]string{ - "actions/checkout": "9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0", - "actions/setup-go": "924ae3a1cded613372ab5595356fb5720e22ba16", - "actions/upload-artifact": "b7c566a772e6b6bfb58ed0dc250532a479d7789f", + owner, name, found := strings.Cut(repository, "/") + if !found || owner == "" || name == "" || strings.Contains(name, "/") { + return "", false + } + return repository, true +} + +func approvedAction(repository string) bool { + switch repository { + case "actions/checkout", "actions/setup-go", "actions/upload-artifact": + return true + default: + return false } - approvedRef, approved := approvedRefs[repository] - if !approved { - return "", ref, false - } - return repository, ref, ref == approvedRef } diff --git a/integration/agentcompat/internal/workflowpolicy/execution_test.go b/integration/agentcompat/internal/workflowpolicy/execution_test.go index 89416031..450d997d 100644 --- a/integration/agentcompat/internal/workflowpolicy/execution_test.go +++ b/integration/agentcompat/internal/workflowpolicy/execution_test.go @@ -73,8 +73,31 @@ func TestPolicy_RejectsUnapprovedAction(t *testing.T) { assertFixtureRejected(t, rejected("unapproved-action.yml", workflowpolicy.RuleRepositoryNotAllowed, "action")) } -func TestPolicy_RejectsMutableActionRef(t *testing.T) { - assertFixtureRejected(t, rejected("mutable-action-ref.yml", workflowpolicy.RuleOtherRepositoryRef, "approved immutable SHA")) +func TestPolicy_RejectsActionReferenceWithoutRef(t *testing.T) { + assertFixtureRejected(t, rejected("action-reference-missing-ref.yml", workflowpolicy.RuleWorkflowStructure, "owner/repository@ref")) +} + +func TestPolicy_RejectsMalformedActionReferences(t *testing.T) { + tests := []struct { + name string + uses string + rule workflowpolicy.Rule + diagnostic string + }{ + {name: "empty ref", uses: "actions/checkout@", rule: workflowpolicy.RuleWorkflowStructure, diagnostic: "owner/repository@ref"}, + {name: "duplicate separator", uses: "actions/checkout@v7@unexpected", rule: workflowpolicy.RuleWorkflowStructure, diagnostic: "owner/repository@ref"}, + {name: "extra action path", uses: "actions/checkout/extra@v7", rule: workflowpolicy.RuleWorkflowStructure, diagnostic: "owner/repository@ref"}, + {name: "whitespace", uses: "actions/checkout @v7", rule: workflowpolicy.RuleWorkflowStructure, diagnostic: "owner/repository@ref"}, + {name: "dynamic ref", uses: "actions/checkout@${{ inputs.ref }}", rule: workflowpolicy.RuleRepositoryNotLiteral, diagnostic: "action reference must be literal"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + workflow := "on:\n pull_request:\nconcurrency: policy\npermissions:\n contents: read\njobs:\n verify:\n runs-on: ubuntu-24.04\n timeout-minutes: 10\n steps:\n - uses: \"" + test.uses + "\"\n with:\n persist-credentials: false\n" + err := workflowpolicy.Verify([]byte(workflow), workflowpolicy.RepositoryNezha) + requireTypedPolicyError(t, err, test.rule) + require.ErrorContains(t, err, test.diagnostic) + }) + } } func TestPolicy_RejectsReusableWorkflowJob(t *testing.T) { diff --git a/integration/agentcompat/internal/workflowpolicy/policy_test.go b/integration/agentcompat/internal/workflowpolicy/policy_test.go index 1cc7ff4d..22b8c5f0 100644 --- a/integration/agentcompat/internal/workflowpolicy/policy_test.go +++ b/integration/agentcompat/internal/workflowpolicy/policy_test.go @@ -33,15 +33,42 @@ func TestPolicy_AcceptsSecureAgentWorkflow(t *testing.T) { require.NoError(t, err) } -func TestPolicy_AcceptsValidatedResolvedOtherRepositoryRef(t *testing.T) { - // Given - path := fixturePath(t, "secure-resolved-ref.yml") +func TestPolicy_AcceptsCrossRepositoryCheckoutRefs(t *testing.T) { + tests := []struct { + name string + fixture string + repository workflowpolicy.Repository + }{ + {name: "default branch", fixture: "cross-repository-default-ref.yml", repository: workflowpolicy.RepositoryNezha}, + {name: "branch", fixture: "secure-agent.yml", repository: workflowpolicy.RepositoryAgent}, + {name: "version tag", fixture: "secure-nezha.yml", repository: workflowpolicy.RepositoryNezha}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + err := workflowpolicy.VerifyFile(fixturePath(t, test.fixture), test.repository) + require.NoError(t, err) + }) + } +} - // When - err := workflowpolicy.VerifyFile(path, workflowpolicy.RepositoryNezha) - - // Then - require.NoError(t, err) +func TestPolicy_RejectsInvalidCrossRepositoryCheckoutRefs(t *testing.T) { + tests := []struct { + name string + ref string + }{ + {name: "dynamic", ref: "${{ inputs.ref }}"}, + {name: "non-string", ref: "false"}, + {name: "empty", ref: "\"\""}, + {name: "whitespace only", ref: "\" \""}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + workflow := "on:\n pull_request:\nconcurrency: policy\npermissions:\n contents: read\njobs:\n verify:\n runs-on: ubuntu-24.04\n timeout-minutes: 10\n steps:\n - uses: actions/checkout@v7.0.1\n with:\n repository: nezhahq/agent\n ref: " + test.ref + "\n persist-credentials: false\n" + err := workflowpolicy.Verify([]byte(workflow), workflowpolicy.RepositoryNezha) + requireTypedPolicyError(t, err, workflowpolicy.RuleRepositoryNotLiteral) + require.ErrorContains(t, err, "checkout ref must be a nonempty literal") + }) + } } func TestPolicy_RejectsPullRequestTarget(t *testing.T) { @@ -104,18 +131,6 @@ func TestPolicy_RejectsNonliteralRepository(t *testing.T) { assertFixtureRejected(t, rejected("nonliteral-repository.yml", workflowpolicy.RuleRepositoryNotLiteral, "literal")) } -func TestPolicy_RejectsMutableOtherRepositoryRef(t *testing.T) { - assertFixtureRejected(t, rejected("mutable-other-repository-ref.yml", workflowpolicy.RuleOtherRepositoryRef, "40")) -} - -func TestPolicy_RejectsUnvalidatedResolvedOtherRepositoryRef(t *testing.T) { - assertFixtureRejected(t, rejectionExpectation{fixture: "unvalidated-resolved-ref.yml", repository: workflowpolicy.RepositoryNezha, rule: workflowpolicy.RuleOtherRepositoryRef, diagnostic: "validated resolver"}) -} - -func TestPolicy_RejectsResolverVariableOverride(t *testing.T) { - assertFixtureRejected(t, rejectionExpectation{fixture: "resolver-variable-override.yml", repository: workflowpolicy.RepositoryNezha, rule: workflowpolicy.RuleOtherRepositoryRef, diagnostic: "validated resolver"}) -} - func TestPolicy_RejectsMissingPersistCredentialsFalse(t *testing.T) { assertFixtureRejected(t, rejected("missing-persist-credentials.yml", workflowpolicy.RulePersistCredentials, "persist-credentials")) } diff --git a/integration/agentcompat/internal/workflowpolicy/quality_workflow_test.go b/integration/agentcompat/internal/workflowpolicy/quality_workflow_test.go index 99e7693e..8192e58a 100644 --- a/integration/agentcompat/internal/workflowpolicy/quality_workflow_test.go +++ b/integration/agentcompat/internal/workflowpolicy/quality_workflow_test.go @@ -164,11 +164,11 @@ func requireCheckoutAndSetupGo(t *testing.T, steps []qualityStep) { t.Helper() require.GreaterOrEqual(t, len(steps), 2) checkout := steps[0] - require.Equal(t, "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0", checkout.Uses) + require.Equal(t, "actions/checkout@v7.0.1", checkout.Uses) require.NotNil(t, checkout.With.PersistCredentials) require.False(t, *checkout.With.PersistCredentials) setupGo := steps[1] - require.Equal(t, "actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16", setupGo.Uses) + require.Equal(t, "actions/setup-go@v7", setupGo.Uses) require.Equal(t, "1.26.x", setupGo.With.GoVersion) require.NotNil(t, setupGo.With.Cache) require.False(t, *setupGo.With.Cache) diff --git a/integration/agentcompat/internal/workflowpolicy/resolver.go b/integration/agentcompat/internal/workflowpolicy/resolver.go deleted file mode 100644 index 978e1027..00000000 --- a/integration/agentcompat/internal/workflowpolicy/resolver.go +++ /dev/null @@ -1,62 +0,0 @@ -package workflowpolicy - -import ( - "regexp" - "strings" - - "gopkg.in/yaml.v3" -) - -var ( - resolvedRefPattern = regexp.MustCompile(`^\$\{\{\s*steps\.([A-Za-z0-9_-]+)\.outputs\.sha\s*\}\}$`) - resolverRemotePattern = regexp.MustCompile(`(?m)^\s*remote=['"]https://github\.com/(nezhahq/(?:agent|nezha))\.git['"]\s*$`) -) - -type refResolver struct { - id string - repository Repository -} - -func validatedRefResolver(step *yaml.Node) (refResolver, bool) { - idNode, hasID := mappingValue(step, "id") - runNode, hasRun := mappingValue(step, "run") - id, literalID := scalarString(idNode) - command, literalRun := scalarString(runNode) - if !hasID || !hasRun || !literalID || !literalRun || strings.TrimSpace(id) == "" { - return refResolver{}, false - } - lines := make([]string, 0, 7) - for _, line := range strings.Split(command, "\n") { - trimmed := strings.TrimSpace(line) - if trimmed != "" { - lines = append(lines, trimmed) - } - } - if len(lines) != 7 || lines[0] != "set -euo pipefail" { - return refResolver{}, false - } - remoteMatch := resolverRemotePattern.FindStringSubmatch(lines[1]) - if len(remoteMatch) != 2 { - return refResolver{}, false - } - repository := Repository(remoteMatch[1]) - branch := "main" - if repository == RepositoryNezha { - branch = "master" - } - expectedLines := []string{ - lines[0], - lines[1], - "mapfile -t refs < <(git ls-remote \"$remote\" refs/heads/" + branch + ")", - "(( ${#refs[@]} == 1 ))", - "sha=${refs[0]%%$'\\t'*}", - `[[ "$sha" =~ ^[0-9a-f]{40}$ ]]`, - `printf 'sha=%s\n' "$sha" >> "$GITHUB_OUTPUT"`, - } - for index, expected := range expectedLines { - if lines[index] != expected { - return refResolver{}, false - } - } - return refResolver{id: id, repository: repository}, true -} diff --git a/integration/agentcompat/internal/workflowpolicy/stress_workflow_test.go b/integration/agentcompat/internal/workflowpolicy/stress_workflow_test.go index c1f774c2..6d28c0d9 100644 --- a/integration/agentcompat/internal/workflowpolicy/stress_workflow_test.go +++ b/integration/agentcompat/internal/workflowpolicy/stress_workflow_test.go @@ -10,7 +10,7 @@ import ( const agentcompatStressTestName = "TestStressPRFullEightAgentExactlyOnce" -func TestPolicy_NezhaStressWorkflowRunsPinnedCrossRepositoryTest(t *testing.T) { +func TestPolicy_NezhaStressWorkflowRunsCrossRepositoryTest(t *testing.T) { // Given data := readNezhaQualityWorkflow(t) var workflow qualityWorkflow @@ -27,7 +27,7 @@ func TestPolicy_NezhaStressWorkflowRunsPinnedCrossRepositoryTest(t *testing.T) { require.Len(t, stressJob.Steps, 6) require.Equal(t, []string{ "Checkout Nezha revision", - "Checkout pinned Agent revision", + "Checkout Agent repository", "Set up Go", "Prepare Dashboard build inputs", "Require named stress test", @@ -42,21 +42,21 @@ func TestPolicy_NezhaStressWorkflowRunsPinnedCrossRepositoryTest(t *testing.T) { }) nezhaCheckout := stressJob.stepNamed(t, "Checkout Nezha revision") - require.Equal(t, "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0", nezhaCheckout.Uses) + require.Equal(t, "actions/checkout@v7.0.1", nezhaCheckout.Uses) require.Empty(t, nezhaCheckout.With.Repository) require.Empty(t, nezhaCheckout.With.Ref) require.Equal(t, "nezha", nezhaCheckout.With.Path) require.False(t, *nezhaCheckout.With.PersistCredentials) - agentCheckout := stressJob.stepNamed(t, "Checkout pinned Agent revision") - require.Equal(t, "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0", agentCheckout.Uses) + agentCheckout := stressJob.stepNamed(t, "Checkout Agent repository") + require.Equal(t, "actions/checkout@v7.0.1", agentCheckout.Uses) require.Equal(t, "nezhahq/agent", agentCheckout.With.Repository) - require.Equal(t, "667e1dd5e166ffef808ec26dc20de85bc33a0a0f", agentCheckout.With.Ref) + require.Empty(t, agentCheckout.With.Ref) require.Equal(t, "agent", agentCheckout.With.Path) require.False(t, *agentCheckout.With.PersistCredentials) setupGo := stressJob.stepNamed(t, "Set up Go") - require.Equal(t, "actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16", setupGo.Uses) + require.Equal(t, "actions/setup-go@v7", setupGo.Uses) require.Equal(t, "1.26.x", setupGo.With.GoVersion) require.False(t, *setupGo.With.Cache) diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/mutable-action-ref.yml b/integration/agentcompat/internal/workflowpolicy/testdata/action-reference-missing-ref.yml similarity index 85% rename from integration/agentcompat/internal/workflowpolicy/testdata/mutable-action-ref.yml rename to integration/agentcompat/internal/workflowpolicy/testdata/action-reference-missing-ref.yml index e7bae2a0..4848510b 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/mutable-action-ref.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/action-reference-missing-ref.yml @@ -8,6 +8,6 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@v7 + - uses: actions/checkout with: persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/artifact-without-redaction.yml b/integration/agentcompat/internal/workflowpolicy/testdata/artifact-without-redaction.yml index a62bf39c..d0c97ada 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/artifact-without-redaction.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/artifact-without-redaction.yml @@ -9,6 +9,6 @@ jobs: timeout-minutes: 10 steps: - run: go test ./... - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 with: path: ${{ runner.temp }}/results diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/cache.yml b/integration/agentcompat/internal/workflowpolicy/testdata/cache.yml index e495b094..d9e9a926 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/cache.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/cache.yml @@ -8,7 +8,7 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/cache@0123456789abcdef0123456789abcdef01234567 + - uses: actions/cache@v4 with: path: bin key: executable-cache diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/conditional-redaction.yml b/integration/agentcompat/internal/workflowpolicy/testdata/conditional-redaction.yml index 9cc60999..58ea4e2c 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/conditional-redaction.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/conditional-redaction.yml @@ -12,7 +12,7 @@ jobs: id: redact-evidence if: false run: go run ./integration/agentcompat/cmd/redact --output "$RUNNER_TEMP/nezha-agentcompat-redacted" - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 if: always() with: path: ${{ runner.temp }}/nezha-agentcompat-redacted diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/mutable-other-repository-ref.yml b/integration/agentcompat/internal/workflowpolicy/testdata/cross-repository-default-ref.yml similarity index 71% rename from integration/agentcompat/internal/workflowpolicy/testdata/mutable-other-repository-ref.yml rename to integration/agentcompat/internal/workflowpolicy/testdata/cross-repository-default-ref.yml index f8ac0fe8..f7bbcad7 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/mutable-other-repository-ref.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/cross-repository-default-ref.yml @@ -8,8 +8,7 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: repository: nezhahq/agent - ref: main persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/download-artifact.yml b/integration/agentcompat/internal/workflowpolicy/testdata/download-artifact.yml index 5437be91..f4b7e095 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/download-artifact.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/download-artifact.yml @@ -8,4 +8,4 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/download-artifact@0123456789abcdef0123456789abcdef01234567 + - uses: actions/download-artifact@v7 diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/missing-persist-credentials.yml b/integration/agentcompat/internal/workflowpolicy/testdata/missing-persist-credentials.yml index fd2712e5..f6e6ef84 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/missing-persist-credentials.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/missing-persist-credentials.yml @@ -8,4 +8,4 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/missing-required-dependency.yml b/integration/agentcompat/internal/workflowpolicy/testdata/missing-required-dependency.yml index 093d71b5..4d784f95 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/missing-required-dependency.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/missing-required-dependency.yml @@ -13,10 +13,10 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 30 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: persist-credentials: false - - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 + - uses: actions/setup-go@v7 with: go-version: "1.26.x" cache: false @@ -25,10 +25,10 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 45 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: persist-credentials: false - - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 + - uses: actions/setup-go@v7 with: go-version: "1.26.x" cache: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/mixed-artifact-paths.yml b/integration/agentcompat/internal/workflowpolicy/testdata/mixed-artifact-paths.yml index 5c9c440c..3149035f 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/mixed-artifact-paths.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/mixed-artifact-paths.yml @@ -10,7 +10,7 @@ jobs: steps: - name: Redact evidence run: go run ./integration/agentcompat/cmd/redact - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 with: path: | ${{ runner.temp }}/redacted-results diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/no-op-redaction.yml b/integration/agentcompat/internal/workflowpolicy/testdata/no-op-redaction.yml index 825a177f..e7327a06 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/no-op-redaction.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/no-op-redaction.yml @@ -10,6 +10,6 @@ jobs: steps: - name: Redact evidence run: "true" - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 with: path: ${{ runner.temp }}/redacted-results diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/nonliteral-repository.yml b/integration/agentcompat/internal/workflowpolicy/testdata/nonliteral-repository.yml index 113856a5..de4ddbd3 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/nonliteral-repository.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/nonliteral-repository.yml @@ -8,8 +8,8 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: repository: ${{ inputs.repository }} - ref: 0123456789abcdef0123456789abcdef01234567 + ref: main persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/nonmapping-uses.yml b/integration/agentcompat/internal/workflowpolicy/testdata/nonmapping-uses.yml index f9bac2e6..873a7f7f 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/nonmapping-uses.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/nonmapping-uses.yml @@ -9,4 +9,4 @@ jobs: timeout-minutes: 10 steps: - uses: - - actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - actions/checkout@v7.0.1 diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/numeric-timeout.yml b/integration/agentcompat/internal/workflowpolicy/testdata/numeric-timeout.yml index 54d06888..83080d3b 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/numeric-timeout.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/numeric-timeout.yml @@ -8,6 +8,6 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10.5 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/persist-credentials-true.yml b/integration/agentcompat/internal/workflowpolicy/testdata/persist-credentials-true.yml index eee6ed9f..7ec52505 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/persist-credentials-true.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/persist-credentials-true.yml @@ -8,6 +8,6 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: persist-credentials: true diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/quoted-false-security-controls.yml b/integration/agentcompat/internal/workflowpolicy/testdata/quoted-false-security-controls.yml index 4d81da3b..b8a81955 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/quoted-false-security-controls.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/quoted-false-security-controls.yml @@ -8,6 +8,6 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: persist-credentials: "false" diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/raw-after-redaction.yml b/integration/agentcompat/internal/workflowpolicy/testdata/raw-after-redaction.yml index a3c84295..6a21836b 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/raw-after-redaction.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/raw-after-redaction.yml @@ -13,7 +13,7 @@ jobs: if: always() run: go run ./integration/agentcompat/cmd/redact --output "$RUNNER_TEMP/nezha-agentcompat-redacted" - run: cp raw-secret "$RUNNER_TEMP/nezha-agentcompat-redacted/raw-secret" - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 if: always() with: path: ${{ runner.temp }}/nezha-agentcompat-redacted diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/redaction-command-append.yml b/integration/agentcompat/internal/workflowpolicy/testdata/redaction-command-append.yml index 68519f6d..81e0d7bf 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/redaction-command-append.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/redaction-command-append.yml @@ -14,7 +14,7 @@ jobs: run: | go run ./integration/agentcompat/cmd/redact --output "$RUNNER_TEMP/nezha-agentcompat-redacted" cp raw-secret "$RUNNER_TEMP/nezha-agentcompat-redacted/raw-secret" - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 if: always() with: path: ${{ runner.temp }}/nezha-agentcompat-redacted diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/resolver-variable-override.yml b/integration/agentcompat/internal/workflowpolicy/testdata/resolver-variable-override.yml deleted file mode 100644 index c6173ad1..00000000 --- a/integration/agentcompat/internal/workflowpolicy/testdata/resolver-variable-override.yml +++ /dev/null @@ -1,25 +0,0 @@ -on: - pull_request: -concurrency: policy -permissions: - contents: read -jobs: - verify: - runs-on: ubuntu-24.04 - timeout-minutes: 10 - steps: - - id: resolve-agent - run: | - set -euo pipefail - remote='https://github.com/nezhahq/agent.git' - mapfile -t refs < <(git ls-remote "$remote" refs/heads/main) - (( ${#refs[@]} == 1 )) - sha=${refs[0]%%$'\t'*} - [[ "$sha" =~ ^[0-9a-f]{40}$ ]] - sha=attacker-controlled - printf 'sha=%s\n' "$sha" >> "$GITHUB_OUTPUT" - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 - with: - repository: nezhahq/agent - ref: ${{ steps.resolve-agent.outputs.sha }} - persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/secure-agent.yml b/integration/agentcompat/internal/workflowpolicy/testdata/secure-agent.yml index ac94147f..079a4877 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/secure-agent.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/secure-agent.yml @@ -9,13 +9,13 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 30 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: persist-credentials: false - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: repository: nezhahq/nezha - ref: fedcba9876543210fedcba9876543210fedcba98 + ref: master path: nezha persist-credentials: false - run: go test ./integration/agentcompat/... diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/secure-nezha.yml b/integration/agentcompat/internal/workflowpolicy/testdata/secure-nezha.yml index a123d0c6..99ad07c3 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/secure-nezha.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/secure-nezha.yml @@ -15,14 +15,14 @@ jobs: timeout-minutes: 30 steps: - name: Checkout Nezha - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + uses: actions/checkout@v7.0.1 with: persist-credentials: false - name: Checkout Agent - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + uses: actions/checkout@v7.0.1 with: repository: nezhahq/agent - ref: 0123456789abcdef0123456789abcdef01234567 + ref: v1.2.3 path: agent persist-credentials: false - name: Test @@ -34,7 +34,7 @@ jobs: if: always() run: go run ./integration/agentcompat/cmd/redact --output "$RUNNER_TEMP/nezha-agentcompat-redacted" - name: Upload redacted evidence - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + uses: actions/upload-artifact@v6 if: always() with: path: ${{ runner.temp }}/nezha-agentcompat-redacted diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/secure-resolved-ref.yml b/integration/agentcompat/internal/workflowpolicy/testdata/secure-resolved-ref.yml deleted file mode 100644 index 92c7824c..00000000 --- a/integration/agentcompat/internal/workflowpolicy/testdata/secure-resolved-ref.yml +++ /dev/null @@ -1,27 +0,0 @@ -name: Secure resolved ref -on: - pull_request: -concurrency: resolved-${{ github.ref }} -permissions: - contents: read -jobs: - verify: - runs-on: ubuntu-24.04 - timeout-minutes: 30 - steps: - - id: resolve-agent - name: Resolve Agent commit - shell: bash - run: | - set -euo pipefail - remote='https://github.com/nezhahq/agent.git' - mapfile -t refs < <(git ls-remote "$remote" refs/heads/main) - (( ${#refs[@]} == 1 )) - sha=${refs[0]%%$'\t'*} - [[ "$sha" =~ ^[0-9a-f]{40}$ ]] - printf 'sha=%s\n' "$sha" >> "$GITHUB_OUTPUT" - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 - with: - repository: nezhahq/agent - ref: ${{ steps.resolve-agent.outputs.sha }} - persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/setup-go-default-cache.yml b/integration/agentcompat/internal/workflowpolicy/testdata/setup-go-default-cache.yml index 5e220263..5b2cf1c2 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/setup-go-default-cache.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/setup-go-default-cache.yml @@ -8,6 +8,6 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 + - uses: actions/setup-go@v7 with: go-version: 1.26.3 diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-action.yml b/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-action.yml index 258a32ac..33790dec 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-action.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-action.yml @@ -8,4 +8,4 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: attacker/exfiltrate@0123456789abcdef0123456789abcdef01234567 + - uses: attacker/exfiltrate@v1 diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-repository.yml b/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-repository.yml index 082668ee..88651cf6 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-repository.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/unapproved-repository.yml @@ -8,8 +8,8 @@ jobs: runs-on: ubuntu-24.04 timeout-minutes: 10 steps: - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 + - uses: actions/checkout@v7.0.1 with: repository: attacker/fork - ref: 0123456789abcdef0123456789abcdef01234567 + ref: main persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/unredacted-artifact-path.yml b/integration/agentcompat/internal/workflowpolicy/testdata/unredacted-artifact-path.yml index aaffc06c..91b12f1d 100644 --- a/integration/agentcompat/internal/workflowpolicy/testdata/unredacted-artifact-path.yml +++ b/integration/agentcompat/internal/workflowpolicy/testdata/unredacted-artifact-path.yml @@ -11,6 +11,6 @@ jobs: - name: Redact evidence id: redact-evidence run: go run ./integration/agentcompat/cmd/redact - - uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f + - uses: actions/upload-artifact@v6 with: path: ${{ runner.temp }}/results diff --git a/integration/agentcompat/internal/workflowpolicy/testdata/unvalidated-resolved-ref.yml b/integration/agentcompat/internal/workflowpolicy/testdata/unvalidated-resolved-ref.yml deleted file mode 100644 index abd2d7ae..00000000 --- a/integration/agentcompat/internal/workflowpolicy/testdata/unvalidated-resolved-ref.yml +++ /dev/null @@ -1,18 +0,0 @@ -on: - pull_request: -concurrency: policy -permissions: - contents: read -jobs: - verify: - runs-on: ubuntu-24.04 - timeout-minutes: 10 - steps: - - id: resolve-agent - run: | - echo "sha=$(git ls-remote https://github.com/nezhahq/agent.git refs/heads/main | cut -f1)" >> "$GITHUB_OUTPUT" - - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 - with: - repository: nezhahq/agent - ref: ${{ steps.resolve-agent.outputs.sha }} - persist-credentials: false diff --git a/integration/agentcompat/internal/workflowpolicy/types.go b/integration/agentcompat/internal/workflowpolicy/types.go index 75ca1aa5..f9564971 100644 --- a/integration/agentcompat/internal/workflowpolicy/types.go +++ b/integration/agentcompat/internal/workflowpolicy/types.go @@ -24,7 +24,6 @@ const ( RuleContainerizedExecution Rule = "containerized-execution" RuleRepositoryNotAllowed Rule = "repository-not-allowed" RuleRepositoryNotLiteral Rule = "repository-not-literal" - RuleOtherRepositoryRef Rule = "other-repository-ref" RulePersistCredentials Rule = "persist-credentials" // #nosec G101 -- GitHub Actions configuration key, not a credential. RuleReusableExecutable Rule = "reusable-executable" RuleContinueOnError Rule = "continue-on-error" diff --git a/service/rpc/nezha.go b/service/rpc/nezha.go index 5ad70c67..e332aa74 100644 --- a/service/rpc/nezha.go +++ b/service/rpc/nezha.go @@ -20,6 +20,10 @@ var _ pb.NezhaServiceServer = (*NezhaHandler)(nil) var NezhaHandlerSingleton *NezhaHandler +// ErrRequestTaskStreamSuperseded is returned when a RequestTask result arrives +// after its stream is no longer the live stream for the authenticated server. +var ErrRequestTaskStreamSuperseded = errors.New("request task stream superseded") + type NezhaHandler struct { Auth *authHandler ioStreams map[string]*ioStreamContext @@ -79,6 +83,18 @@ func clearRequestTaskStream(clientID uint64, captured *model.Server, stream pb.N captured.ClearTaskStreamIfCurrent(stream) } +// currentRequestTaskServer authorizes a received result against the live +// ServerShared entry. Server pointer replacement is valid when it inherited +// the same task stream holder; only a missing entry or different stream makes +// a received result stale. +func currentRequestTaskServer(clientID uint64, stream pb.NezhaService_RequestTaskServer) (*model.Server, error) { + current, ok := singleton.ServerShared.Get(clientID) + if !ok || current == nil || current.GetTaskStream() != stream { + return nil, ErrRequestTaskStreamSuperseded + } + return current, nil +} + func (s *NezhaHandler) RequestTask(stream pb.NezhaService_RequestTaskServer) error { var clientID uint64 var err error @@ -106,6 +122,10 @@ func (s *NezhaHandler) RequestTask(stream pb.NezhaService_RequestTaskServer) err log.Printf("NEZHA>> RequestTask error: %v, clientID: %d\n", err, clientID) return err } + server, err = currentRequestTaskServer(clientID, stream) + if err != nil { + return err + } switch result.GetType() { case model.TaskTypeCommand: // 处理上报的计划任务 diff --git a/service/rpc/request_task_security_test.go b/service/rpc/request_task_security_test.go index 3a86c4e2..3b57016d 100644 --- a/service/rpc/request_task_security_test.go +++ b/service/rpc/request_task_security_test.go @@ -17,11 +17,12 @@ import ( ) type requestTaskSecurityStream struct { - ctx context.Context - results []*pb.TaskResult - onRecv func() - onSend func(*pb.Task) - sendErr error + ctx context.Context + results []*pb.TaskResult + onRecv func() + onResult func() + onSend func(*pb.Task) + sendErr error } func (s *requestTaskSecurityStream) Send(task *pb.Task) error { @@ -40,6 +41,11 @@ func (s *requestTaskSecurityStream) Recv() (*pb.TaskResult, error) { } result := s.results[0] s.results = s.results[1:] + if s.onResult != nil { + onResult := s.onResult + s.onResult = nil + onResult() + } return result, nil } diff --git a/service/rpc/request_task_stale_stream_test.go b/service/rpc/request_task_stale_stream_test.go index 8924c3d9..95885255 100644 --- a/service/rpc/request_task_stale_stream_test.go +++ b/service/rpc/request_task_stale_stream_test.go @@ -6,6 +6,7 @@ import ( "testing" "github.com/nezhahq/nezha/model" + pb "github.com/nezhahq/nezha/proto" "github.com/nezhahq/nezha/service/singleton" ) @@ -44,3 +45,91 @@ func TestRequestTaskCleanupDetachesStreamFromCurrentServerAfterEdit(t *testing.T t.Fatalf("edited server must report offline after the agent stream dropped, got %T", got) } } + +func TestRequestTaskRejectsResultWhenServerDeletedAfterRecv(t *testing.T) { + reporter := requestTaskSecurityServer(7, 200, "10101010-1010-1010-1010-101010101010") + cronTask := requestTaskSecurityCron(42, 200, model.CronCoverAll, nil) + setupRequestTaskSecurityFixture(t, []*model.Server{reporter}, []*model.Cron{cronTask}, map[uint64]model.UserInfo{ + 200: {Role: model.RoleMember}, + }, map[string]uint64{"reporter-secret": 200}) + + stream := requestTaskSecurityAuthedStream("reporter-secret", reporter.UUID) + stream.results = []*pb.TaskResult{cronTaskResult(cronTask.ID, true)} + stream.onResult = func() { + singleton.ServerShared.Delete([]uint64{reporter.ID}) + } + + err := NewNezhaHandler().RequestTask(stream) + if !errors.Is(err, ErrRequestTaskStreamSuperseded) { + t.Fatalf("expected stale RequestTask stream error, got %v", err) + } + assertCronResultNotUpdated(t, cronTask.ID) +} + +func TestRequestTaskRejectsResultWhenNewerStreamSupersedesOld(t *testing.T) { + reporter := requestTaskSecurityServer(7, 200, "20202020-2020-2020-2020-202020202020") + cronTask := requestTaskSecurityCron(42, 200, model.CronCoverAll, nil) + setupRequestTaskSecurityFixture(t, []*model.Server{reporter}, []*model.Cron{cronTask}, map[uint64]model.UserInfo{ + 200: {Role: model.RoleMember}, + }, map[string]uint64{"reporter-secret": 200}) + + current, ok := singleton.ServerShared.Get(reporter.ID) + if !ok { + t.Fatalf("server %d not found", reporter.ID) + } + newer := &requestTaskSecurityStream{ctx: context.Background()} + stream := requestTaskSecurityAuthedStream("reporter-secret", reporter.UUID) + stream.results = []*pb.TaskResult{cronTaskResult(cronTask.ID, true)} + stream.onResult = func() { + current.SetTaskStream(newer) + } + + err := NewNezhaHandler().RequestTask(stream) + if !errors.Is(err, ErrRequestTaskStreamSuperseded) { + t.Fatalf("expected superseded RequestTask stream error, got %v", err) + } + if got := current.GetTaskStream(); got != newer { + t.Fatalf("old stream cleanup must preserve newer stream, got %T", got) + } + assertCronResultNotUpdated(t, cronTask.ID) +} + +func TestRequestTaskAcceptsResultAfterServerPointerReplacementWithSameStream(t *testing.T) { + reporter := requestTaskSecurityServer(7, 200, "30303030-3030-3030-3030-303030303030") + cronTask := requestTaskSecurityCron(42, 200, model.CronCoverAll, nil) + setupRequestTaskSecurityFixture(t, []*model.Server{reporter}, []*model.Cron{cronTask}, map[uint64]model.UserInfo{ + 200: {Role: model.RoleMember}, + }, map[string]uint64{"reporter-secret": 200}) + + old, ok := singleton.ServerShared.Get(reporter.ID) + if !ok { + t.Fatalf("server %d not found", reporter.ID) + } + stream := requestTaskSecurityAuthedStream("reporter-secret", reporter.UUID) + stream.results = []*pb.TaskResult{cronTaskResult(cronTask.ID, true)} + stream.onResult = func() { + replacement := &model.Server{Common: model.Common{ID: old.ID, UserID: old.UserID}, UUID: old.UUID, Name: "replacement"} + replacement.CopyFromRunningServer(old) + singleton.ServerShared.Update(replacement, "") + } + + err := NewNezhaHandler().RequestTask(stream) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected RequestTask to finish after accepted result, got %v", err) + } + if !cronLastResult(t, cronTask.ID) { + t.Fatal("result on a replacement server that inherited the stream must be accepted") + } +} + +func assertCronResultNotUpdated(t *testing.T, cronID uint64) { + t.Helper() + + var cronTask model.Cron + if err := singleton.DB.First(&cronTask, cronID).Error; err != nil { + t.Fatal(err) + } + if cronTask.LastResult || !cronTask.LastExecutedAt.IsZero() { + t.Fatalf("stale RequestTask result must not mutate cron, got last_result=%t last_executed_at=%s", cronTask.LastResult, cronTask.LastExecutedAt) + } +} diff --git a/service/singleton/server.go b/service/singleton/server.go index ea758ea8..a74a7f62 100644 --- a/service/singleton/server.go +++ b/service/singleton/server.go @@ -6,6 +6,7 @@ import ( "log" "slices" "strings" + "sync" "github.com/nezhahq/nezha/model" "github.com/nezhahq/nezha/pkg/ddns" @@ -15,6 +16,10 @@ import ( type ServerClass struct { class[uint64, *model.Server] + // lifecycleMu serializes changes to the authoritative server entries with + // synchronous ServiceSentinel report processing. + lifecycleMu sync.RWMutex + uuidToID map[string]uint64 sortedListForGuest []*model.Server @@ -71,7 +76,26 @@ func ownerIsAdmin(ownerUID uint64) bool { return userIsAdmin(ownerUID) } +func (c *ServerClass) lockLifecycleRead() { + c.lifecycleMu.RLock() +} + +func (c *ServerClass) unlockLifecycleRead() { + c.lifecycleMu.RUnlock() +} + +func (c *ServerClass) lockLifecycleWrite() { + c.lifecycleMu.Lock() +} + +func (c *ServerClass) unlockLifecycleWrite() { + c.lifecycleMu.Unlock() +} + func (c *ServerClass) Update(s *model.Server, uuid string) { + c.lockLifecycleWrite() + defer c.unlockLifecycleWrite() + c.listMu.Lock() c.list[s.ID] = s @@ -91,6 +115,9 @@ func (c *ServerClass) Update(s *model.Server, uuid string) { } func (c *ServerClass) Delete(idList []uint64) { + c.lockLifecycleWrite() + defer c.unlockLifecycleWrite() + c.listMu.Lock() for _, id := range idList { @@ -107,6 +134,17 @@ func (c *ServerClass) Delete(idList []uint64) { c.sortList() } +// setUserID updates in-memory ownership under the server lifecycle lock so a +// transfer cannot change authorization during synchronous report processing. +func (c *ServerClass) setUserID(id, userID uint64) { + c.lockLifecycleWrite() + defer c.unlockLifecycleWrite() + + if s, ok := c.Get(id); ok && s != nil { + s.SetUserID(userID) + } +} + func (c *ServerClass) GetSortedListForGuest() []*model.Server { c.sortedListMu.RLock() defer c.sortedListMu.RUnlock() diff --git a/service/singleton/server_transfer.go b/service/singleton/server_transfer.go index fe87aef8..cef41476 100644 --- a/service/singleton/server_transfer.go +++ b/service/singleton/server_transfer.go @@ -663,11 +663,9 @@ func (c *ServerTransferClass) Initiate(tx *gorm.DB, serverID, fromUserID, toUser // and admits the old AgentSecret on the happy "owner match" path — // bypassing the bounded pending-tolerance contract. func (c *ServerTransferClass) Register(t *model.ServerTransfer) { - if s, ok := ServerShared.Get(t.ServerID); ok && s != nil { - // SetUserID over atomic write — auth.go hot path concurrently - // reads this field; a plain assignment would be a data race. - s.SetUserID(t.ToUserID) - } + // SetUserID uses an atomic write because auth.go reads this hot-path field + // concurrently; ServerClass also serializes it with service reports. + ServerShared.setUserID(t.ServerID, t.ToUserID) c.mu.Lock() c.pending[t.ServerID] = t @@ -1163,9 +1161,7 @@ func (c *ServerTransferClass) revertTransition(transferID uint64, newStatus mode // no longer admits the destination user's global AgentSecret via // ServerShared.GetUserID() == userId on the happy "owner match" path. if transitionedByThisCall { - if s, ok := ServerShared.Get(t.ServerID); ok && s != nil { - s.SetUserID(t.FromUserID) - } + ServerShared.setUserID(t.ServerID, t.FromUserID) } // Self-heal: any non-Pending DB status invalidates the in-memory entry — diff --git a/service/singleton/servicesentinel.go b/service/singleton/servicesentinel.go index 2a98aab6..dd8882ff 100644 --- a/service/singleton/servicesentinel.go +++ b/service/singleton/servicesentinel.go @@ -394,7 +394,16 @@ func (ss *ServiceSentinel) Delete(ids []uint64) { delete(ss.serviceStatusToday, id) // 停掉定时任务 - CronShared.Remove(ss.services[id].CronJobID) + // GHSA-jx78-55p5-rwv5 (Finding 2): guard against a caller supplying an id + // that does not exist in the in-memory registry. CheckPermission returns + // vacuously true for unknown ids, so the controller layer cannot prevent + // this. Without the guard, ss.services[id] is nil and the .CronJobID + // field access panics, aborting the Delete loop before the remaining valid + // ids are cleaned from memory — their service records were already deleted + // from the database, producing zombie services. + if svc := ss.services[id]; svc != nil { + CronShared.Remove(svc.CronJobID) + } delete(ss.services, id) delete(ss.monthlyStatus, id) @@ -536,238 +545,252 @@ func (ss *ServiceSentinel) Close() { func (ss *ServiceSentinel) worker() { // 从服务状态汇报管道获取汇报的服务数据 for r := range ss.serviceReportChannel { - cs, _ := ss.Get(r.Data.GetId()) - reporter, _ := ServerShared.Get(r.Reporter) - // 入站结果必须匹配出站任务派发边界，避免 agent 伪造其他服务 ID 写入监控状态。 - if !canReportServiceResult(cs, reporter, r.Data.GetType()) { - log.Printf("NEZHA>> Incorrect service monitor report %+v", r) - continue - } - if ss.serviceReportValidatedHook != nil { - ss.serviceReportValidatedHook(r.Data.GetId()) - } - - mh := r.Data - // Serialize Delete and Update before this accepted report causes any side effect. - ss.serviceResponseDataStoreLock.Lock() - serviceStatusToday := ss.serviceStatusToday[mh.GetId()] - serviceCurrentStatusData := ss.serviceCurrentStatusData[mh.GetId()] - currentService, serviceExists := ss.Get(mh.GetId()) - if serviceStatusToday == nil || serviceCurrentStatusData == nil || !serviceExists || - !canReportServiceResult(currentService, reporter, mh.GetType()) { - ss.serviceResponseDataStoreLock.Unlock() - continue - } - cs = currentService - - if mh.Type == model.TaskTypeTCPPing || mh.Type == model.TaskTypeICMPPing { - // TCP/ICMP Ping 使用平均值计算后再写入 - serviceTcpMap, ok := ss.serviceResponsePing[mh.GetId()] - if !ok { - serviceTcpMap = make(map[uint64]*pingStore) - ss.serviceResponsePing[mh.GetId()] = serviceTcpMap - } - ts, ok := serviceTcpMap[r.Reporter] - if !ok { - ts = &pingStore{} - } - ts.count++ - ts.ping = (ts.ping*float64(ts.count-1) + float64(mh.Delay)) / float64(ts.count) - if mh.Successful { - ts.successCount++ - } - if ts.count == Conf.AvgPingCount { - if TSDBEnabled() { - if err := TSDBShared.WriteServiceMetrics(&tsdb.ServiceMetrics{ - ServiceID: mh.GetId(), - ServerID: r.Reporter, - Timestamp: time.Now(), - Delay: ts.ping, - Successful: ts.successCount*2 >= ts.count, - }); err != nil { - log.Printf("NEZHA>> Failed to save service monitor metrics to TSDB: %v", err) - } - } else { - if err := DB.Create(&model.ServiceHistory{ - ServiceID: mh.GetId(), - AvgDelay: ts.ping, - Data: mh.Data, - ServerID: r.Reporter, - }).Error; err != nil { - log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err) - } + serverShared := ServerShared + func() { + defer func() { + if recovered := recover(); recovered != nil { + log.Printf("NEZHA>> Service monitor report processing panicked: %v", recovered) } - ts.count = 0 - ts.ping = 0 - ts.successCount = 0 - } - serviceTcpMap[r.Reporter] = ts - } else { + }() + ss.processReport(r, serverShared) + }() + } +} + +func (ss *ServiceSentinel) processReport(r ReportData, serverShared *ServerClass) { + serverShared.lockLifecycleRead() + defer serverShared.unlockLifecycleRead() + + cs, _ := ss.Get(r.Data.GetId()) + reporter, _ := serverShared.Get(r.Reporter) + // 入站结果必须匹配出站任务派发边界，避免 agent 伪造其他服务 ID 写入监控状态。 + if !canReportServiceResult(cs, reporter, r.Data.GetType()) { + log.Printf("NEZHA>> Incorrect service monitor report %+v", r) + return + } + if ss.serviceReportValidatedHook != nil { + ss.serviceReportValidatedHook(r.Data.GetId()) + } + + mh := r.Data + m := serverShared.GetList() + // Serialize Delete and Update before this accepted report causes any side effect. + ss.serviceResponseDataStoreLock.Lock() + defer ss.serviceResponseDataStoreLock.Unlock() + serviceStatusToday := ss.serviceStatusToday[mh.GetId()] + serviceCurrentStatusData := ss.serviceCurrentStatusData[mh.GetId()] + currentService, serviceExists := ss.Get(mh.GetId()) + if serviceStatusToday == nil || serviceCurrentStatusData == nil || !serviceExists || + !canReportServiceResult(currentService, reporter, mh.GetType()) { + return + } + cs = currentService + + if mh.Type == model.TaskTypeTCPPing || mh.Type == model.TaskTypeICMPPing { + // TCP/ICMP Ping 使用平均值计算后再写入 + serviceTcpMap, ok := ss.serviceResponsePing[mh.GetId()] + if !ok { + serviceTcpMap = make(map[uint64]*pingStore) + ss.serviceResponsePing[mh.GetId()] = serviceTcpMap + } + ts, ok := serviceTcpMap[r.Reporter] + if !ok { + ts = &pingStore{} + } + ts.count++ + ts.ping = (ts.ping*float64(ts.count-1) + float64(mh.Delay)) / float64(ts.count) + if mh.Successful { + ts.successCount++ + } + if ts.count == Conf.AvgPingCount { if TSDBEnabled() { if err := TSDBShared.WriteServiceMetrics(&tsdb.ServiceMetrics{ ServiceID: mh.GetId(), ServerID: r.Reporter, Timestamp: time.Now(), - Delay: float64(mh.Delay), - Successful: mh.Successful, + Delay: ts.ping, + Successful: ts.successCount*2 >= ts.count, }); err != nil { log.Printf("NEZHA>> Failed to save service monitor metrics to TSDB: %v", err) } - } - } - - // 写入当天状态 - if mh.Successful { - serviceStatusToday.Delay = (serviceStatusToday.Delay*float64(serviceStatusToday.Up) + - float64(mh.Delay)) / float64(serviceStatusToday.Up+1) - serviceStatusToday.Up++ - } else { - serviceStatusToday.Down++ - } - - currentTime := time.Now() - if serviceCurrentStatusData.t.IsZero() { - serviceCurrentStatusData.t = currentTime - } - - // 写入当前数据 - if serviceCurrentStatusData.t.Before(currentTime) { - serviceCurrentStatusData.t = currentTime.Add(30 * time.Second) - serviceCurrentStatusData.result = append(serviceCurrentStatusData.result, mh) - } - - // 更新当前状态 - ss.serviceResponseDataStore[mh.GetId()] = serviceResponseData{} - - // 永远是最新的 30 个数据的状态 [01:00, 02:00, 03:00] -> [04:00, 02:00, 03: 00] - for _, cs := range serviceCurrentStatusData.result { - if cs.GetId() > 0 { - rd := ss.serviceResponseDataStore[mh.GetId()] - if cs.Successful { - rd.Up++ - rd.Delay = (rd.Delay*float64(rd.Up-1) + float64(cs.Delay)) / float64(rd.Up) - } else { - rd.Down++ - } - ss.serviceResponseDataStore[mh.GetId()] = rd - } - } - - // 计算在线率， - var stateCode uint8 - { - upPercent := uint64(0) - rd := ss.serviceResponseDataStore[mh.GetId()] - if rd.Down+rd.Up > 0 { - upPercent = rd.Up * 100 / (rd.Down + rd.Up) - } - stateCode = GetStatusCode(upPercent) - } - - if len(serviceCurrentStatusData.result) == _CurrentStatusSize { - serviceCurrentStatusData.t = currentTime - if !TSDBEnabled() { - rd := ss.serviceResponseDataStore[mh.GetId()] + } else { if err := DB.Create(&model.ServiceHistory{ ServiceID: mh.GetId(), - AvgDelay: rd.Delay, + AvgDelay: ts.ping, Data: mh.Data, - Up: rd.Up, - Down: rd.Down, + ServerID: r.Reporter, }).Error; err != nil { log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err) } } - serviceCurrentStatusData.result = serviceCurrentStatusData.result[:0] + ts.count = 0 + ts.ping = 0 + ts.successCount = 0 } - - m := ServerShared.GetList() - // 延迟报警 - if mh.Delay > 0 { - delayCheck(&r, m, cs, mh) - } - - // 状态变更报警+触发任务执行 - if stateCode == StatusDown || stateCode != serviceCurrentStatusData.lastStatus { - lastStatus := serviceCurrentStatusData.lastStatus - // 存储新的状态值 - serviceCurrentStatusData.lastStatus = stateCode - - notifyCheck(&r, m, cs, mh, lastStatus, stateCode) - } - - // TLS 证书报警 - if ss.serviceReportBeforeTLSSideEffectsHook != nil { - ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId()) - } - var errMsg string - if strings.HasPrefix(mh.Data, "SSL证书错误：") { - // i/o timeout、connection timeout、EOF 错误 - if !strings.HasSuffix(mh.Data, "timeout") && - !strings.HasSuffix(mh.Data, "EOF") && - !strings.HasSuffix(mh.Data, "timed out") { - errMsg = mh.Data - if cs.Notify { - muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), "network") - go NotificationShared.SendNotification(cs.NotificationGroupID, Localizer.Tf("[TLS] Fetch cert info failed, Reporter: %s, Error: %s", cs.Name, errMsg), muteLabel) - } + serviceTcpMap[r.Reporter] = ts + } else { + if TSDBEnabled() { + if err := TSDBShared.WriteServiceMetrics(&tsdb.ServiceMetrics{ + ServiceID: mh.GetId(), + ServerID: r.Reporter, + Timestamp: time.Now(), + Delay: float64(mh.Delay), + Successful: mh.Successful, + }); err != nil { + log.Printf("NEZHA>> Failed to save service monitor metrics to TSDB: %v", err) } - } else { - // 清除网络错误静音缓存 - NotificationShared.UnMuteNotification(cs.NotificationGroupID, NotificationMuteLabel.ServiceTLS(mh.GetId(), "network")) + } + } - var newCert = strings.Split(mh.Data, "|") - if len(newCert) > 1 { - enableNotify := cs.Notify + // 写入当天状态 + if mh.Successful { + serviceStatusToday.Delay = (serviceStatusToday.Delay*float64(serviceStatusToday.Up) + + float64(mh.Delay)) / float64(serviceStatusToday.Up+1) + serviceStatusToday.Up++ + } else { + serviceStatusToday.Down++ + } - // 首次获取证书信息时，缓存证书信息 - if ss.tlsCertCache[mh.GetId()] == "" { - ss.tlsCertCache[mh.GetId()] = mh.Data + currentTime := time.Now() + if serviceCurrentStatusData.t.IsZero() { + serviceCurrentStatusData.t = currentTime + } + + // 写入当前数据 + if serviceCurrentStatusData.t.Before(currentTime) { + serviceCurrentStatusData.t = currentTime.Add(30 * time.Second) + serviceCurrentStatusData.result = append(serviceCurrentStatusData.result, mh) + } + + // 更新当前状态 + ss.serviceResponseDataStore[mh.GetId()] = serviceResponseData{} + + // 永远是最新的 30 个数据的状态 [01:00, 02:00, 03:00] -> [04:00, 02:00, 03: 00] + for _, cs := range serviceCurrentStatusData.result { + if cs.GetId() > 0 { + rd := ss.serviceResponseDataStore[mh.GetId()] + if cs.Successful { + rd.Up++ + rd.Delay = (rd.Delay*float64(rd.Up-1) + float64(cs.Delay)) / float64(rd.Up) + } else { + rd.Down++ + } + ss.serviceResponseDataStore[mh.GetId()] = rd + } + } + + // 计算在线率， + var stateCode uint8 + { + upPercent := uint64(0) + rd := ss.serviceResponseDataStore[mh.GetId()] + if rd.Down+rd.Up > 0 { + upPercent = rd.Up * 100 / (rd.Down + rd.Up) + } + stateCode = GetStatusCode(upPercent) + } + + if len(serviceCurrentStatusData.result) == _CurrentStatusSize { + serviceCurrentStatusData.t = currentTime + if !TSDBEnabled() { + rd := ss.serviceResponseDataStore[mh.GetId()] + if err := DB.Create(&model.ServiceHistory{ + ServiceID: mh.GetId(), + AvgDelay: rd.Delay, + Data: mh.Data, + Up: rd.Up, + Down: rd.Down, + }).Error; err != nil { + log.Printf("NEZHA>> Failed to save service monitor metrics: %v", err) + } + } + serviceCurrentStatusData.result = serviceCurrentStatusData.result[:0] + } + + // 延迟报警 + if mh.Delay > 0 { + delayCheck(&r, m, cs, mh) + } + + // 状态变更报警+触发任务执行 + if stateCode == StatusDown || stateCode != serviceCurrentStatusData.lastStatus { + lastStatus := serviceCurrentStatusData.lastStatus + // 存储新的状态值 + serviceCurrentStatusData.lastStatus = stateCode + + notifyCheck(&r, m, cs, mh, lastStatus, stateCode) + } + + // TLS 证书报警 + if ss.serviceReportBeforeTLSSideEffectsHook != nil { + ss.serviceReportBeforeTLSSideEffectsHook(mh.GetId()) + } + var errMsg string + if strings.HasPrefix(mh.Data, "SSL证书错误：") { + // i/o timeout、connection timeout、EOF 错误 + if !strings.HasSuffix(mh.Data, "timeout") && + !strings.HasSuffix(mh.Data, "EOF") && + !strings.HasSuffix(mh.Data, "timed out") { + errMsg = mh.Data + if cs.Notify { + muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), "network") + go NotificationShared.SendNotification(cs.NotificationGroupID, Localizer.Tf("[TLS] Fetch cert info failed, Reporter: %s, Error: %s", cs.Name, errMsg), muteLabel) + } + } + } else { + // 清除网络错误静音缓存 + NotificationShared.UnMuteNotification(cs.NotificationGroupID, NotificationMuteLabel.ServiceTLS(mh.GetId(), "network")) + + var newCert = strings.Split(mh.Data, "|") + if len(newCert) > 1 { + enableNotify := cs.Notify + + // 首次获取证书信息时，缓存证书信息 + if ss.tlsCertCache[mh.GetId()] == "" { + ss.tlsCertCache[mh.GetId()] = mh.Data + } + + oldCert := strings.Split(ss.tlsCertCache[mh.GetId()], "|") + isCertChanged := false + expiresOld, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", oldCert[1]) + expiresNew, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", newCert[1]) + + // 证书变更时，更新缓存 + if oldCert[0] != newCert[0] && !expiresNew.Equal(expiresOld) { + isCertChanged = true + ss.tlsCertCache[mh.GetId()] = mh.Data + } + + notificationGroupID := cs.NotificationGroupID + serviceName := cs.Name + + // 需要发送提醒 + if enableNotify { + // 证书过期提醒 + if expiresNew.Before(time.Now().AddDate(0, 0, 7)) { + expiresTimeStr := expiresNew.Format("2006-01-02 15:04:05") + errMsg = Localizer.Tf( + "The TLS certificate will expire within seven days. Expiration time: %s", + expiresTimeStr, + ) + + // 静音规则： 服务id+证书过期时间 + // 用于避免多个监测点对相同证书同时报警 + muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), fmt.Sprintf("expire_%s", expiresTimeStr)) + go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel) } - oldCert := strings.Split(ss.tlsCertCache[mh.GetId()], "|") - isCertChanged := false - expiresOld, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", oldCert[1]) - expiresNew, _ := time.Parse("2006-01-02 15:04:05 -0700 MST", newCert[1]) + // 证书变更提醒 + if isCertChanged { + errMsg = Localizer.Tf( + "TLS certificate changed, old: issuer %s, expires at %s; new: issuer %s, expires at %s", + oldCert[0], expiresOld.Format("2006-01-02 15:04:05"), newCert[0], expiresNew.Format("2006-01-02 15:04:05")) - // 证书变更时，更新缓存 - if oldCert[0] != newCert[0] && !expiresNew.Equal(expiresOld) { - isCertChanged = true - ss.tlsCertCache[mh.GetId()] = mh.Data - } - - notificationGroupID := cs.NotificationGroupID - serviceName := cs.Name - - // 需要发送提醒 - if enableNotify { - // 证书过期提醒 - if expiresNew.Before(time.Now().AddDate(0, 0, 7)) { - expiresTimeStr := expiresNew.Format("2006-01-02 15:04:05") - errMsg = Localizer.Tf( - "The TLS certificate will expire within seven days. Expiration time: %s", - expiresTimeStr, - ) - - // 静音规则： 服务id+证书过期时间 - // 用于避免多个监测点对相同证书同时报警 - muteLabel := NotificationMuteLabel.ServiceTLS(mh.GetId(), fmt.Sprintf("expire_%s", expiresTimeStr)) - go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), muteLabel) - } - - // 证书变更提醒 - if isCertChanged { - errMsg = Localizer.Tf( - "TLS certificate changed, old: issuer %s, expires at %s; new: issuer %s, expires at %s", - oldCert[0], expiresOld.Format("2006-01-02 15:04:05"), newCert[0], expiresNew.Format("2006-01-02 15:04:05")) - - // 证书变更后会自动更新缓存，所以不需要静音 - go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), "") - } + // 证书变更后会自动更新缓存，所以不需要静音 + go NotificationShared.SendNotification(notificationGroupID, fmt.Sprintf("[TLS] %s %s", serviceName, errMsg), "") } } } - ss.serviceResponseDataStoreLock.Unlock() } } @@ -776,17 +799,25 @@ func delayCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh return } + // GHSA-jx78-55p5-rwv5 (incomplete fix of GHSA-qjpp-gffx-2wm9): the server + // map snapshot m is taken outside serviceResponseDataStoreLock and + // ServerShared has its own independent lock, so a concurrent batch-delete of + // the reporter's server can remove the entry between the pre-lock validation + // and this point. Guard against the nil pointer before using the server. + reporterServer := m[r.Reporter] + if reporterServer == nil { + return + } + notificationGroupID := ss.NotificationGroupID minMuteLabel := NotificationMuteLabel.ServiceLatencyMin(mh.GetId()) maxMuteLabel := NotificationMuteLabel.ServiceLatencyMax(mh.GetId()) if mh.Delay > ss.MaxLatency { // 延迟超过最大值 - reporterServer := m[r.Reporter] msg := Localizer.Tf("[Latency] %s %2f > %2f, Reporter: %s", ss.Name, mh.Delay, ss.MaxLatency, reporterServer.Name) go NotificationShared.SendNotification(notificationGroupID, msg, minMuteLabel) } else if mh.Delay < ss.MinLatency { // 延迟低于最小值 - reporterServer := m[r.Reporter] msg := Localizer.Tf("[Latency] %s %2f < %2f, Reporter: %s", ss.Name, mh.Delay, ss.MinLatency, reporterServer.Name) go NotificationShared.SendNotification(notificationGroupID, msg, maxMuteLabel) } else { @@ -798,10 +829,16 @@ func delayCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh func notifyCheck(r *ReportData, m map[uint64]*model.Server, ss *model.Service, mh *pb.TaskResult, lastStatus, stateCode uint8) { + // GHSA-jx78-55p5-rwv5: guard against concurrent server deletion (same TOCTOU + // class as the 2026-07-21 fix, a few dozen lines lower in the same worker). + // ServerShared has its own lock; m is a snapshot taken outside + // serviceResponseDataStoreLock, so the server may have been removed between + // the pre-lock validation and here. + reporterServer := m[r.Reporter] + // 判断是否需要发送通知 isNeedSendNotification := ss.Notify && (lastStatus != 0 || stateCode == StatusDown) - if isNeedSendNotification { - reporterServer := m[r.Reporter] + if isNeedSendNotification && reporterServer != nil { notificationGroupID := ss.NotificationGroupID notificationMsg := Localizer.Tf("[%s] %s Reporter: %s, Error: %s", StatusCodeToString(stateCode), ss.Name, reporterServer.Name, mh.Data) muteLabel := NotificationMuteLabel.ServiceStateChanged(mh.GetId()) @@ -816,8 +853,7 @@ func notifyCheck(r *ReportData, m map[uint64]*model.Server, // 判断是否需要触发任务 isNeedTriggerTask := ss.EnableTriggerTask && lastStatus != 0 - if isNeedTriggerTask { - reporterServer := m[r.Reporter] + if isNeedTriggerTask && reporterServer != nil { if stateCode == StatusGood && lastStatus != stateCode { // 当前状态正常 前序状态非正常时 触发恢复任务 go CronShared.SendTriggerTasks(ss.RecoverTriggerTasks, reporterServer.ID, ss.UserID) diff --git a/service/singleton/servicesentinel_lifecycle_test.go b/service/singleton/servicesentinel_lifecycle_test.go index 0941b453..af8a24ac 100644 --- a/service/singleton/servicesentinel_lifecycle_test.go +++ b/service/singleton/servicesentinel_lifecycle_test.go @@ -13,8 +13,183 @@ import ( "github.com/nezhahq/nezha/model" ) +// Regression markers for Finding 1 and Finding 2 of GHSA-jx78-55p5-rwv5 +// (incomplete fix of GHSA-qjpp-gffx-2wm9). +const ( + concurrentServerDeleteSuccessMarker = "ghsa-jx78-55p5-rwv5-finding1-no-crash" + deleteUnknownIDSuccessMarker = "ghsa-jx78-55p5-rwv5-finding2-no-zombie" +) + const serviceSentinelLifecycleSuccessMarker = "service-sentinel-stale-report-lifecycle-success" +func TestServiceSentinelReporterDeleteWaitsForSynchronousReportProcessing(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + service := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "lifecycle-service", + Type: model.TaskTypeTCPPing, + Target: "lifecycle.example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + addServiceMonitorSecurityService(t, ss, service) + + reportValidated := make(chan struct{}) + releaseReport := make(chan struct{}) + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(releaseReport) }) } + ss.serviceReportValidatedHook = func(serviceID uint64) { + if serviceID == service.ID { + close(reportValidated) + <-releaseReport + } + } + t.Cleanup(func() { + release() + ss.Close() + }) + + ss.Dispatch(serviceMonitorResult(1, service.ID, model.TaskTypeTCPPing, true)) + select { + case <-reportValidated: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + + deleteDone := make(chan struct{}) + go func() { + ServerShared.Delete([]uint64{1}) + close(deleteDone) + }() + select { + case <-deleteDone: + t.Fatal("server deletion returned before the accepted report completed") + case <-time.After(25 * time.Millisecond): + } + + release() + select { + case <-deleteDone: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + + var historyCount int64 + if err := DB.Model(&model.ServiceHistory{}). + Where("service_id = ? AND server_id = ?", service.ID, 1). + Count(&historyCount).Error; err != nil { + t.Fatal(err) + } + if historyCount != 1 { + t.Fatalf("expected report side effects before deletion returned, got %d history rows", historyCount) + } + if _, ok := ServerShared.Get(1); ok { + t.Fatal("expected reporter to be deleted after the report completed") + } +} + +func TestServiceSentinelWorkerRejectsReportAfterReporterDeletion(t *testing.T) { + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + service := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "deleted-reporter-service", + Type: model.TaskTypeTCPPing, + Target: "deleted-reporter.example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + addServiceMonitorSecurityService(t, ss, service) + + ServerShared.Delete([]uint64{1}) + ss.Dispatch(serviceMonitorResult(1, service.ID, model.TaskTypeTCPPing, true)) + ss.Close() + + var historyCount int64 + if err := DB.Model(&model.ServiceHistory{}). + Where("service_id = ? AND server_id = ?", service.ID, 1). + Count(&historyCount).Error; err != nil { + t.Fatal(err) + } + if historyCount != 0 { + t.Fatalf("expected no history after reporter deletion, got %d rows", historyCount) + } + ss.serviceResponseDataStoreLock.RLock() + _, pingCached := ss.serviceResponsePing[service.ID] + _, responseCached := ss.serviceResponseDataStore[service.ID] + stats := ss.serviceStatusToday[service.ID] + ss.serviceResponseDataStoreLock.RUnlock() + if pingCached { + t.Fatal("expected no ping cache side effect after reporter deletion") + } + if responseCached { + t.Fatal("expected no response cache side effect after reporter deletion") + } + if stats == nil || stats.Up != 0 || stats.Down != 0 { + t.Fatalf("expected no stats side effect after reporter deletion, got %+v", stats) + } +} + +func TestServiceSentinelWorkerRecoversPerReportPanic(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + panicService := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "panic-service", + Type: model.TaskTypeHTTPGet, + Target: "https://panic.example.invalid", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + validService := &model.Service{ + Common: model.Common{ID: 20, UserID: 1}, + Name: "valid-service", + Type: model.TaskTypeTCPPing, + Target: "valid.example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + addServiceMonitorSecurityService(t, ss, panicService) + addServiceMonitorSecurityService(t, ss, validService) + ss.serviceReportBeforeTLSSideEffectsHook = func(serviceID uint64) { + if serviceID == panicService.ID { + panic("test service report panic") + } + } + + ss.Dispatch(serviceMonitorResult(1, panicService.ID, model.TaskTypeHTTPGet, true)) + ss.Dispatch(serviceMonitorResult(1, validService.ID, model.TaskTypeTCPPing, true)) + waitForServiceHistory(t, validService.ID, 1) + ss.Close() + if !ss.serviceResponseDataStoreLock.TryLock() { + t.Fatal("panic leaked the service response lock") + } + ss.serviceResponseDataStoreLock.Unlock() + + deleteDone := make(chan struct{}) + go func() { + ServerShared.Delete([]uint64{1}) + close(deleteDone) + }() + select { + case <-deleteDone: + case <-ctx.Done(): + t.Fatal("panic leaked a lifecycle lock: " + ctx.Err().Error()) + } +} + func TestServiceSentinelWorkerIgnoresStaleReportAfterDeletion(t *testing.T) { if os.Getenv("NEZHA_SERVICE_SENTINEL_LIFECYCLE_CHILD") == "1" { testServiceSentinelWorkerIgnoresStaleReportAfterDeletionChild(t) @@ -299,3 +474,173 @@ func TestServiceSentinelWorkerHoldsResponseLockDuringTLSSideEffects(t *testing.T t.Fatalf("expected TLS cache %q, got %q", report.Data.Data, cachedCertificate) } } + +// TestServiceSentinelWorkerSurvivesConcurrentReporterServerDelete is a +// regression test for GHSA-jx78-55p5-rwv5 Finding 1 (incomplete fix of +// GHSA-qjpp-gffx-2wm9). +// +// The vulnerability: after the 2026-07-21 fix, the worker re-validates the +// service under serviceResponseDataStoreLock, but then takes a fresh snapshot +// m := ServerShared.GetList() with no guard. A concurrent batch-delete of the +// reporter's own server removes it between the pre-lock validation and the +// GetList call, so m[r.Reporter] is nil. delayCheck and notifyCheck then +// dereference m[r.Reporter].Name unconditionally — SIGSEGV. +// +// The subprocess-isolation pattern is used because the pre-fix code path +// panicked (nil pointer dereference in an unrecovered goroutine), which would +// crash the whole test binary rather than simply failing a single test. +func TestServiceSentinelWorkerSurvivesConcurrentReporterServerDelete(t *testing.T) { + if os.Getenv("NEZHA_SENTINEL_CONCURRENT_DELETE_CHILD") == "1" { + testServiceSentinelWorkerSurvivesConcurrentReporterServerDeleteChild(t) + return + } + + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + child := exec.CommandContext(ctx, os.Args[0], + "-test.run=^TestServiceSentinelWorkerSurvivesConcurrentReporterServerDelete$", + "-test.v", + ) + child.Env = append(os.Environ(), "NEZHA_SENTINEL_CONCURRENT_DELETE_CHILD=1") + + output, err := child.CombinedOutput() + if ctx.Err() != nil { + t.Fatalf("child process timed out: %v\n%s", ctx.Err(), output) + } + if err != nil { + t.Fatalf("child process crashed (likely nil deref in delayCheck/notifyCheck): %v\n%s", err, output) + } + if !strings.Contains(string(output), concurrentServerDeleteSuccessMarker) { + t.Fatalf("child did not print success marker:\n%s", output) + } +} + +func testServiceSentinelWorkerSurvivesConcurrentReporterServerDeleteChild(t *testing.T) { + // Given: a reporter server and a service with latency-alerting enabled so + // that delayCheck (the vulnerable sink at line 785) is exercised on every + // dispatch. MaxLatency=1 ensures delay=12 always exceeds the threshold and + // the notification branch (not just the mute-clear branch) is taken. + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + service := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "latency-service", + Type: model.TaskTypeTCPPing, + Target: "example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + LatencyNotify: true, + MaxLatency: 1, + } + addServiceMonitorSecurityService(t, ss, service) + + reportProcessing := make(chan struct{}) + releaseWorker := make(chan struct{}) + var releaseOnce sync.Once + releaseWorkerFn := func() { releaseOnce.Do(func() { close(releaseWorker) }) } + + // serviceReportValidatedHook runs while the report holds the lifecycle read + // lock. Deletion must therefore run in another goroutine and wait until this + // hook releases; attempting Delete here would try to upgrade the RWMutex. + ss.serviceReportValidatedHook = func(serviceID uint64) { + if serviceID == service.ID { + close(reportProcessing) + <-releaseWorker + } + } + t.Cleanup(func() { + releaseWorkerFn() + ss.Close() + }) + + // When + ss.Dispatch(serviceMonitorResult(1, service.ID, model.TaskTypeTCPPing, true)) + select { + case <-reportProcessing: + case <-t.Context().Done(): + t.Fatal(t.Context().Err()) + } + deleteDone := make(chan struct{}) + go func() { + ServerShared.Delete([]uint64{1}) + close(deleteDone) + }() + releaseWorkerFn() + select { + case <-deleteDone: + case <-t.Context().Done(): + t.Fatal(t.Context().Err()) + } + ss.Close() + + // Then: no crash; the worker handled the nil reporter gracefully. + if _, err := fmt.Fprintln(os.Stdout, concurrentServerDeleteSuccessMarker); err != nil { + t.Fatal(err) + } +} + +// TestServiceSentinelDeleteWithUnknownIDDoesNotLeaveZombies is a regression +// test for GHSA-jx78-55p5-rwv5 Finding 2 (low severity). +// +// The vulnerability: ServiceSentinel.Delete iterates the caller-supplied id +// slice and does CronShared.Remove(ss.services[id].CronJobID) without checking +// whether id is present in ss.services. CheckPermission returns vacuously +// true for unknown ids, so the controller layer cannot block this path. +// ss.services[unknownID] returns nil, and .CronJobID panics. Because the +// panic aborts the loop, every id ordered AFTER the bogus one is never removed +// from the in-memory registry even though its database row was already deleted, +// producing zombie services that keep dispatching cron probes. +func TestServiceSentinelDeleteWithUnknownIDDoesNotLeaveZombies(t *testing.T) { + // Given: one legitimate service (ID 10) registered in the sentinel. + ss := newServiceMonitorSecurityHarness(t, + &model.Server{Common: model.Common{ID: 1, UserID: 1}, Name: "reporter"}, + ) + service := &model.Service{ + Common: model.Common{ID: 10, UserID: 1}, + Name: "real-service", + Type: model.TaskTypeTCPPing, + Target: "example.invalid:443", + Duration: 3600, + Cover: model.ServiceCoverIgnoreAll, + SkipServers: map[uint64]bool{1: true}, + } + addServiceMonitorSecurityService(t, ss, service) + + // When: Delete is called with a bogus ID first, then the real service ID. + // Before the fix this panicked on ss.services[99999].CronJobID and left + // service 10 as a zombie. + ss.Delete([]uint64{99999, service.ID}) + + // Then: the real service must be fully removed from every in-memory map. + ss.serviceResponseDataStoreLock.RLock() + _, todayPresent := ss.serviceStatusToday[service.ID] + _, pingPresent := ss.serviceResponsePing[service.ID] + ss.serviceResponseDataStoreLock.RUnlock() + + ss.servicesLock.RLock() + _, servicePresent := ss.services[service.ID] + ss.servicesLock.RUnlock() + + ss.monthlyStatusLock.Lock() + _, monthlyPresent := ss.monthlyStatus[service.ID] + ss.monthlyStatusLock.Unlock() + + if todayPresent { + t.Error("zombie: serviceStatusToday still contains the deleted service") + } + if pingPresent { + t.Error("zombie: serviceResponsePing still contains the deleted service") + } + if servicePresent { + t.Error("zombie: services map still contains the deleted service") + } + if monthlyPresent { + t.Error("zombie: monthlyStatus still contains the deleted service") + } + + if _, err := fmt.Fprintln(os.Stdout, deleteUnknownIDSuccessMarker); err != nil { + t.Fatal(err) + } +} diff --git a/service/singleton/testhelpers.go b/service/singleton/testhelpers.go index eb60b9c6..8e889a27 100644 --- a/service/singleton/testhelpers.go +++ b/service/singleton/testhelpers.go @@ -20,6 +20,9 @@ func NewEmptyServerClassForTest() *ServerClass { // InsertForTest 把一个 server 直接塞进内存表与排序快照，跳过 DB & InitServer 逻辑。 // 调用方需保证 server.ID 已经设置。 func (c *ServerClass) InsertForTest(s *model.Server) { + c.lockLifecycleWrite() + defer c.unlockLifecycleWrite() + c.listMu.Lock() c.list[s.ID] = s if s.UUID != "" {