diff --git a/go.mod b/go.mod index 118391f29..d1159f0d9 100644 --- a/go.mod +++ b/go.mod @@ -442,7 +442,7 @@ require ( golang.org/x/crypto v0.46.0 golang.org/x/mod v0.31.0 // indirect golang.org/x/net v0.48.0 // indirect - golang.org/x/sync v0.19.0 // indirect + golang.org/x/sync v0.19.0 golang.org/x/sys v0.40.0 // indirect golang.org/x/text v0.32.0 // indirect golang.org/x/time v0.14.0 // indirect diff --git a/pkg/policies/concurrency_test.go b/pkg/policies/concurrency_test.go new file mode 100644 index 000000000..0269d0f80 --- /dev/null +++ b/pkg/policies/concurrency_test.go @@ -0,0 +1,233 @@ +// +// Copyright 2026 The Chainloop Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package policies + +import ( + "context" + "os" + "sync" + "testing" + + v12 "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1" + v1 "github.com/chainloop-dev/chainloop/pkg/attestation/crafter/api/attestation/v1" + intoto "github.com/in-toto/attestation/go/v1" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/encoding/protojson" +) + +// TestConcurrentVerifyStatement runs VerifyStatement from multiple goroutines +// to exercise errgroup parallelism and catch race conditions. +// Run with: go test -race -count=1 -run TestConcurrentVerifyStatement ./pkg/policies/... +func TestConcurrentVerifyStatement(t *testing.T) { + logger := zerolog.Nop() + + schema := &v12.CraftingSchema{ + Policies: &v12.Policies{ + Attestation: []*v12.PolicyAttachment{ + {Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/workflow.yaml"}}, + {Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/materials.yaml"}}, + }, + }, + } + + statement := loadStatementForTest(t, "testdata/statement.json") + + pv := NewPolicyVerifier(schema.GetPolicies(), nil, &logger) + + // Run multiple VerifyStatement calls concurrently + const goroutines = 10 + var wg sync.WaitGroup + errs := make([]error, goroutines) + results := make([][]*v1.PolicyEvaluation, goroutines) + + for i := range goroutines { + wg.Add(1) + go func() { + defer wg.Done() + res, err := pv.VerifyStatement(context.Background(), statement) + errs[i] = err + results[i] = res + }() + } + + wg.Wait() + + // All calls should succeed + for i := range goroutines { + assert.NoError(t, errs[i], "goroutine %d failed", i) + } + + // All calls should return the same number of evaluations + for i := 1; i < goroutines; i++ { + assert.Equal(t, len(results[0]), len(results[i]), + "goroutine %d returned different number of evaluations", i) + } +} + +// TestConcurrentVerifyMaterial runs VerifyMaterial from multiple goroutines. +func TestConcurrentVerifyMaterial(t *testing.T) { + logger := zerolog.Nop() + + schema := &v12.CraftingSchema{ + Policies: &v12.Policies{ + Materials: []*v12.PolicyAttachment{ + { + Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/sbom_syft.yaml"}, + }, + }, + }, + } + + material := &v1.Attestation_Material{ + Id: "sbom", + MaterialType: v12.CraftingSchema_Material_SBOM_SPDX_JSON, + M: &v1.Attestation_Material_Artifact_{ + Artifact: &v1.Attestation_Material_Artifact{}, + }, + } + + pv := NewPolicyVerifier(schema.GetPolicies(), nil, &logger) + + const goroutines = 10 + var wg sync.WaitGroup + errs := make([]error, goroutines) + results := make([][]*v1.PolicyEvaluation, goroutines) + + for i := range goroutines { + wg.Add(1) + go func() { + defer wg.Done() + res, err := pv.VerifyMaterial(context.Background(), material, "testdata/sbom-spdx.json") + errs[i] = err + results[i] = res + }() + } + + wg.Wait() + + for i := range goroutines { + assert.NoError(t, errs[i], "goroutine %d failed", i) + } + + for i := 1; i < goroutines; i++ { + assert.Equal(t, len(results[0]), len(results[i]), + "goroutine %d returned different number of evaluations", i) + } +} + +// TestWithMaxConcurrency verifies the WithMaxConcurrency option is applied. +func TestWithMaxConcurrency(t *testing.T) { + logger := zerolog.Nop() + + // Test default + pv := NewPolicyVerifier(nil, nil, &logger) + assert.Greater(t, pv.maxConcurrency, 0, "default maxConcurrency should be positive") + + // Test custom value + pv = NewPolicyVerifier(nil, nil, &logger, WithMaxConcurrency(5)) + assert.Equal(t, 5, pv.maxConcurrency) + + // Test zero defaults to computed default + pv = NewPolicyVerifier(nil, nil, &logger, WithMaxConcurrency(0)) + assert.Equal(t, defaultMaxConcurrency, pv.maxConcurrency) + + // Test negative defaults to computed default + pv = NewPolicyVerifier(nil, nil, &logger, WithMaxConcurrency(-1)) + assert.Equal(t, defaultMaxConcurrency, pv.maxConcurrency) +} + +// TestVerifyStatementWithConcurrencyOne ensures sequential mode (maxConcurrency=1) +// produces identical results to default parallel mode. +func TestVerifyStatementWithConcurrencyOne(t *testing.T) { + logger := zerolog.Nop() + + schema := &v12.CraftingSchema{ + Policies: &v12.Policies{ + Attestation: []*v12.PolicyAttachment{ + {Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/workflow.yaml"}}, + {Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/materials.yaml"}}, + }, + }, + } + + statement := loadStatementForTest(t, "testdata/statement.json") + + // Sequential + pvSeq := NewPolicyVerifier(schema.GetPolicies(), nil, &logger, WithMaxConcurrency(1)) + seqResults, seqErr := pvSeq.VerifyStatement(context.Background(), statement) + + // Parallel (default) + pvPar := NewPolicyVerifier(schema.GetPolicies(), nil, &logger) + parResults, parErr := pvPar.VerifyStatement(context.Background(), statement) + + require.NoError(t, seqErr) + require.NoError(t, parErr) + assert.Equal(t, len(seqResults), len(parResults), + "sequential and parallel should return same number of evaluations") + + // Build name→result maps for comparison (order may differ) + seqByName := make(map[string]*v1.PolicyEvaluation) + for _, ev := range seqResults { + seqByName[ev.Name] = ev + } + + for _, ev := range parResults { + seqEv, ok := seqByName[ev.Name] + assert.True(t, ok, "parallel result has policy %q not found in sequential", ev.Name) + if ok { + assert.Equal(t, len(seqEv.Violations), len(ev.Violations), + "policy %q: violation count mismatch", ev.Name) + assert.Equal(t, seqEv.Skipped, ev.Skipped, + "policy %q: skipped mismatch", ev.Name) + } + } +} + +// TestErrGroupCancellation verifies that when one policy evaluation fails, +// the errgroup context is cancelled and propagated. +func TestErrGroupCancellation(t *testing.T) { + logger := zerolog.Nop() + + schema := &v12.CraftingSchema{ + Policies: &v12.Policies{ + Attestation: []*v12.PolicyAttachment{ + {Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/workflow.yaml"}}, + // This policy ref does not exist — will cause an error + {Policy: &v12.PolicyAttachment_Ref{Ref: "file://testdata/nonexistent_policy.yaml"}}, + }, + }, + } + + statement := loadStatementForTest(t, "testdata/statement.json") + + pv := NewPolicyVerifier(schema.GetPolicies(), nil, &logger) + _, err := pv.VerifyStatement(context.Background(), statement) + + assert.Error(t, err, "should fail when a policy file does not exist") + assert.IsType(t, &PolicyError{}, err, "error should be a PolicyError") +} + +func loadStatementForTest(t *testing.T, file string) *intoto.Statement { + t.Helper() + stContent, err := os.ReadFile(file) + require.NoError(t, err) + var statement intoto.Statement + err = protojson.Unmarshal(stContent, &statement) + require.NoError(t, err) + return &statement +} diff --git a/pkg/policies/engine/wasm/engine.go b/pkg/policies/engine/wasm/engine.go index 9b191971e..dc5f2b2fc 100644 --- a/pkg/policies/engine/wasm/engine.go +++ b/pkg/policies/engine/wasm/engine.go @@ -19,6 +19,7 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" "github.com/chainloop-dev/chainloop/pkg/policies/engine" @@ -28,6 +29,12 @@ import ( "github.com/rs/zerolog" ) +// setLogLevelOnce ensures extism.SetLogLevel is called exactly once, +// since it modifies global state and is not safe for concurrent calls. +// Note: the log level is determined by the first NewEngine() call in the +// process. Subsequent engines with different log levels will not override it. +var setLogLevelOnce sync.Once + // Ensure Engine implements PolicyEngine interface var _ engine.PolicyEngine = (*Engine)(nil) @@ -59,6 +66,23 @@ func NewEngine(opts ...engine.Option) *Engine { logger = &noopLogger } + // Set WASM plugin log level exactly once across all engine instances. + // extism.SetLogLevel modifies global state and is not safe for concurrent calls. + setLogLevelOnce.Do(func() { + switch { + case logger.GetLevel() <= zerolog.TraceLevel: + extism.SetLogLevel(extism.LogLevelTrace) + case logger.GetLevel() <= zerolog.DebugLevel: + extism.SetLogLevel(extism.LogLevelDebug) + case logger.GetLevel() <= zerolog.InfoLevel: + extism.SetLogLevel(extism.LogLevelInfo) + case logger.GetLevel() <= zerolog.WarnLevel: + extism.SetLogLevel(extism.LogLevelWarn) + default: + extism.SetLogLevel(extism.LogLevelError) + } + }) + return &Engine{ executionTimeout: executionTimeout, logger: logger, @@ -70,21 +94,6 @@ func NewEngine(opts ...engine.Option) *Engine { func (e *Engine) Verify(ctx context.Context, policy *engine.Policy, input []byte, args map[string]any) (*engine.EvaluationResult, error) { e.logger.Debug().Str("policy", policy.Name).Int("wasm_size", len(policy.Source)).Int("input_size", len(input)).Int("args_count", len(args)).Msg("Starting WASM policy execution") - // Enable WASM plugin logging based on logger level - // This allows LogInfo(), LogDebug(), etc. from the WASM policy to be visible - switch { - case e.logger.GetLevel() <= zerolog.TraceLevel: - extism.SetLogLevel(extism.LogLevelTrace) - case e.logger.GetLevel() <= zerolog.DebugLevel: - extism.SetLogLevel(extism.LogLevelDebug) - case e.logger.GetLevel() <= zerolog.InfoLevel: - extism.SetLogLevel(extism.LogLevelInfo) - case e.logger.GetLevel() <= zerolog.WarnLevel: - extism.SetLogLevel(extism.LogLevelWarn) - default: - extism.SetLogLevel(extism.LogLevelError) - } - // Prepare config with args if present configMap := make(map[string]string) if len(args) > 0 { diff --git a/pkg/policies/group_loader.go b/pkg/policies/group_loader.go index 5e24bd40b..9f2ccb496 100644 --- a/pkg/policies/group_loader.go +++ b/pkg/policies/group_loader.go @@ -24,10 +24,12 @@ import ( "os" "path/filepath" "sync" + "time" pb "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1" v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1" crv1 "github.com/google/go-containerregistry/pkg/v1" + "golang.org/x/sync/singleflight" ) // GroupLoader defines the interface for policy loaders from contract attachments @@ -99,8 +101,6 @@ func (l *HTTPSGroupLoader) Load(_ context.Context, attachment *v1.PolicyGroupAtt // ChainloopGroupLoader loads groups referenced with chainloop://provider/name URLs type ChainloopGroupLoader struct { Client pb.AttestationServiceClient - - cacheMutex sync.Mutex } type groupWithReference struct { @@ -108,7 +108,16 @@ type groupWithReference struct { reference *PolicyDescriptor } -var remoteGroupCache = make(map[string]*groupWithReference) +// remoteGroupFetchTimeout bounds gRPC calls inside singleflight to prevent +// indefinite blocking. Independent of any caller's context deadline so that +// all singleflight waiters get uniform timeout behavior. +const remoteGroupFetchTimeout = 30 * time.Second + +var ( + remoteGroupCache = make(map[string]*groupWithReference) + remoteGroupCacheMutex sync.Mutex + remoteGroupFlight singleflight.Group +) func NewChainloopGroupLoader(client pb.AttestationServiceClient) *ChainloopGroupLoader { return &ChainloopGroupLoader{Client: client} @@ -117,43 +126,63 @@ func NewChainloopGroupLoader(client pb.AttestationServiceClient) *ChainloopGroup func (c *ChainloopGroupLoader) Load(ctx context.Context, attachment *v1.PolicyGroupAttachment) (*v1.PolicyGroup, *PolicyDescriptor, error) { ref := attachment.GetRef() - c.cacheMutex.Lock() - defer c.cacheMutex.Unlock() - + // Fast path: check cache under lock + remoteGroupCacheMutex.Lock() if v, ok := remoteGroupCache[ref]; ok { + remoteGroupCacheMutex.Unlock() return v.group, v.reference, nil } + remoteGroupCacheMutex.Unlock() + + // Use singleflight to coalesce concurrent fetches for the same ref. + // Use context.WithoutCancel so the winning goroutine's context cancellation + // doesn't propagate to waiters sharing the same singleflight key. + result, err, _ := remoteGroupFlight.Do(ref, func() (interface{}, error) { + if !IsProviderScheme(ref) { + return nil, fmt.Errorf("invalid group reference %q", ref) + } - if !IsProviderScheme(ref) { - return nil, nil, fmt.Errorf("invalid group reference %q", ref) - } + providerRef := ProviderParts(ref) + // Use a fixed timeout independent of any caller's context so that all + // singleflight waiters get uniform behavior regardless of which goroutine + // won the race. + sfCtx, cancel := context.WithTimeout(context.Background(), remoteGroupFetchTimeout) + defer cancel() + + resp, err := c.Client.GetPolicyGroup(sfCtx, &pb.AttestationServiceGetPolicyGroupRequest{ + Provider: providerRef.Provider, + GroupName: providerRef.Name, + OrgName: providerRef.OrgName, + }) + if err != nil { + return nil, fmt.Errorf("requesting remote group (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err) + } - providerRef := ProviderParts(ref) + h, err := crv1.NewHash(resp.Reference.GetDigest()) + if err != nil { + return nil, fmt.Errorf("parsing digest: %w", err) + } - resp, err := c.Client.GetPolicyGroup(ctx, &pb.AttestationServiceGetPolicyGroupRequest{ - Provider: providerRef.Provider, - GroupName: providerRef.Name, - OrgName: providerRef.OrgName, - }) - if err != nil { - return nil, nil, fmt.Errorf("requesting remote group (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err) - } + orgName := providerRef.OrgName + if u, err := url.Parse(resp.Reference.GetUrl()); err == nil { + if orgParam := u.Query().Get("org"); orgParam != "" { + orgName = orgParam + } + } - h, err := crv1.NewHash(resp.Reference.GetDigest()) - if err != nil { - return nil, nil, fmt.Errorf("parsing digest: %w", err) - } + reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h) + cached := &groupWithReference{group: resp.GetGroup(), reference: reference} - orgName := providerRef.OrgName - // Extract organization name from URL if present - if u, err := url.Parse(resp.Reference.GetUrl()); err == nil { - if orgParam := u.Query().Get("org"); orgParam != "" { - orgName = orgParam - } + remoteGroupCacheMutex.Lock() + remoteGroupCache[ref] = cached + remoteGroupCacheMutex.Unlock() + + return cached, nil + }) + if err != nil { + return nil, nil, err } - reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h) - // cache result - remoteGroupCache[ref] = &groupWithReference{group: resp.GetGroup(), reference: reference} - return resp.GetGroup(), reference, nil + cached := result.(*groupWithReference) + return cached.group, cached.reference, nil } diff --git a/pkg/policies/loader.go b/pkg/policies/loader.go index 995bd5461..49745baa4 100644 --- a/pkg/policies/loader.go +++ b/pkg/policies/loader.go @@ -26,11 +26,13 @@ import ( "path/filepath" "strings" "sync" + "time" pb "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1" v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/unmarshal" crv1 "github.com/google/go-containerregistry/pkg/v1" + "golang.org/x/sync/singleflight" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" ) @@ -158,8 +160,6 @@ func (l *HTTPSLoader) Load(_ context.Context, attachment *v1.PolicyAttachment) ( // ChainloopLoader loads policies referenced with chainloop://provider/name URLs type ChainloopLoader struct { Client pb.AttestationServiceClient - - cacheMutex sync.Mutex } type policyWithReference struct { @@ -167,7 +167,16 @@ type policyWithReference struct { reference *PolicyDescriptor } -var remotePolicyCache = make(map[string]*policyWithReference) +// remotePolicyFetchTimeout bounds gRPC calls inside singleflight to prevent +// indefinite blocking. Independent of any caller's context deadline so that +// all singleflight waiters get uniform timeout behavior. +const remotePolicyFetchTimeout = 30 * time.Second + +var ( + remotePolicyCache = make(map[string]*policyWithReference) + remotePolicyCacheMutex sync.Mutex + remotePolicyFlight singleflight.Group +) func NewChainloopLoader(client pb.AttestationServiceClient) *ChainloopLoader { return &ChainloopLoader{Client: client} @@ -176,46 +185,67 @@ func NewChainloopLoader(client pb.AttestationServiceClient) *ChainloopLoader { func (c *ChainloopLoader) Load(ctx context.Context, attachment *v1.PolicyAttachment) (*v1.Policy, *PolicyDescriptor, error) { ref := attachment.GetRef() - c.cacheMutex.Lock() - defer c.cacheMutex.Unlock() - + // Fast path: check cache under lock + remotePolicyCacheMutex.Lock() if v, ok := remotePolicyCache[ref]; ok { + remotePolicyCacheMutex.Unlock() return v.policy, v.reference, nil } + remotePolicyCacheMutex.Unlock() + + // Use singleflight to coalesce concurrent fetches for the same ref. + // This ensures exactly one gRPC call per ref regardless of concurrency. + // Use context.WithoutCancel so the winning goroutine's context cancellation + // doesn't propagate to waiters sharing the same singleflight key. + result, err, _ := remotePolicyFlight.Do(ref, func() (interface{}, error) { + if !IsProviderScheme(ref) { + return nil, fmt.Errorf("invalid policy reference %q", ref) + } - if !IsProviderScheme(ref) { - return nil, nil, fmt.Errorf("invalid policy reference %q", ref) - } + providerRef := ProviderParts(ref) + // Use a fixed timeout independent of any caller's context so that all + // singleflight waiters get uniform behavior regardless of which goroutine + // won the race. The caller's cancellation/deadline is intentionally not + // inherited to avoid one caller's short deadline failing the shared call. + sfCtx, cancel := context.WithTimeout(context.Background(), remotePolicyFetchTimeout) + defer cancel() + + resp, err := c.Client.GetPolicy(sfCtx, &pb.AttestationServiceGetPolicyRequest{ + Provider: providerRef.Provider, + PolicyName: providerRef.Name, + OrgName: providerRef.OrgName, + }) + if err != nil { + return nil, fmt.Errorf("requesting remote policy (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err) + } - providerRef := ProviderParts(ref) + h, err := crv1.NewHash(resp.Reference.GetDigest()) + if err != nil { + return nil, fmt.Errorf("parsing digest: %w", err) + } - resp, err := c.Client.GetPolicy(ctx, &pb.AttestationServiceGetPolicyRequest{ - Provider: providerRef.Provider, - PolicyName: providerRef.Name, - OrgName: providerRef.OrgName, - }) - if err != nil { - return nil, nil, fmt.Errorf("requesting remote policy (provider: %s, name: %s): %w", providerRef.Provider, providerRef.Name, err) - } + orgName := providerRef.OrgName + if u, err := url.Parse(resp.Reference.GetUrl()); err == nil { + if orgParam := u.Query().Get("org"); orgParam != "" { + orgName = orgParam + } + } - h, err := crv1.NewHash(resp.Reference.GetDigest()) - if err != nil { - return nil, nil, fmt.Errorf("parsing digest: %w", err) - } + reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h) + cached := &policyWithReference{policy: resp.GetPolicy(), reference: reference} - orgName := providerRef.OrgName - // Extract organization name from URL if present - if u, err := url.Parse(resp.Reference.GetUrl()); err == nil { - if orgParam := u.Query().Get("org"); orgParam != "" { - orgName = orgParam - } - } + remotePolicyCacheMutex.Lock() + remotePolicyCache[ref] = cached + remotePolicyCacheMutex.Unlock() - reference := policyReferenceResourceDescriptor(providerRef.Name, resp.Reference.GetUrl(), orgName, h) + return cached, nil + }) + if err != nil { + return nil, nil, err + } - // cache result - remotePolicyCache[ref] = &policyWithReference{policy: resp.GetPolicy(), reference: reference} - return resp.GetPolicy(), reference, nil + cached := result.(*policyWithReference) + return cached.policy, cached.reference, nil } func unmarshallResource(raw []byte, ref string, digest string, dest proto.Message) (*PolicyDescriptor, error) { diff --git a/pkg/policies/policies.go b/pkg/policies/policies.go index 486cf58ee..2000f62b5 100644 --- a/pkg/policies/policies.go +++ b/pkg/policies/policies.go @@ -22,6 +22,7 @@ import ( "fmt" "net/url" "path/filepath" + "runtime" "slices" "strings" @@ -40,6 +41,7 @@ import ( "github.com/chainloop-dev/chainloop/pkg/policies/engine" "github.com/chainloop-dev/chainloop/pkg/policies/engine/rego" "github.com/chainloop-dev/chainloop/pkg/policies/engine/wasm" + "golang.org/x/sync/errgroup" ) type PolicyError struct { @@ -72,6 +74,11 @@ const ( EvalPhasePush ) +// defaultMaxConcurrency is the default number of concurrent policy evaluations. +// Kept conservative to avoid contention on remote state optimistic locking +// and to limit pressure on AI-enabled policies. Can be overridden via WithMaxConcurrency. +var defaultMaxConcurrency = max(runtime.NumCPU(), 5) + type PolicyVerifier struct { policies *v1.Policies logger *zerolog.Logger @@ -82,6 +89,7 @@ type PolicyVerifier struct { includeRawData bool enablePrint bool evalPhase EvalPhase + maxConcurrency int } var _ Verifier = (*PolicyVerifier)(nil) @@ -93,6 +101,7 @@ type PolicyVerifierOptions struct { EnablePrint bool GRPCConn *grpc.ClientConn EvalPhase EvalPhase + MaxConcurrency int } type PolicyVerifierOption func(*PolicyVerifierOptions) @@ -133,12 +142,23 @@ func WithEvalPhase(phase EvalPhase) PolicyVerifierOption { } } +func WithMaxConcurrency(n int) PolicyVerifierOption { + return func(o *PolicyVerifierOptions) { + o.MaxConcurrency = n + } +} + func NewPolicyVerifier(policies *v1.Policies, client v13.AttestationServiceClient, logger *zerolog.Logger, opts ...PolicyVerifierOption) *PolicyVerifier { options := &PolicyVerifierOptions{} for _, opt := range opts { opt(options) } + maxConcurrency := options.MaxConcurrency + if maxConcurrency <= 0 { + maxConcurrency = defaultMaxConcurrency + } + return &PolicyVerifier{ policies: policies, client: client, @@ -149,6 +169,7 @@ func NewPolicyVerifier(policies *v1.Policies, client v13.AttestationServiceClien includeRawData: options.IncludeRawData, enablePrint: options.EnablePrint, evalPhase: options.EvalPhase, + maxConcurrency: maxConcurrency, } } @@ -171,14 +192,29 @@ func (pv *PolicyVerifier) VerifyMaterial(ctx context.Context, material *v12.Atte return nil, NewPolicyError(err) } - for _, attachment := range attachments { - ev, err := pv.evaluatePolicyAttachment(ctx, attachment, subject, - &evalOpts{kind: material.MaterialType, name: material.GetId()}, - ) - if err != nil { - return nil, NewPolicyError(err) - } + results := make([]*v12.PolicyEvaluation, len(attachments)) + g, gCtx := errgroup.WithContext(ctx) + g.SetLimit(pv.maxConcurrency) + for i, attachment := range attachments { + g.Go(func() error { + ev, err := pv.evaluatePolicyAttachment(gCtx, attachment, subject, + &evalOpts{kind: material.MaterialType, name: material.GetId()}, + ) + if err != nil { + return NewPolicyError(err) + } + results[i] = ev + return nil + }) + } + + if err := g.Wait(); err != nil { + return nil, err + } + + // Filter nil entries (skipped policies) + for _, ev := range results { if ev != nil { result = append(result, ev) } @@ -425,19 +461,36 @@ func ComputeArguments(name string, inputs []*v1.PolicyInput, args map[string]str // VerifyStatement verifies that the statement is compliant with the policies present in the schema func (pv *PolicyVerifier) VerifyStatement(ctx context.Context, statement *intoto.Statement) ([]*v12.PolicyEvaluation, error) { - result := make([]*v12.PolicyEvaluation, 0) policies := pv.policies.GetAttestation() - for _, policyAtt := range policies { - material, err := protojson.Marshal(statement) - if err != nil { - return nil, NewPolicyError(err) - } - ev, err := pv.evaluatePolicyAttachment(ctx, policyAtt, material, &evalOpts{kind: v1.CraftingSchema_Material_ATTESTATION}) - if err != nil { - return nil, NewPolicyError(err) - } + // Marshal statement once — it's read-only input shared across evaluations + statementJSON, err := protojson.Marshal(statement) + if err != nil { + return nil, NewPolicyError(err) + } + + results := make([]*v12.PolicyEvaluation, len(policies)) + g, gCtx := errgroup.WithContext(ctx) + g.SetLimit(pv.maxConcurrency) + + for i, policyAtt := range policies { + g.Go(func() error { + ev, err := pv.evaluatePolicyAttachment(gCtx, policyAtt, statementJSON, &evalOpts{kind: v1.CraftingSchema_Material_ATTESTATION}) + if err != nil { + return NewPolicyError(err) + } + results[i] = ev + return nil + }) + } + + if err := g.Wait(); err != nil { + return nil, err + } + // Filter nil entries (skipped policies) + result := make([]*v12.PolicyEvaluation, 0, len(policies)) + for _, ev := range results { if ev != nil { result = append(result, ev) } diff --git a/pkg/policies/policy_groups.go b/pkg/policies/policy_groups.go index f77bdd97f..bbd339278 100644 --- a/pkg/policies/policy_groups.go +++ b/pkg/policies/policy_groups.go @@ -27,6 +27,7 @@ import ( "github.com/chainloop-dev/chainloop/pkg/templates" intoto "github.com/in-toto/attestation/go/v1" "github.com/rs/zerolog" + "golang.org/x/sync/errgroup" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" ) @@ -74,44 +75,58 @@ func (pgv *PolicyGroupVerifier) VerifyMaterial(ctx context.Context, material *ap return nil, NewPolicyError(err) } - for _, policyAtt := range policyAtts { - // Check if policy should be skipped - skip, policyName, err := pgv.shouldSkipPolicy(ctx, policyAtt, groupAtt.GetSkip()) - if err != nil { - return nil, NewPolicyError(fmt.Errorf("failed to check if policy should be skipped: %w", err)) - } - - if skip { - pgv.logger.Debug().Str("policy", policyName).Msg("skipping attestation policy per skip list") - continue - } - - // Load material content - subject, err := material.GetEvaluableContent(path) - if err != nil { - return nil, NewPolicyError(err) - } + // Load material content once for all policies in this group + subject, err := material.GetEvaluableContent(path) + if err != nil { + return nil, NewPolicyError(err) + } - ev, err := pgv.evaluatePolicyAttachment(ctx, applyGroupGate(policyAtt, groupAtt), subject, - &evalOpts{kind: material.MaterialType, name: material.GetId(), bindings: groupArgs}, - ) - if err != nil { - return nil, NewPolicyError(err) - } + groupResults := make([]*api.PolicyEvaluation, len(policyAtts)) + g, gCtx := errgroup.WithContext(ctx) + g.SetLimit(pgv.maxConcurrency) + + for i, policyAtt := range policyAtts { + g.Go(func() error { + // Check if policy should be skipped + skip, policyName, err := pgv.shouldSkipPolicy(gCtx, policyAtt, groupAtt.GetSkip()) + if err != nil { + return NewPolicyError(fmt.Errorf("failed to check if policy should be skipped: %w", err)) + } + + if skip { + pgv.logger.Debug().Str("policy", policyName).Msg("skipping attestation policy per skip list") + return nil + } + + ev, err := pgv.evaluatePolicyAttachment(gCtx, applyGroupGate(policyAtt, groupAtt), subject, + &evalOpts{kind: material.MaterialType, name: material.GetId(), bindings: groupArgs}, + ) + if err != nil { + return NewPolicyError(err) + } + + if ev != nil { + // Assign group reference to this evaluation + ev.GroupReference = &api.PolicyEvaluation_Reference{ + Name: group.GetMetadata().GetName(), + Digest: desc.GetDigest(), + Uri: desc.GetURI(), + OrgName: desc.GetOrgName(), + } + } + groupResults[i] = ev + return nil + }) + } - if ev == nil { - // no evaluation, skip - continue - } + if err := g.Wait(); err != nil { + return nil, err + } - // Assign group reference to this evaluation - ev.GroupReference = &api.PolicyEvaluation_Reference{ - Name: group.GetMetadata().GetName(), - Digest: desc.GetDigest(), - Uri: desc.GetURI(), - OrgName: desc.GetOrgName(), + for _, ev := range groupResults { + if ev != nil { + result = append(result, ev) } - result = append(result, ev) } } @@ -138,44 +153,59 @@ func (pgv *PolicyGroupVerifier) VerifyStatement(ctx context.Context, statement * return nil, NewPolicyError(err) } - for _, attachment := range group.GetSpec().GetPolicies().GetAttestation() { - // Check if policy should be skipped - skip, policyName, err := pgv.shouldSkipPolicy(ctx, attachment, groupAtt.GetSkip()) - if err != nil { - return nil, NewPolicyError(fmt.Errorf("failed to check if policy should be skipped: %w", err)) - } - - if skip { - pgv.logger.Debug().Str("policy", policyName).Msg("skipping attestation policy per skip list") - continue - } - - material, err := protojson.Marshal(statement) - if err != nil { - return nil, NewPolicyError(err) - } + // Marshal statement once for all policies in this group + statementJSON, err := protojson.Marshal(statement) + if err != nil { + return nil, NewPolicyError(err) + } - ev, err := pgv.evaluatePolicyAttachment(ctx, applyGroupGate(attachment, groupAtt), material, - &evalOpts{kind: v1.CraftingSchema_Material_ATTESTATION, bindings: groupArgs}, - ) - if err != nil { - return nil, NewPolicyError(err) - } + attestationPolicies := group.GetSpec().GetPolicies().GetAttestation() + groupResults := make([]*api.PolicyEvaluation, len(attestationPolicies)) + g, gCtx := errgroup.WithContext(ctx) + g.SetLimit(pgv.maxConcurrency) + + for i, attachment := range attestationPolicies { + g.Go(func() error { + // Check if policy should be skipped + skip, policyName, err := pgv.shouldSkipPolicy(gCtx, attachment, groupAtt.GetSkip()) + if err != nil { + return NewPolicyError(fmt.Errorf("failed to check if policy should be skipped: %w", err)) + } + + if skip { + pgv.logger.Debug().Str("policy", policyName).Msg("skipping attestation policy per skip list") + return nil + } + + ev, err := pgv.evaluatePolicyAttachment(gCtx, applyGroupGate(attachment, groupAtt), statementJSON, + &evalOpts{kind: v1.CraftingSchema_Material_ATTESTATION, bindings: groupArgs}, + ) + if err != nil { + return NewPolicyError(err) + } + + if ev != nil { + // Assign group reference to this evaluation + ev.GroupReference = &api.PolicyEvaluation_Reference{ + Name: group.GetMetadata().GetName(), + Digest: desc.GetDigest(), + Uri: desc.GetURI(), + OrgName: desc.GetOrgName(), + } + } + groupResults[i] = ev + return nil + }) + } - if ev == nil { - // no evaluation, skip - continue - } + if err := g.Wait(); err != nil { + return nil, err + } - // Assign group reference to this evaluation - ev.GroupReference = &api.PolicyEvaluation_Reference{ - Name: group.GetMetadata().GetName(), - Digest: desc.GetDigest(), - Uri: desc.GetURI(), - OrgName: desc.GetOrgName(), + for _, ev := range groupResults { + if ev != nil { + result = append(result, ev) } - - result = append(result, ev) } }