diff --git a/README.md b/README.md index 301eb49c..831ef757 100644 --- a/README.md +++ b/README.md @@ -58,9 +58,30 @@ export GOPROXY=http://cachew.example.com/gomod,direct gomod { proxy = "https://proxy.golang.org" private-paths = ["github.com/myorg/*"] + + package-policy { + socket { + api-url = "https://api.socket.dev" + organization = "my-socket-org" + token = "${SOCKET_SECURITY_API_TOKEN}" + } + } } ``` +When `package-policy` is configured, Cachew evaluates the PURL for each cold, +canonical-version public module before it contacts the Go module origin. Branch +and revision `.info` queries pass through so the Go proxy can resolve them; the +resulting canonical version's module files are evaluated before download. Cached +module files bypass the check. The `socket` provider sends the PURL to Socket; +modules matching `private-paths` are not sent. + +The warm-path cache probe does not read or backfill the module body. It also +disables origin fallback for that request. If the object is evicted between the +probe and goproxy's body read, goproxy returns its temporary `404` rather than +fetching an unevaluated body; a later client retry performs a normal miss and +policy evaluation. + ### Hermit Caches [Hermit](https://cashapp.github.io/hermit/) package downloads. GitHub release URLs are automatically routed through the `github-releases` strategy. @@ -95,6 +116,14 @@ codeartifact "example-111122223333.d.codeartifact.us-east-1.amazonaws.com" { domain-owner = "111122223333" region = "us-east-1" role-arn = "arn:aws:iam::111122223333:role/cachew-codeartifact-read" + + package-policy { + socket { + api-url = "https://api.socket.dev" + organization = "my-socket-org" + token = "${SOCKET_SECURITY_API_TOKEN}" + } + } } ``` @@ -106,6 +135,69 @@ The role needs `codeartifact:GetAuthorizationToken` and its underlying principal needs `sts:GetServiceBearerToken`. Repository read permissions remain governed by the CodeArtifact resource policy. +The optional `package-policy` block is a provider-independent package-admission +interface based on standard Package URLs (PURLs). Its `socket` provider checks +cold npm and PyPI artifact PURLs with the configured organization's [Socket +package policy](https://docs.socket.dev/reference/batchpackagefetchbyorg) before +Cachew mints a CodeArtifact token or contacts the repository. Another provider +can implement the same PURL-to-decision interface without changing the +CodeArtifact or Go module strategies. + +For the Socket provider, a policy action of `error`, or a package Socket cannot +resolve, returns `403`. Pending analysis or an unavailable Socket API returns +`503` with `Retry-After`; these outcomes fail closed and never fall through to +CodeArtifact. Cache hits do not recheck the policy because Cachew only admits +complete, origin-declared immutable bodies. + +A later policy change does not automatically invalidate an admitted object. +Cachew's generic `delete` operation accepts one exact cache key, but CodeArtifact +can store multiple representations of one URL. The unhashed key material is the +origin URL followed by these optional lines, in this order: + +```text +https://codeartifact.example.com/npm/repository/package/-/package-1.0.0.tgz +Accept= +Accept-Encoding= +``` + +`cachew delete codeartifact ` hashes that material and deletes the +matching object from every backend configured in that Cachew deployment. Run it +against every deployment that may hold a local tier. A separate key exists for +every observed header combination, and Cachew cannot currently list or delete +keys by package path or PURL. The command is therefore a complete targeted purge +only when every request variant is known. Otherwise operators must clear every +backing tier with deployment-specific tooling or wait for the origin TTL; Cachew +does not currently provide a reliable targeted purge for that incident. + +Socket receives the public ecosystem, package name, and version from the +CodeArtifact request path. Cachew does not send package contents, CodeArtifact +credentials, repository names, or AWS identity to Socket. Operators should +still treat the package name and version as data crossing from their +CodeArtifact environment to Socket's SaaS boundary. + +The Socket token needs only the `packages:list` scope. Keep it out of the HCL +file by using an environment placeholder as shown above and inject +`SOCKET_SECURITY_API_TOKEN` into the Cachew container from the deployment's +secret manager. For example, Kubernetes can source the environment variable +from a `Secret` with `env[].valueFrom.secretKeyRef`; the secret does not need to +be exposed to package-manager clients. + +Policy outcomes and API latency are exported as +`cachew.package_policy.evaluations_total` and +`cachew.package_policy.evaluation_duration_seconds`. The evaluation counter has +bounded provider and outcome attributes (`allow`, `deny`, `pending`, +`unavailable`, or `not_applicable`); package names and versions are not metric +labels. The latency histogram covers actual provider evaluations. Unsupported +ecosystems, non-package metadata or query paths, and excluded private Go modules +record `not_applicable` so gaps in enforcement coverage remain visible. Metadata +GETs can dominate that outcome, so dashboards should chart it separately and +exclude it from allow/deny availability ratios. `HEAD` requests are not counted +because they cannot admit a package body. + +Concurrent requests for the same PURL share one in-flight provider call. Cachew +discards the decision after that call completes: a later cold request performs a +fresh evaluation, so coalescing does not delay a changed Socket verdict. + Cachew checks its cache for every full CodeArtifact `GET` without a query string, range, or encoded path separator. On a miss, it stores only a successful, complete response with a positive shared freshness lifetime and an origin policy diff --git a/cachew.hcl b/cachew.hcl index a9265f82..c2f525df 100644 --- a/cachew.hcl +++ b/cachew.hcl @@ -11,6 +11,13 @@ # domain-owner = "111122223333" # region = "us-east-1" # role-arn = "arn:aws:iam::111122223333:role/cachew-codeartifact-read" +# package-policy { +# socket { +# api-url = "https://api.socket.dev" +# organization = "my-socket-org" +# token = "${SOCKET_SECURITY_API_TOKEN}" +# } +# } # } state = "./state" @@ -70,6 +77,13 @@ strategy github-releases { strategy gomod { proxy = "https://proxy.golang.org" + # package-policy { + # socket { + # api-url = "https://api.socket.dev" + # organization = "my-socket-org" + # token = "${SOCKET_SECURITY_API_TOKEN}" + # } + # } } strategy hermit { } diff --git a/internal/packagepolicy/config_test.go b/internal/packagepolicy/config_test.go new file mode 100644 index 00000000..dbb3796c --- /dev/null +++ b/internal/packagepolicy/config_test.go @@ -0,0 +1,49 @@ +package packagepolicy_test + +import ( + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "github.com/alecthomas/hcl/v2" + + "github.com/block/cachew/internal/packagepolicy" +) + +type policyConfigEnvelope struct { + PackagePolicy *packagepolicy.Config `hcl:"package-policy,block,optional"` +} + +func TestPackagePolicyConfigRoundTripsThroughHCL(t *testing.T) { + input := []byte(` +package-policy { + socket { + api-url = "https://socket.example.com" + organization = "example-org" + token = "test-token" + timeout = "45s" + } +} +`) + var config policyConfigEnvelope + assert.NoError(t, hcl.Unmarshal(input, &config)) + assert.NotZero(t, config.PackagePolicy) + assert.Equal(t, &packagepolicy.SocketConfig{ + APIURL: "https://socket.example.com", + Organization: "example-org", + Token: "test-token", + Timeout: 45 * time.Second, + }, config.PackagePolicy.Socket) + + encoded, err := hcl.Marshal(&config) + assert.NoError(t, err) + var roundTripped policyConfigEnvelope + assert.NoError(t, hcl.Unmarshal(encoded, &roundTripped)) + assert.Equal(t, config, roundTripped) +} + +func TestPackagePolicyConfigIsOptional(t *testing.T) { + var config policyConfigEnvelope + assert.NoError(t, hcl.Unmarshal(nil, &config)) + assert.Zero(t, config.PackagePolicy) +} diff --git a/internal/packagepolicy/evaluator.go b/internal/packagepolicy/evaluator.go new file mode 100644 index 00000000..1e25dade --- /dev/null +++ b/internal/packagepolicy/evaluator.go @@ -0,0 +1,42 @@ +// Package packagepolicy evaluates package URLs before Cachew fetches package bodies. +package packagepolicy + +import ( + "context" + + "github.com/alecthomas/errors" +) + +// Config selects and configures one package policy provider. +type Config struct { + Socket *SocketConfig `hcl:"socket,block,optional" help:"Socket organization policy provider."` +} + +// Verdict is the policy result for a package URL. +type Verdict string + +const ( + VerdictAllow Verdict = "allow" + VerdictDeny Verdict = "deny" + VerdictPending Verdict = "pending" +) + +// Decision is an aggregated package policy result. +type Decision struct { + Verdict Verdict + Reasons []string +} + +// Evaluator checks package URLs against a package policy. +type Evaluator interface { + Evaluate(context.Context, string) (Decision, error) + ObserveNotApplicable(context.Context) +} + +// New creates the configured package policy evaluator. +func New(config Config) (Evaluator, error) { + if config.Socket == nil { + return nil, errors.New("package policy: provider is required") + } + return newSocketEvaluator(*config.Socket, false) +} diff --git a/internal/packagepolicy/http.go b/internal/packagepolicy/http.go new file mode 100644 index 00000000..b748bdc9 --- /dev/null +++ b/internal/packagepolicy/http.go @@ -0,0 +1,54 @@ +package packagepolicy + +import ( + "net/http" + "regexp" + "strings" +) + +var safeReasonPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,64}$`) + +// AllowRequest reports whether an HTTP package request may continue and writes a +// fail-closed response when the decision is not an allow. +func AllowRequest(w http.ResponseWriter, decision Decision, err error) bool { + if err != nil { + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("X-Cachew-Package-Policy", "unavailable") + http.Error(w, "Package security policy unavailable", http.StatusServiceUnavailable) + return false + } + switch decision.Verdict { + case VerdictAllow: + return true + case VerdictDeny: + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("X-Cachew-Package-Policy", "deny") + message := "Package denied by security policy" + if reasons := safeReasons(decision.Reasons); len(reasons) > 0 { + message += ": " + strings.Join(reasons, ", ") + } + http.Error(w, message, http.StatusForbidden) + return false + case VerdictPending: + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("X-Cachew-Package-Policy", "pending") + w.Header().Set("Retry-After", "5") + http.Error(w, "Package security analysis pending", http.StatusServiceUnavailable) + return false + default: + w.Header().Set("Cache-Control", "no-store") + w.Header().Set("X-Cachew-Package-Policy", "unavailable") + http.Error(w, "Package security policy unavailable", http.StatusServiceUnavailable) + return false + } +} + +func safeReasons(reasons []string) []string { + safe := make([]string, 0, len(reasons)) + for _, reason := range reasons { + if safeReasonPattern.MatchString(reason) { + safe = append(safe, reason) + } + } + return safe +} diff --git a/internal/packagepolicy/metrics.go b/internal/packagepolicy/metrics.go new file mode 100644 index 00000000..b5d3082f --- /dev/null +++ b/internal/packagepolicy/metrics.go @@ -0,0 +1,63 @@ +package packagepolicy + +import ( + "context" + "time" + + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + + cachewmetrics "github.com/block/cachew/internal/metrics" +) + +type metricRecorder interface { + record(context.Context, Decision, error, time.Duration) + recordNotApplicable(context.Context) +} + +func (m *clientMetrics) recordNotApplicable(ctx context.Context) { + m.evaluations.Add(ctx, 1, metric.WithAttributes( + attribute.String("provider", m.provider), + attribute.String("outcome", "not_applicable"), + )) +} + +type clientMetrics struct { + provider string + evaluations metric.Int64Counter + duration metric.Float64Histogram +} + +func newMetrics(provider string) *clientMetrics { + meter := otel.Meter("cachew.package_policy") + return &clientMetrics{ + provider: provider, + evaluations: cachewmetrics.NewMetric[metric.Int64Counter]( + meter, + "cachew.package_policy.evaluations_total", + "{evaluations}", + "Package policy evaluations by provider and outcome", + ), + duration: cachewmetrics.NewHistogram( + meter, + "cachew.package_policy.evaluation_duration_seconds", + "s", + "Package policy evaluation duration by provider and outcome", + cachewmetrics.LatencyBuckets(), + ), + } +} + +func (m *clientMetrics) record(ctx context.Context, decision Decision, err error, duration time.Duration) { + outcome := string(decision.Verdict) + if err != nil || outcome == "" { + outcome = "unavailable" + } + attrs := metric.WithAttributes( + attribute.String("provider", m.provider), + attribute.String("outcome", outcome), + ) + m.evaluations.Add(ctx, 1, attrs) + m.duration.Record(ctx, duration.Seconds(), attrs) +} diff --git a/internal/packagepolicy/metrics_test.go b/internal/packagepolicy/metrics_test.go new file mode 100644 index 00000000..e3fc04a9 --- /dev/null +++ b/internal/packagepolicy/metrics_test.go @@ -0,0 +1,36 @@ +package packagepolicy //nolint:testpackage // White-box coverage is required for metric recorder injection. + +import ( + "context" + "sync/atomic" + "testing" + "time" + + "github.com/alecthomas/assert/v2" +) + +type recordingMetrics struct { + evaluations atomic.Int32 + notApplicable atomic.Int32 + recorded chan struct{} +} + +func (r *recordingMetrics) record(context.Context, Decision, error, time.Duration) { + r.evaluations.Add(1) + if r.recorded != nil { + r.recorded <- struct{}{} + } +} + +func (r *recordingMetrics) recordNotApplicable(context.Context) { + r.notApplicable.Add(1) +} + +func TestObserveNotApplicableRecordsProviderMetric(t *testing.T) { + metrics := &recordingMetrics{} + evaluator := &socketEvaluator{metrics: metrics} + + evaluator.ObserveNotApplicable(t.Context()) + + assert.Equal(t, int32(1), metrics.notApplicable.Load()) +} diff --git a/internal/packagepolicy/purl.go b/internal/packagepolicy/purl.go new file mode 100644 index 00000000..a98dfbbd --- /dev/null +++ b/internal/packagepolicy/purl.go @@ -0,0 +1,111 @@ +package packagepolicy + +import ( + "net/url" + "regexp" + "slices" + "strings" + + "golang.org/x/mod/module" +) + +var pypiNormalizationPattern = regexp.MustCompile(`[-_.]+`) + +// PackageURLForCodeArtifact derives a PURL from an immutable npm or PyPI CodeArtifact asset path. +func PackageURLForCodeArtifact(path string) (string, bool) { + parts, ok := decodedPathParts(path) + if !ok || len(parts) < 5 { + return "", false + } + switch parts[0] { + case "npm": + return npmPackageURL(parts) + case "pypi": + return pypiPackageURL(parts) + default: + return "", false + } +} + +func npmPackageURL(parts []string) (string, bool) { + var namespace, name, separator, filename string + switch { + case len(parts) == 5: + name, separator, filename = parts[2], parts[3], parts[4] + case len(parts) == 6 && strings.HasPrefix(parts[2], "@"): + namespace, name, separator, filename = parts[2], parts[3], parts[4], parts[5] + default: + return "", false + } + if separator != "-" || !strings.HasSuffix(filename, ".tgz") { + return "", false + } + version, ok := strings.CutPrefix(strings.TrimSuffix(filename, ".tgz"), name+"-") + if !ok || version == "" || name == "" { + return "", false + } + if namespace != "" { + return "pkg:npm/" + escapePURLSegment(namespace) + "/" + escapePURLSegment(name) + "@" + escapePURLSegment(version), true + } + return "pkg:npm/" + escapePURLSegment(name) + "@" + escapePURLSegment(version), true +} + +func pypiPackageURL(parts []string) (string, bool) { + if len(parts) != 6 || parts[2] != "simple" || parts[3] == "" || parts[4] == "" || parts[5] == "" { + return "", false + } + name := pypiNormalizationPattern.ReplaceAllString(strings.ToLower(parts[3]), "-") + return "pkg:pypi/" + escapePURLSegment(name) + "@" + escapePURLSegment(parts[4]), true +} + +// PackageURLForGoModule derives a PURL from a versioned Go module proxy path. +func PackageURLForGoModule(path string) (string, bool) { + path = strings.TrimPrefix(path, "/") + modulePath, asset, ok := strings.Cut(path, "/@v/") + if !ok || modulePath == "" { + return "", false + } + var escapedVersion string + for _, suffix := range []string{".info", ".mod", ".zip"} { + if version, found := strings.CutSuffix(asset, suffix); found { + escapedVersion = version + break + } + } + if escapedVersion == "" { + return "", false + } + name, err := module.UnescapePath(modulePath) + if err != nil { + return "", false + } + version, err := module.UnescapeVersion(escapedVersion) + if err != nil { + return "", false + } + if module.Check(name, version) != nil || version != module.CanonicalVersion(version) { + return "", false + } + return "pkg:golang/" + escapePURLPath(name) + "@" + escapePURLSegment(version), true +} + +func decodedPathParts(path string) ([]string, bool) { + parts := strings.Split(strings.Trim(path, "/"), "/") + if slices.Contains(parts, "") { + return nil, false + } + return parts, true +} + +func escapePURLPath(path string) string { + parts := strings.Split(path, "/") + for i, part := range parts { + parts[i] = escapePURLSegment(part) + } + return strings.Join(parts, "/") +} + +func escapePURLSegment(value string) string { + escaped := url.PathEscape(value) + return strings.ReplaceAll(escaped, "@", "%40") +} diff --git a/internal/packagepolicy/purl_test.go b/internal/packagepolicy/purl_test.go new file mode 100644 index 00000000..ac399fa6 --- /dev/null +++ b/internal/packagepolicy/purl_test.go @@ -0,0 +1,101 @@ +package packagepolicy_test + +import ( + "testing" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/internal/packagepolicy" +) + +func TestPackageURLForCodeArtifact(t *testing.T) { + tests := []struct { + name string + path string + purl string + ok bool + }{ + { + name: "npm package", + path: "/npm/repository/chromatitle-js/-/chromatitle-js-1.0.0.tgz", + purl: "pkg:npm/chromatitle-js@1.0.0", + ok: true, + }, + { + name: "scoped npm package", + path: "/npm/repository/@ctrl/tinycolor/-/tinycolor-4.1.1.tgz", + purl: "pkg:npm/%40ctrl/tinycolor@4.1.1", + ok: true, + }, + { + name: "PyPI wheel", + path: "/pypi/repository/simple/requests/2.32.3/requests-2.32.3-py3-none-any.whl", + purl: "pkg:pypi/requests@2.32.3", + ok: true, + }, + { + name: "PyPI normalized package", + path: "/pypi/repository/simple/My_Package/1.0.0/My_Package-1.0.0.tar.gz", + purl: "pkg:pypi/my-package@1.0.0", + ok: true, + }, + { + name: "double-encoded separator is not decoded twice", + path: "/npm/repository/package%2Fname/-/package%2Fname-1.0.0.tgz", + purl: "pkg:npm/package%252Fname@1.0.0", + ok: true, + }, + {name: "npm metadata", path: "/npm/repository/chromatitle-js", ok: false}, + {name: "PyPI metadata", path: "/pypi/repository/simple/requests/", ok: false}, + {name: "unsupported format", path: "/maven/repository/example.jar", ok: false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + purl, ok := packagepolicy.PackageURLForCodeArtifact(test.path) + assert.Equal(t, test.ok, ok) + assert.Equal(t, test.purl, purl) + }) + } +} + +func TestPackageURLForGoModule(t *testing.T) { + tests := []struct { + name string + path string + purl string + ok bool + }{ + { + name: "module zip", + path: "/github.com/pkg/errors/@v/v0.9.1.zip", + purl: "pkg:golang/github.com/pkg/errors@v0.9.1", + ok: true, + }, + { + name: "escaped uppercase module", + path: "/github.com/!azure/azure-sdk-for-go/@v/v1.2.3.mod", + purl: "pkg:golang/github.com/Azure/azure-sdk-for-go@v1.2.3", + ok: true, + }, + { + name: "canonical pseudo-version", + path: "/github.com/pkg/errors/@v/v0.0.0-20200101000000-abcdefabcdef.info", + purl: "pkg:golang/github.com/pkg/errors@v0.0.0-20200101000000-abcdefabcdef", + ok: true, + }, + {name: "branch query", path: "/github.com/pkg/errors/@v/master.info", ok: false}, + {name: "commit query", path: "/github.com/pkg/errors/@v/abcdef1234567890.info", ok: false}, + {name: "version list", path: "/github.com/pkg/errors/@v/list", ok: false}, + {name: "latest", path: "/github.com/pkg/errors/@latest", ok: false}, + {name: "invalid", path: "/not-a-module", ok: false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + purl, ok := packagepolicy.PackageURLForGoModule(test.path) + assert.Equal(t, test.ok, ok) + assert.Equal(t, test.purl, purl) + }) + } +} diff --git a/internal/packagepolicy/socket.go b/internal/packagepolicy/socket.go new file mode 100644 index 00000000..c62744e8 --- /dev/null +++ b/internal/packagepolicy/socket.go @@ -0,0 +1,272 @@ +package packagepolicy + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/url" + "regexp" + "slices" + "strconv" + "strings" + "time" + + "github.com/alecthomas/errors" + "golang.org/x/sync/singleflight" +) + +const ( + defaultAPIURL = "https://api.socket.dev" + defaultTimeout = 30 * time.Second + maxResponseBytes = 4 << 20 + maxResponseLineSize = 1 << 20 +) + +var organizationPattern = regexp.MustCompile(`^[A-Za-z0-9_-]+$`) + +// SocketConfig configures Socket's organization-scoped PURL evaluator. +type SocketConfig struct { + APIURL string `hcl:"api-url,optional" help:"Socket API origin." default:"https://api.socket.dev"` + Organization string `hcl:"organization" help:"Socket organization slug whose security policy is evaluated."` + Token string `hcl:"token" help:"Socket API token with packages:list scope. Use an environment variable placeholder rather than a literal secret."` + Timeout time.Duration `hcl:"timeout,optional" help:"Maximum time Socket may spend resolving and scanning a package." default:"30s"` +} + +type socketEvaluator struct { + endpoint *url.URL + token string + timeoutSec int + httpClient *http.Client + metrics metricRecorder + inflight singleflight.Group +} + +var _ Evaluator = (*socketEvaluator)(nil) + +// ObserveNotApplicable records a request that Cachew could not map to a PURL. +func (c *socketEvaluator) ObserveNotApplicable(ctx context.Context) { + c.metrics.recordNotApplicable(ctx) +} + +func newSocketEvaluator(config SocketConfig, allowHTTP bool) (*socketEvaluator, error) { + if config.APIURL == "" { + config.APIURL = defaultAPIURL + } + if config.Timeout == 0 { + config.Timeout = defaultTimeout + } + if !organizationPattern.MatchString(config.Organization) { + return nil, errors.New("socket policy: organization must be a non-empty slug") + } + if config.Token == "" { + return nil, errors.New("socket policy: token is required") + } + if config.Timeout < time.Second || config.Timeout > 20*time.Minute { + return nil, errors.New("socket policy: timeout must be between 1s and 20m") + } + + base, err := url.Parse(config.APIURL) + if err != nil { + return nil, errors.Wrap(err, "socket policy: parse API URL") + } + validScheme := base.Scheme == "https" || (allowHTTP && base.Scheme == "http") + if !validScheme || base.Host == "" || base.User != nil || base.RawQuery != "" || base.Fragment != "" || (base.Path != "" && base.Path != "/") { + return nil, errors.New("socket policy: API URL must be an HTTPS origin") + } + endpoint := base.JoinPath("v0", "orgs", config.Organization, "purl") + + return &socketEvaluator{ + endpoint: endpoint, + token: config.Token, + timeoutSec: int(config.Timeout / time.Second), + httpClient: &http.Client{ + Timeout: config.Timeout + 5*time.Second, + CheckRedirect: func(_ *http.Request, _ []*http.Request) error { + return http.ErrUseLastResponse + }, + }, + metrics: newMetrics("socket"), + }, nil +} + +// Evaluate returns the strictest policy result across every artifact Socket returns. +func (c *socketEvaluator) Evaluate(ctx context.Context, purl string) (Decision, error) { + resultCh := c.inflight.DoChan(purl, func() (any, error) { + requestCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), c.httpClient.Timeout) + defer cancel() + return c.evaluateProvider(requestCtx, purl) + }) + select { + case <-ctx.Done(): + return Decision{}, errors.Wrap(context.Cause(ctx), "socket policy: wait for shared evaluation") + case result := <-resultCh: + if result.Err != nil { + return Decision{}, result.Err + } + sharedDecision, ok := result.Val.(Decision) + if !ok { + return Decision{}, errors.New("socket policy: invalid shared evaluation") + } + return sharedDecision, nil + } +} + +func (c *socketEvaluator) evaluateProvider(ctx context.Context, purl string) (decision Decision, err error) { + started := time.Now() + defer func() { c.metrics.record(context.WithoutCancel(ctx), decision, err, time.Since(started)) }() + + body, err := json.Marshal(purlRequest{Components: []purlComponent{{PURL: purl}}}) + if err != nil { + return Decision{}, errors.Wrap(err, "socket policy: encode request") + } + + endpoint := *c.endpoint + query := endpoint.Query() + query.Set("alerts", "true") + query.Set("compact", "true") + query.Set("poll", "true") + query.Set("purlErrors", "false") + query.Set("timeoutSec", strconv.Itoa(c.timeoutSec)) + endpoint.RawQuery = query.Encode() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint.String(), bytes.NewReader(body)) + if err != nil { + return Decision{}, errors.Wrap(err, "socket policy: build request") + } + req.Header.Set("Accept", "application/x-ndjson") + req.Header.Set("Authorization", "Bearer "+c.token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("User-Agent", "cachew") + + resp, err := c.httpClient.Do(req) + if err != nil { + return Decision{}, errors.Wrap(err, "socket policy: request failed") + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + if _, copyErr := io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)); copyErr != nil { + return Decision{}, errors.New("socket policy: read error response") + } + return Decision{}, errors.Errorf("socket policy: API returned %s", resp.Status) + } + + limited := &io.LimitedReader{R: resp.Body, N: maxResponseBytes + 1} + decision, err = evaluateStream(limited, purl) + if err != nil { + return Decision{}, err + } + if limited.N == 0 { + return Decision{}, errors.New("socket policy: response exceeds size limit") + } + return decision, nil +} + +type purlRequest struct { + Components []purlComponent `json:"components"` +} + +type purlComponent struct { + PURL string `json:"purl"` +} + +type apiArtifact struct { + StreamType string `json:"_type"` + InputPURL string `json:"inputPurl"` + Type string `json:"type"` + Name string `json:"name"` + Version string `json:"version"` + Alerts []alert `json:"alerts"` +} + +type alert struct { + Type string `json:"type"` + Action string `json:"action"` +} + +type evaluationState struct { + resolved bool + pending bool + unscanned bool + denied map[string]struct{} + records int +} + +func evaluateStream(r io.Reader, requestedPURL string) (Decision, error) { + scanner := bufio.NewScanner(r) + scanner.Buffer(make([]byte, 64<<10), maxResponseLineSize) + state := evaluationState{denied: make(map[string]struct{})} + + for scanner.Scan() { + if strings.TrimSpace(scanner.Text()) == "" { + continue + } + state.records++ + var decoded apiArtifact + if err := json.Unmarshal(scanner.Bytes(), &decoded); err != nil { + return Decision{}, errors.Wrap(err, "socket policy: decode response") + } + if err := state.add(decoded, requestedPURL); err != nil { + return Decision{}, err + } + } + if err := scanner.Err(); err != nil { + return Decision{}, errors.Wrap(err, "socket policy: read response") + } + return state.decision() +} + +func (s *evaluationState) add(artifact apiArtifact, requestedPURL string) error { + if artifact.StreamType != "" { + return errors.Errorf("socket policy: unexpected stream record %q", artifact.StreamType) + } + if artifact.InputPURL != "" && artifact.InputPURL != requestedPURL { + return errors.New("socket policy: response PURL does not match request") + } + if artifact.Type != "" && artifact.Name != "" && artifact.Version != "" { + s.resolved = true + } + for _, alert := range artifact.Alerts { + switch alert.Type { + case "pendingScan": + s.pending = true + continue + case "notFound": + s.unscanned = true + continue + } + switch alert.Action { + case "error": + s.denied[alert.Type] = struct{}{} + case "ignore", "monitor", "warn": + case "": + return errors.Errorf("socket policy: alert %q has no policy action", alert.Type) + default: + return errors.Errorf("socket policy: unsupported policy action %q", alert.Action) + } + } + return nil +} + +func (s *evaluationState) decision() (Decision, error) { + if s.records == 0 { + return Decision{}, errors.New("socket policy: empty response") + } + if len(s.denied) > 0 { + reasons := make([]string, 0, len(s.denied)) + for reason := range s.denied { + reasons = append(reasons, reason) + } + slices.Sort(reasons) + return Decision{Verdict: VerdictDeny, Reasons: reasons}, nil + } + if s.pending { + return Decision{Verdict: VerdictPending, Reasons: []string{"pendingScan"}}, nil + } + if s.unscanned || !s.resolved { + return Decision{Verdict: VerdictDeny, Reasons: []string{"notFound"}}, nil + } + return Decision{Verdict: VerdictAllow}, nil +} diff --git a/internal/packagepolicy/socket_test.go b/internal/packagepolicy/socket_test.go new file mode 100644 index 00000000..93c40ec8 --- /dev/null +++ b/internal/packagepolicy/socket_test.go @@ -0,0 +1,274 @@ +package packagepolicy //nolint:testpackage // White-box coverage is required for HTTP transport injection. + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + "github.com/alecthomas/errors" +) + +type roundTripperFunc func(*http.Request) (*http.Response, error) + +func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { + return f(r) +} + +const ( + testOrganization = "example-org" + testToken = "socket-test-token" + testPURL = "pkg:npm/chromatitle-js@1.0.0" + testAllowResponse = `{"type":"npm","name":"chromatitle-js","version":"1.0.0","alerts":[]}` +) + +type blockedSocketTestHarness struct { + client *socketEvaluator + requests atomic.Int32 + started chan struct{} + release chan struct{} +} + +func newBlockedSocketTestHarness(t *testing.T, metrics metricRecorder) *blockedSocketTestHarness { + t.Helper() + harness := &blockedSocketTestHarness{ + started: make(chan struct{}), + release: make(chan struct{}), + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + if harness.requests.Add(1) == 1 { + close(harness.started) + } + <-harness.release + _, _ = w.Write([]byte(testAllowResponse)) + })) + t.Cleanup(server.Close) + client, err := newSocketEvaluator(SocketConfig{ + APIURL: server.URL, + Organization: testOrganization, + Token: testToken, + }, true) + assert.NoError(t, err) + client.metrics = metrics + harness.client = client + return harness +} + +func TestNewSelectsSocketProvider(t *testing.T) { + evaluator, err := New(Config{Socket: &SocketConfig{Organization: testOrganization, Token: testToken}}) + assert.NoError(t, err) + assert.NotZero(t, evaluator) + + _, err = New(Config{}) + assert.Error(t, err) +} + +func TestClientEvaluatesOrganizationPolicy(t *testing.T) { + tests := []struct { + name string + response string + verdict Verdict + reasons []string + }{ + { + name: "allows policy-approved package", + response: `{"type":"npm","name":"lodash","version":"4.17.21","alerts":[{"type":"unpopularPackage","action":"monitor"}]}`, + verdict: VerdictAllow, + }, + { + name: "denies policy error", + response: `{"type":"npm","name":"chromatitle-js","version":"1.0.0","alerts":[{"type":"malware","action":"error"}]}`, + verdict: VerdictDeny, + reasons: []string{"malware"}, + }, + { + name: "waits for pending analysis", + response: `{"type":"npm","name":"new-package","version":"1.0.0","alerts":[{"type":"pendingScan","action":"ignore"}]}`, + verdict: VerdictPending, + reasons: []string{"pendingScan"}, + }, + { + name: "denies unscanned package", + response: `{"type":"npm","name":"unknown-package","version":"1.0.0","alerts":[{"type":"notFound","action":"ignore"}]}`, + verdict: VerdictDeny, + reasons: []string{"notFound"}, + }, + { + name: "denies if any package artifact is blocked", + response: `{"type":"pypi","name":"example","version":"1.0.0","release":"py3-none-any-whl","alerts":[]} +{"type":"pypi","name":"example","version":"1.0.0","release":"tar-gz","alerts":[{"type":"malware","action":"error"}]}`, + verdict: VerdictDeny, + reasons: []string{"malware"}, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, http.MethodPost, r.Method) + assert.Equal(t, "/v0/orgs/example-org/purl", r.URL.Path) + assert.Equal(t, "Bearer "+testToken, r.Header.Get("Authorization")) + assert.Equal(t, "true", r.URL.Query().Get("alerts")) + assert.Equal(t, "true", r.URL.Query().Get("compact")) + assert.Equal(t, "true", r.URL.Query().Get("poll")) + assert.Equal(t, "false", r.URL.Query().Get("purlErrors")) + assert.Equal(t, "30", r.URL.Query().Get("timeoutSec")) + + var body struct { + Components []struct { + PURL string `json:"purl"` + } `json:"components"` + } + assert.NoError(t, json.NewDecoder(r.Body).Decode(&body)) + assert.Equal(t, testPURL, body.Components[0].PURL) + w.Header().Set("Content-Type", "application/x-ndjson") + _, _ = w.Write([]byte(test.response + "\n")) + })) + t.Cleanup(server.Close) + + client, err := newSocketEvaluator(SocketConfig{ + APIURL: server.URL, + Organization: testOrganization, + Token: testToken, + Timeout: 30 * time.Second, + }, true) + assert.NoError(t, err) + + decision, err := client.Evaluate(context.Background(), testPURL) + assert.NoError(t, err) + assert.Equal(t, test.verdict, decision.Verdict) + assert.Equal(t, test.reasons, decision.Reasons) + }) + } +} + +func TestClientCoalescesConcurrentEvaluations(t *testing.T) { + metrics := &recordingMetrics{} + harness := newBlockedSocketTestHarness(t, metrics) + + type result struct { + decision Decision + err error + } + const callers = 16 + begin := make(chan struct{}) + results := make(chan result, callers) + var ready sync.WaitGroup + ready.Add(callers) + for range callers { + go func() { + ready.Done() + <-begin + decision, err := harness.client.Evaluate(t.Context(), testPURL) + results <- result{decision: decision, err: err} + }() + } + ready.Wait() + close(begin) + <-harness.started + time.Sleep(25 * time.Millisecond) + requestCount := harness.requests.Load() + close(harness.release) + assert.Equal(t, int32(1), requestCount) + + for range callers { + evaluation := <-results + assert.NoError(t, evaluation.err) + assert.Equal(t, VerdictAllow, evaluation.decision.Verdict) + } + assert.Equal(t, int32(1), metrics.evaluations.Load()) + + decision, err := harness.client.Evaluate(t.Context(), testPURL) + assert.NoError(t, err) + assert.Equal(t, VerdictAllow, decision.Verdict) + assert.Equal(t, int32(2), harness.requests.Load()) + assert.Equal(t, int32(2), metrics.evaluations.Load()) +} + +func TestClientRecordsSharedEvaluationAfterCallerCancellation(t *testing.T) { + metrics := &recordingMetrics{recorded: make(chan struct{}, 1)} + harness := newBlockedSocketTestHarness(t, metrics) + + ctx, cancel := context.WithCancel(t.Context()) + result := make(chan error, 1) + go func() { + _, err := harness.client.Evaluate(ctx, testPURL) + result <- err + }() + <-harness.started + cancel() + assert.True(t, errors.Is(<-result, context.Canceled)) + assert.Equal(t, int32(0), metrics.evaluations.Load()) + close(harness.release) + <-metrics.recorded + assert.Equal(t, int32(1), metrics.evaluations.Load()) +} + +func TestClientFailsClosedOnInvalidResponses(t *testing.T) { + tests := []struct { + name string + statusCode int + response string + }{ + {name: "upstream failure", statusCode: http.StatusTooManyRequests}, + {name: "malformed stream", statusCode: http.StatusOK, response: `{"type":`}, + {name: "empty stream", statusCode: http.StatusOK}, + {name: "unknown policy action", statusCode: http.StatusOK, response: `{"type":"npm","name":"example","version":"1.0.0","alerts":[{"type":"malware","action":"future-action"}]}`}, + {name: "mismatched PURL", statusCode: http.StatusOK, response: `{"inputPurl":"pkg:npm/other@1.0.0","type":"npm","name":"other","version":"1.0.0","alerts":[]}`}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(test.statusCode) + _, _ = w.Write([]byte(test.response)) + })) + t.Cleanup(server.Close) + client, err := newSocketEvaluator(SocketConfig{APIURL: server.URL, Organization: testOrganization, Token: testToken}, true) + assert.NoError(t, err) + + _, err = client.Evaluate(context.Background(), testPURL) + assert.Error(t, err) + }) + } +} + +func TestClientPreservesTransportFailureForCallerLogging(t *testing.T) { + transportErr := errors.New("dial Socket API") + client, err := newSocketEvaluator(SocketConfig{ + APIURL: "https://socket.example.com", + Organization: testOrganization, + Token: testToken, + }, false) + assert.NoError(t, err) + client.httpClient.Transport = roundTripperFunc(func(_ *http.Request) (*http.Response, error) { + return nil, transportErr + }) + + _, err = client.Evaluate(context.Background(), testPURL) + assert.True(t, errors.Is(err, transportErr)) +} + +func TestClientDoesNotForwardTokenAcrossRedirects(t *testing.T) { + redirectRequests := 0 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + redirectRequests++ + _, _ = w.Write([]byte(`{"type":"npm","name":"example","version":"1.0.0","alerts":[]}`)) + })) + t.Cleanup(target.Close) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Redirect(w, &http.Request{}, target.URL, http.StatusFound) + })) + t.Cleanup(server.Close) + client, err := newSocketEvaluator(SocketConfig{APIURL: server.URL, Organization: testOrganization, Token: testToken}, true) + assert.NoError(t, err) + + _, err = client.Evaluate(context.Background(), testPURL) + assert.Error(t, err) + assert.Equal(t, 0, redirectRequests) +} diff --git a/internal/strategy/codeartifact.go b/internal/strategy/codeartifact.go index 8218811a..3209d33c 100644 --- a/internal/strategy/codeartifact.go +++ b/internal/strategy/codeartifact.go @@ -15,32 +15,35 @@ import ( "github.com/block/cachew/internal/cache" "github.com/block/cachew/internal/httputil" "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/packagepolicy" ) const codeArtifactUsername = "aws" // CodeArtifactConfig configures an authenticated, read-only CodeArtifact origin. type CodeArtifactConfig struct { - Target string `hcl:"target,label" help:"The CodeArtifact origin URL to proxy requests to."` - ProxyBaseURL string `hcl:"proxy-base-url" help:"The public Cachew origin used when rewriting package metadata URLs."` - Domain string `hcl:"domain" help:"The CodeArtifact domain name."` - DomainOwner string `hcl:"domain-owner" help:"The AWS account ID that owns the CodeArtifact domain."` - Region string `hcl:"region" help:"The AWS region containing the CodeArtifact domain."` - RoleARN string `hcl:"role-arn" help:"The read-only IAM role to assume when minting CodeArtifact tokens."` + Target string `hcl:"target,label" help:"The CodeArtifact origin URL to proxy requests to."` + ProxyBaseURL string `hcl:"proxy-base-url" help:"The public Cachew origin used when rewriting package metadata URLs."` + Domain string `hcl:"domain" help:"The CodeArtifact domain name."` + DomainOwner string `hcl:"domain-owner" help:"The AWS account ID that owns the CodeArtifact domain."` + Region string `hcl:"region" help:"The AWS region containing the CodeArtifact domain."` + RoleARN string `hcl:"role-arn" help:"The read-only IAM role to assume when minting CodeArtifact tokens."` + PackagePolicy *packagepolicy.Config `hcl:"package-policy,block,optional" help:"Optional package security policy enforced before cold npm and PyPI artifact reads."` } // CodeArtifact caches origin-declared immutable responses and passes all other // authenticated reads through. type CodeArtifact struct { - target *url.URL - proxyBase *url.URL - prefix string - tokens codeArtifactTokenSource - cache cache.Cache - client *http.Client - logger *slog.Logger - metric codeArtifactMetricRecorder - fills singleflight.Group + target *url.URL + proxyBase *url.URL + prefix string + tokens codeArtifactTokenSource + cache cache.Cache + client *http.Client + logger *slog.Logger + metric codeArtifactMetricRecorder + packagePolicy packagepolicy.Evaluator + fills singleflight.Group } var _ Strategy = (*CodeArtifact)(nil) @@ -64,7 +67,17 @@ func NewCodeArtifact(ctx context.Context, config CodeArtifactConfig, configuredC if err != nil { return nil, err } - return newCodeArtifact(ctx, config, mux, tokens, configuredCache, false) + strategy, err := newCodeArtifact(ctx, config, mux, tokens, configuredCache, false) + if err != nil { + return nil, err + } + if config.PackagePolicy != nil { + strategy.packagePolicy, err = packagepolicy.New(*config.PackagePolicy) + if err != nil { + return nil, errors.Wrap(err, "create package policy") + } + } + return strategy, nil } func newCodeArtifact( @@ -187,14 +200,14 @@ func (c *CodeArtifact) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func (c *CodeArtifact) serveOrigin(w http.ResponseWriter, r *http.Request, mode codeArtifactCacheMode) { + if !c.allowPackage(w, r) { + return + } rewriteMetadata := shouldRewriteCodeArtifactMetadata(c.originURL(r).Path) - token, err := c.tokens.Token(r.Context(), 0) - if err != nil { - c.metric.recordAuth(r.Context(), codeArtifactAuthFailure) - c.writeError(w, r, errors.Wrap(err, "obtain CodeArtifact authorization")) + token, authorized := c.authorizationToken(w, r) + if !authorized { return } - c.metric.recordAuth(r.Context(), token.event) resp, err := c.do(r, token.value) if err != nil { @@ -274,6 +287,33 @@ func (c *CodeArtifact) serveOrigin(w http.ResponseWriter, r *http.Request, mode } } +func (c *CodeArtifact) authorizationToken(w http.ResponseWriter, r *http.Request) (codeArtifactToken, bool) { + token, err := c.tokens.Token(r.Context(), 0) + if err != nil { + c.metric.recordAuth(r.Context(), codeArtifactAuthFailure) + c.writeError(w, r, errors.Wrap(err, "obtain CodeArtifact authorization")) + return codeArtifactToken{}, false + } + c.metric.recordAuth(r.Context(), token.event) + return token, true +} + +func (c *CodeArtifact) allowPackage(w http.ResponseWriter, r *http.Request) bool { + if c.packagePolicy == nil || r.Method != http.MethodGet { + return true + } + purl, ok := packagepolicy.PackageURLForCodeArtifact(c.originURL(r).Path) + if !ok { + c.packagePolicy.ObserveNotApplicable(r.Context()) + return true + } + decision, err := c.packagePolicy.Evaluate(r.Context(), purl) + if err != nil { + c.logger.ErrorContext(r.Context(), "Package policy evaluation failed", "error", err) + } + return packagepolicy.AllowRequest(w, decision, err) +} + func (c *CodeArtifact) rewriteOriginMetadata( resp *http.Response, headers http.Header, diff --git a/internal/strategy/codeartifact_test.go b/internal/strategy/codeartifact_test.go index 05492748..de1220f7 100644 --- a/internal/strategy/codeartifact_test.go +++ b/internal/strategy/codeartifact_test.go @@ -1,6 +1,7 @@ package strategy //nolint:testpackage // White-box coverage is required for token and transport injection. import ( + "bytes" "context" "encoding/base64" "fmt" @@ -23,6 +24,7 @@ import ( "github.com/block/cachew/internal/cache" "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/packagepolicy" ) const ( @@ -193,6 +195,121 @@ func (f codeArtifactRoundTripperFunc) RoundTrip(r *http.Request) (*http.Response return f(r) } +type recordingPackagePolicy struct { + decision packagepolicy.Decision + err error + purls []string + notApplicable int +} + +func (r *recordingPackagePolicy) Evaluate(_ context.Context, purl string) (packagepolicy.Decision, error) { + r.purls = append(r.purls, purl) + return r.decision, r.err +} + +func (r *recordingPackagePolicy) ObserveNotApplicable(context.Context) { + r.notApplicable++ +} + +func TestCodeArtifactRecordsUnsupportedPolicyRequest(t *testing.T) { + target, err := url.Parse("https://codeartifact.example.com") + assert.NoError(t, err) + policy := &recordingPackagePolicy{} + strategy := &CodeArtifact{ + target: target, + prefix: "/codeartifact.example.com", + packagePolicy: policy, + } + request := httptest.NewRequest( + http.MethodGet, + "/codeartifact.example.com/maven/repository/example.jar", + nil, + ) + + assert.True(t, strategy.allowPackage(httptest.NewRecorder(), request)) + assert.Equal(t, 1, policy.notApplicable) + assert.Equal(t, []string(nil), policy.purls) +} + +func TestCodeArtifactEnforcesPackagePolicyBeforeOriginAuthentication(t *testing.T) { + tests := []struct { + name string + decision packagepolicy.Decision + err error + statusCode int + policy string + }{ + { + name: "denied package", + decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictDeny, Reasons: []string{"malware"}}, + statusCode: http.StatusForbidden, + policy: "deny", + }, + { + name: "pending package", + decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictPending, Reasons: []string{"pendingScan"}}, + statusCode: http.StatusServiceUnavailable, + policy: "pending", + }, + { + name: "policy unavailable", + err: errors.New("Socket API unavailable"), + statusCode: http.StatusServiceUnavailable, + policy: "unavailable", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var logs bytes.Buffer + var originRequests atomic.Int32 + mux, originServer, tokenServer, strategy, ctx := newTestCachingCodeArtifact(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + originRequests.Add(1) + w.WriteHeader(http.StatusOK) + })) + strategy.logger = slog.New(slog.NewJSONHandler(&logs, nil)) + policy := &recordingPackagePolicy{decision: test.decision, err: test.err} + strategy.packagePolicy = policy + + w := httptest.NewRecorder() + path := "/npm/repository/chromatitle-js/-/chromatitle-js-1.0.0.tgz" + mux.ServeHTTP(w, httptest.NewRequest(http.MethodGet, codeArtifactPath(originServer, path), nil).WithContext(ctx)) + + assert.Equal(t, test.statusCode, w.Code) + assert.Equal(t, test.policy, w.Header().Get("X-Cachew-Package-Policy")) + assert.Equal(t, []string{"pkg:npm/chromatitle-js@1.0.0"}, policy.purls) + assert.Equal(t, 0, tokenServer.requestCount()) + assert.Equal(t, int32(0), originRequests.Load()) + if test.err != nil { + assert.Contains(t, logs.String(), test.err.Error()) + } + }) + } +} + +func TestCodeArtifactCachedPackageBypassesPackagePolicy(t *testing.T) { + var originRequests atomic.Int32 + origin := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + originRequests.Add(1) + w.Header().Set("Cache-Control", testCodeArtifactCacheControl) + _, _ = w.Write([]byte(testCodeArtifactBody)) + }) + mux, originServer, _, strategy, ctx := newTestCachingCodeArtifact(t, origin) + policy := &recordingPackagePolicy{decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictAllow}} + strategy.packagePolicy = policy + path := codeArtifactPath(originServer, "/npm/repository/lodash/-/lodash-4.17.21.tgz") + + for range 2 { + w := httptest.NewRecorder() + mux.ServeHTTP(w, httptest.NewRequest(http.MethodGet, path, nil).WithContext(ctx)) + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, testCodeArtifactBody, w.Body.String()) + } + + assert.Equal(t, []string{"pkg:npm/lodash@4.17.21"}, policy.purls) + assert.Equal(t, int32(1), originRequests.Load()) +} + func TestCodeArtifactSanitizesOriginTransportErrors(t *testing.T) { target, err := url.Parse("https://codeartifact.example.com") assert.NoError(t, err) diff --git a/internal/strategy/gomod/cacher.go b/internal/strategy/gomod/cacher.go index 467c8e84..35627ceb 100644 --- a/internal/strategy/gomod/cacher.go +++ b/internal/strategy/gomod/cacher.go @@ -15,6 +15,11 @@ type goproxyCacher struct { cache cache.Cache } +func (g *goproxyCacher) Exists(ctx context.Context, name string) bool { + _, err := g.cache.Stat(ctx, cache.NewKey(name)) + return err == nil +} + func (g *goproxyCacher) Get(ctx context.Context, name string) (io.ReadCloser, error) { key := cache.NewKey(name) diff --git a/internal/strategy/gomod/fetcher.go b/internal/strategy/gomod/fetcher.go index c3d98c8f..40936aef 100644 --- a/internal/strategy/gomod/fetcher.go +++ b/internal/strategy/gomod/fetcher.go @@ -31,7 +31,11 @@ func NewCompositeFetcher( } func (c *CompositeFetcher) IsPrivate(modulePath string) bool { - for _, pattern := range c.patterns { + return isPrivateModule(c.patterns, modulePath) +} + +func isPrivateModule(patterns []string, modulePath string) bool { + for _, pattern := range patterns { matched, err := path.Match(pattern, modulePath) if err == nil && matched { return true diff --git a/internal/strategy/gomod/gomod.go b/internal/strategy/gomod/gomod.go index cc3a3d97..915a3ae9 100644 --- a/internal/strategy/gomod/gomod.go +++ b/internal/strategy/gomod/gomod.go @@ -6,16 +6,21 @@ import ( "net/http" "net/url" "os/exec" + "strings" "github.com/alecthomas/errors" "github.com/goproxy/goproxy" + "golang.org/x/mod/module" "github.com/block/cachew/internal/cache" "github.com/block/cachew/internal/gitclone" "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/packagepolicy" "github.com/block/cachew/internal/strategy" ) +const disableModuleFetchHeader = "Disable-Module-Fetch" + func Register(r *strategy.Registry, cloneManager gitclone.ManagerProvider) { strategy.Register(r, "gomod", "Caches Go module proxy requests.", func(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux) (*Strategy, error) { return New(ctx, config, cache, mux, cloneManager) @@ -23,17 +28,21 @@ func Register(r *strategy.Registry, cloneManager gitclone.ManagerProvider) { } type Config struct { - Proxy string `hcl:"proxy,optional" help:"Upstream Go module proxy URL (defaults to proxy.golang.org)" default:"https://proxy.golang.org"` - PrivatePaths []string `hcl:"private-paths,optional" help:"Module path patterns for private repositories"` + Proxy string `hcl:"proxy,optional" help:"Upstream Go module proxy URL (defaults to proxy.golang.org)" default:"https://proxy.golang.org"` + PrivatePaths []string `hcl:"private-paths,optional" help:"Module path patterns for private repositories"` + PackagePolicy *packagepolicy.Config `hcl:"package-policy,block,optional" help:"Optional package security policy enforced before cold public module downloads."` } type Strategy struct { - config Config - cache cache.Cache - logger *slog.Logger - proxy *url.URL - goproxy *goproxy.Goproxy - cloneManager *gitclone.Manager + config Config + cache cache.Cache + logger *slog.Logger + proxy *url.URL + goproxy *goproxy.Goproxy + cacher *goproxyCacher + packagePolicy packagepolicy.Evaluator + proxyHandler http.Handler + cloneManager *gitclone.Manager } var _ strategy.Strategy = (*Strategy)(nil) @@ -62,6 +71,12 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux proxy: parsedURL, cloneManager: cloneManager, } + if config.PackagePolicy != nil { + s.packagePolicy, err = packagepolicy.New(*config.PackagePolicy) + if err != nil { + return nil, errors.Wrap(err, "create package policy") + } + } publicFetcher := &goproxy.GoFetcher{ Env: []string{ @@ -80,12 +95,11 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux s.logger.InfoContext(ctx, "Configured private module support", "private-paths", config.PrivatePaths) } + s.cacher = &goproxyCacher{cache: cache} s.goproxy = &goproxy.Goproxy{ Logger: s.logger, Fetcher: fetcher, - Cacher: &goproxyCacher{ - cache: cache, - }, + Cacher: s.cacher, ProxiedSumDBs: []string{ "sum.golang.org https://sum.golang.org", }, @@ -93,11 +107,61 @@ func New(ctx context.Context, config Config, cache cache.Cache, mux strategy.Mux s.logger.InfoContext(ctx, "Initialized Go module proxy strategy", "proxy", s.proxy) - mux.Handle("GET /gomod/{path...}", http.StripPrefix("/gomod", s.goproxy)) + s.proxyHandler = http.StripPrefix("/gomod", s.goproxy) + mux.Handle("GET /gomod/{path...}", http.HandlerFunc(s.serveHTTP)) return s, nil } +func (s *Strategy) serveHTTP(w http.ResponseWriter, r *http.Request) { + path := strings.TrimPrefix(r.URL.Path, "/gomod/") + purl, ok := packagepolicy.PackageURLForGoModule("/" + path) + if s.packagePolicy == nil { + s.proxyHandler.ServeHTTP(w, r) + return + } + if !ok || s.privateModulePath(path) { + s.packagePolicy.ObserveNotApplicable(r.Context()) + s.proxyHandler.ServeHTTP(w, r) + return + } + if s.cached(path, r) { + r.Header.Set(disableModuleFetchHeader, "true") + s.proxyHandler.ServeHTTP(w, r) + return + } + decision, err := s.packagePolicy.Evaluate(r.Context(), purl) + if err != nil { + s.logger.ErrorContext(r.Context(), "Package policy evaluation failed", "error", err) + } + if !packagepolicy.AllowRequest(w, decision, err) { + return + } + s.proxyHandler.ServeHTTP(w, r) +} + +func (s *Strategy) cached(path string, r *http.Request) bool { + if s.cacher == nil || r.URL.RawQuery != "" || r.Header.Get("Range") != "" { + return false + } + return s.cacher.Exists(r.Context(), path) +} + +func (s *Strategy) privateModulePath(requestPath string) bool { + if len(s.config.PrivatePaths) == 0 { + return false + } + escapedPath, _, ok := strings.Cut(requestPath, "/@v/") + if !ok { + return false + } + modulePath, err := module.UnescapePath(escapedPath) + if err != nil { + return false + } + return isPrivateModule(s.config.PrivatePaths, modulePath) +} + func (s *Strategy) String() string { return "gomod:" + s.proxy.Host } diff --git a/internal/strategy/gomod/package_policy_test.go b/internal/strategy/gomod/package_policy_test.go new file mode 100644 index 00000000..bf8f8a35 --- /dev/null +++ b/internal/strategy/gomod/package_policy_test.go @@ -0,0 +1,176 @@ +package gomod //nolint:testpackage // White-box coverage is required for policy and cache injection. + +import ( + "bytes" + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/cachew/internal/cache" + "github.com/block/cachew/internal/logging" + "github.com/block/cachew/internal/packagepolicy" +) + +type recordingPackagePolicy struct { + decision packagepolicy.Decision + err error + purls []string + notApplicable int +} + +type cacheProbe struct { + cache.Cache + statCalls int + openCalls int +} + +func (c *cacheProbe) Stat(ctx context.Context, key cache.Key, opts ...cache.Option) (http.Header, error) { + c.statCalls++ + return c.Cache.Stat(ctx, key, opts...) +} + +func (c *cacheProbe) Open(ctx context.Context, key cache.Key, opts ...cache.Option) (io.ReadCloser, http.Header, error) { + c.openCalls++ + return c.Cache.Open(ctx, key, opts...) +} + +func (r *recordingPackagePolicy) Evaluate(_ context.Context, purl string) (packagepolicy.Decision, error) { + r.purls = append(r.purls, purl) + return r.decision, r.err +} + +func (r *recordingPackagePolicy) ObserveNotApplicable(context.Context) { + r.notApplicable++ +} + +func TestGoModuleEnforcesPackagePolicyBeforeOrigin(t *testing.T) { + tests := []struct { + name string + decision packagepolicy.Decision + err error + statusCode int + policy string + }{ + { + name: "denied package", + decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictDeny, Reasons: []string{"malware"}}, + statusCode: http.StatusForbidden, + policy: "deny", + }, + { + name: "pending package", + decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictPending}, + statusCode: http.StatusServiceUnavailable, + policy: "pending", + }, + { + name: "policy unavailable", + err: io.ErrUnexpectedEOF, + statusCode: http.StatusServiceUnavailable, + policy: "unavailable", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var logs bytes.Buffer + var originRequests int + policy := &recordingPackagePolicy{decision: test.decision, err: test.err} + strategy := &Strategy{ + packagePolicy: policy, + logger: slog.New(slog.NewJSONHandler(&logs, nil)), + proxyHandler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + originRequests++ + w.WriteHeader(http.StatusOK) + }), + } + + w := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/gomod/github.com/pkg/errors/@v/v0.9.1.zip", nil) + strategy.serveHTTP(w, request) + + assert.Equal(t, test.statusCode, w.Code) + assert.Equal(t, test.policy, w.Header().Get("X-Cachew-Package-Policy")) + assert.Equal(t, []string{"pkg:golang/github.com/pkg/errors@v0.9.1"}, policy.purls) + assert.Equal(t, 0, originRequests) + if test.err != nil { + assert.Contains(t, logs.String(), test.err.Error()) + } + }) + } +} + +func TestGoModuleCachedPackageBypassesPackagePolicy(t *testing.T) { + _, ctx := logging.Configure(context.Background(), logging.Config{Level: slog.LevelError}) + memory, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 1, MaxTTL: time.Hour}) + assert.NoError(t, err) + t.Cleanup(func() { assert.NoError(t, memory.Close()) }) + probe := &cacheProbe{Cache: memory} + cacher := &goproxyCacher{cache: probe} + cacheName := "github.com/pkg/errors/@v/v0.9.1.zip" + assert.NoError(t, cacher.Put(ctx, cacheName, strings.NewReader("cached module"))) + + policy := &recordingPackagePolicy{decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictDeny}} + strategy := &Strategy{ + packagePolicy: policy, + cacher: cacher, + proxyHandler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assert.Equal(t, "true", r.Header.Get("Disable-Module-Fetch")) + _, _ = io.WriteString(w, "cached module") + }), + } + w := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/gomod/"+cacheName, nil) + strategy.serveHTTP(w, request) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "cached module", w.Body.String()) + assert.Equal(t, []string(nil), policy.purls) + assert.Equal(t, 1, probe.statCalls) + assert.Equal(t, 0, probe.openCalls) +} + +func TestGoModulePrivatePackageBypassesPackagePolicy(t *testing.T) { + policy := &recordingPackagePolicy{decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictDeny}} + strategy := &Strategy{ + config: Config{PrivatePaths: []string{"github.com/myorg/*"}}, + packagePolicy: policy, + proxyHandler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "private module") + }), + } + w := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/gomod/github.com/myorg/private/@v/v1.0.0.zip", nil) + strategy.serveHTTP(w, request) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "private module", w.Body.String()) + assert.Equal(t, []string(nil), policy.purls) + assert.Equal(t, 1, policy.notApplicable) +} + +func TestGoModuleBranchQueryBypassesPackagePolicy(t *testing.T) { + policy := &recordingPackagePolicy{decision: packagepolicy.Decision{Verdict: packagepolicy.VerdictDeny}} + strategy := &Strategy{ + packagePolicy: policy, + proxyHandler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "resolved branch") + }), + } + w := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/gomod/github.com/pkg/errors/@v/master.info", nil) + + strategy.serveHTTP(w, request) + + assert.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, "resolved branch", w.Body.String()) + assert.Equal(t, []string(nil), policy.purls) + assert.Equal(t, 1, policy.notApplicable) +}