From 4aced0cb96cf0bca77d644b8ac452ff25d455fe9 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Thu, 23 Apr 2026 17:09:48 +0530 Subject: [PATCH 1/2] feat: add VEX Hub certifier Signed-off-by: Abhishek --- cmd/guaccollect/cmd/vexhub.go | 197 +++++++++++++++ cmd/guacone/cmd/vexhub.go | 357 +++++++++++++++++++++++++++ pkg/certifier/certifier.go | 1 + pkg/certifier/vexhub/vexhub.go | 359 ++++++++++++++++++++++++++++ pkg/certifier/vexhub/vexhub_test.go | 263 ++++++++++++++++++++ 5 files changed, 1177 insertions(+) create mode 100644 cmd/guaccollect/cmd/vexhub.go create mode 100644 cmd/guacone/cmd/vexhub.go create mode 100644 pkg/certifier/vexhub/vexhub.go create mode 100644 pkg/certifier/vexhub/vexhub_test.go diff --git a/cmd/guaccollect/cmd/vexhub.go b/cmd/guaccollect/cmd/vexhub.go new file mode 100644 index 0000000000..e77d6bedd7 --- /dev/null +++ b/cmd/guaccollect/cmd/vexhub.go @@ -0,0 +1,197 @@ +// +// Copyright 2024 The GUAC 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 cmd + +import ( + "context" + "fmt" + "net/http" + "os" + "time" + + "github.com/Khan/genqlient/graphql" + "github.com/guacsec/guac/pkg/assembler/clients/generated" + "github.com/guacsec/guac/pkg/certifier" + "github.com/guacsec/guac/pkg/certifier/certify" + "github.com/guacsec/guac/pkg/certifier/components/root_package" + "github.com/guacsec/guac/pkg/certifier/vexhub" + "github.com/guacsec/guac/pkg/cli" + "github.com/guacsec/guac/pkg/logging" + "github.com/guacsec/guac/pkg/metrics" + "github.com/spf13/cobra" + "github.com/spf13/viper" +) + +const ( + vexHubQuerySize = 1000 +) + +type vexHubOptions struct { + graphqlEndpoint string + headerFile string + pubsubAddr string + blobAddr string + poll bool + interval time.Duration + publishToQueue bool + addedLatency *time.Duration + batchSize int + lastScan *int + enableOtel bool + manifestURL string +} + +var vexHubCmd = &cobra.Command{ + Use: "vexhub [flags]", + Short: "runs the VEX Hub certifier", + Long: ` +guaccollect vexhub runs the VEX Hub certifier that queries VEX repositories +(conforming to the VEX Repo Spec) for VEX statements affecting packages in GUAC. +Ingestion to GUAC happens via an event stream (NATS) to allow for decoupling of +the collectors from the ingestion into GUAC.`, + Run: func(cmd *cobra.Command, args []string) { + opts, err := validateVEXHubFlags( + viper.GetString("gql-addr"), + viper.GetString("header-file"), + viper.GetString("pubsub-addr"), + viper.GetString("blob-addr"), + viper.GetString("interval"), + viper.GetBool("service-poll"), + viper.GetBool("publish-to-queue"), + viper.GetString("certifier-latency"), + viper.GetInt("certifier-batch-size"), + viper.GetInt("last-scan"), + viper.GetBool("enable-otel"), + viper.GetString("vexhub-manifest-url"), + ) + if err != nil { + fmt.Printf("unable to validate flags: %v\n", err) + _ = cmd.Help() + os.Exit(1) + } + + ctx := logging.WithLogger(context.Background()) + logger := logging.FromContext(ctx) + + if opts.enableOtel { + shutdown, err := metrics.SetupOTelSDK(ctx) + if err != nil { + logger.Fatalf("Error setting up Otel: %v", err) + } + defer func() { + if err := shutdown(ctx); err != nil { + logger.Errorf("Error on Otel shutdown: %v", err) + } + }() + } + + vexHubCertifierFunc := func() certifier.Certifier { + return vexhub.NewVEXHubCertifier(opts.manifestURL) + } + if err := certify.RegisterCertifier(vexHubCertifierFunc, certifier.CertifierVEXHub); err != nil { + logger.Fatalf("unable to register certifier: %v", err) + } + + transport := cli.HTTPHeaderTransport(ctx, opts.headerFile, http.DefaultTransport) + httpClient := http.Client{Transport: transport} + gqlclient := graphql.NewClient(opts.graphqlEndpoint, &httpClient) + + packageQueryFunc, err := getVEXHubPackageQuery(gqlclient, opts.batchSize, opts.addedLatency, opts.lastScan) + if err != nil { + logger.Errorf("error: %v", err) + os.Exit(1) + } + + initializeNATsandCertifier(ctx, opts.blobAddr, opts.pubsubAddr, opts.poll, opts.publishToQueue, opts.interval, packageQueryFunc()) + }, +} + +func getVEXHubPackageQuery(client graphql.Client, batchSize int, addedLatency *time.Duration, lastScan *int) (func() certifier.QueryComponents, error) { + return func() certifier.QueryComponents { + packageQuery := root_package.NewPackageQuery(client, generated.QueryTypeVulnerability, batchSize, vexHubQuerySize, addedLatency, lastScan) + return packageQuery + }, nil +} + +func validateVEXHubFlags( + graphqlEndpoint, + headerFile, + pubsubAddr, + blobAddr, + interval string, + poll bool, + pubToQueue bool, + certifierLatencyStr string, + batchSize int, + lastScan int, + enableOtel bool, + manifestURL string, +) (vexHubOptions, error) { + var opts vexHubOptions + + opts.graphqlEndpoint = graphqlEndpoint + opts.headerFile = headerFile + opts.pubsubAddr = pubsubAddr + opts.blobAddr = blobAddr + opts.poll = poll + opts.publishToQueue = pubToQueue + opts.enableOtel = enableOtel + opts.manifestURL = manifestURL + + i, err := time.ParseDuration(interval) + if err != nil { + return opts, fmt.Errorf("failed to parser duration with error: %w", err) + } + opts.interval = i + + if certifierLatencyStr != "" { + addedLatency, err := time.ParseDuration(certifierLatencyStr) + if err != nil { + return opts, fmt.Errorf("failed to parser duration with error: %w", err) + } + opts.addedLatency = &addedLatency + } else { + opts.addedLatency = nil + } + + opts.batchSize = batchSize + if lastScan != 0 { + opts.lastScan = &lastScan + } + return opts, nil +} + +func init() { + set, err := cli.BuildFlags([]string{"interval", + "header-file", "certifier-latency", + "certifier-batch-size", "last-scan"}) + if err != nil { + fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err) + os.Exit(1) + } + vexHubCmd.Flags().String("vexhub-manifest-url", vexhub.DefaultManifestURL, + "URL of the VEX repository manifest (vex-repository.json)") + vexHubCmd.PersistentFlags().AddFlagSet(set) + if err := viper.BindPFlags(vexHubCmd.PersistentFlags()); err != nil { + fmt.Fprintf(os.Stderr, "failed to bind flags: %v", err) + os.Exit(1) + } + if err := viper.BindPFlags(vexHubCmd.Flags()); err != nil { + fmt.Fprintf(os.Stderr, "failed to bind flags: %v", err) + os.Exit(1) + } + rootCmd.AddCommand(vexHubCmd) +} diff --git a/cmd/guacone/cmd/vexhub.go b/cmd/guacone/cmd/vexhub.go new file mode 100644 index 0000000000..8a359d70ee --- /dev/null +++ b/cmd/guacone/cmd/vexhub.go @@ -0,0 +1,357 @@ +// +// Copyright 2024 The GUAC 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 cmd + +import ( + "context" + "fmt" + "net/http" + "os" + "os/signal" + "sync" + "sync/atomic" + "syscall" + "time" + + "github.com/Khan/genqlient/graphql" + "github.com/guacsec/guac/pkg/assembler/clients/generated" + "github.com/guacsec/guac/pkg/certifier" + "github.com/guacsec/guac/pkg/certifier/certify" + "github.com/guacsec/guac/pkg/certifier/components/root_package" + "github.com/guacsec/guac/pkg/certifier/vexhub" + "github.com/guacsec/guac/pkg/cli" + csub_client "github.com/guacsec/guac/pkg/collectsub/client" + "github.com/guacsec/guac/pkg/handler/processor" + "github.com/guacsec/guac/pkg/ingestor" + "github.com/guacsec/guac/pkg/logging" + "github.com/guacsec/guac/pkg/metrics" + "github.com/spf13/cobra" + "github.com/spf13/viper" +) + +const ( + vexHubQuerySize = 1000 +) + +type vexHubOptions struct { + graphqlEndpoint string + headerFile string + poll bool + csubClientOptions csub_client.CsubClientOptions + interval time.Duration + queryVulnOnIngestion bool + queryLicenseOnIngestion bool + queryEOLOnIngestion bool + queryDepsDevOnIngestion bool + addedLatency *time.Duration + batchSize int + lastScan *int + enableOtel bool + manifestURL string +} + +var vexHubCmd = &cobra.Command{ + Use: "vexhub [flags]", + Short: "runs the VEX Hub certifier", + Long: "Queries VEX repositories (conforming to the VEX Repo Spec) for VEX statements affecting packages in GUAC", + Run: func(cmd *cobra.Command, args []string) { + opts, err := validateVEXHubFlags( + viper.GetString("gql-addr"), + viper.GetString("header-file"), + viper.GetString("interval"), + viper.GetString("csub-addr"), + viper.GetBool("poll"), + viper.GetBool("csub-tls"), + viper.GetBool("csub-tls-skip-verify"), + viper.GetBool("add-vuln-on-ingest"), + viper.GetBool("add-license-on-ingest"), + viper.GetBool("add-eol-on-ingest"), + viper.GetBool("add-depsdev-on-ingest"), + viper.GetString("certifier-latency"), + viper.GetInt("certifier-batch-size"), + viper.GetInt("last-scan"), + viper.GetBool("enable-otel"), + viper.GetString("vexhub-manifest-url"), + ) + if err != nil { + fmt.Printf("unable to validate flags: %v\n", err) + _ = cmd.Help() + os.Exit(1) + } + + ctx := logging.WithLogger(context.Background()) + logger := logging.FromContext(ctx) + transport := cli.HTTPHeaderTransport(ctx, opts.headerFile, http.DefaultTransport) + + if opts.enableOtel { + shutdown, err := metrics.SetupOTelSDK(ctx) + if err != nil { + logger.Fatalf("Error setting up Otel: %v", err) + } + defer func() { + if err := shutdown(ctx); err != nil { + logger.Errorf("Error on Otel shutdown: %v", err) + } + }() + } + + vexHubCertifierFunc := func() certifier.Certifier { + return vexhub.NewVEXHubCertifier(opts.manifestURL) + } + if err := certify.RegisterCertifier(vexHubCertifierFunc, certifier.CertifierVEXHub); err != nil { + logger.Fatalf("unable to register certifier: %v", err) + } + + csubClient, err := csub_client.NewClient(opts.csubClientOptions) + if err != nil { + logger.Infof("collectsub client initialization failed, this ingestion will not pull in any additional data through the collectsub service: %v", err) + csubClient = nil + } else { + defer csubClient.Close() + } + + httpClient := http.Client{Transport: transport} + gqlclient := graphql.NewClient(opts.graphqlEndpoint, &httpClient) + packageQuery := root_package.NewPackageQuery(gqlclient, generated.QueryTypeVulnerability, opts.batchSize, vexHubQuerySize, opts.addedLatency, opts.lastScan) + + totalNum := 0 + docChan := make(chan *processor.Document) + ingestionStop := make(chan bool, 1) + tickInterval := 30 * time.Second + ticker := time.NewTicker(tickInterval) + ctx, cf := context.WithCancel(ctx) + + var gotErr int32 + var wg sync.WaitGroup + ingestion := func() { + defer wg.Done() + var totalDocs []*processor.Document + const threshold = 1000 + stop := false + for !stop { + select { + case <-ticker.C: + if len(totalDocs) > 0 { + err = ingestor.MergedIngest(ctx, + totalDocs, + opts.graphqlEndpoint, + transport, + csubClient, + opts.queryVulnOnIngestion, + opts.queryLicenseOnIngestion, + opts.queryEOLOnIngestion, + opts.queryDepsDevOnIngestion, + ) + if err != nil { + stop = true + atomic.StoreInt32(&gotErr, 1) + logger.Errorf("unable to ingest documents: %v", err) + } + totalDocs = []*processor.Document{} + } + ticker.Reset(tickInterval) + case d := <-docChan: + totalNum += 1 + totalDocs = append(totalDocs, d) + if len(totalDocs) >= threshold { + err = ingestor.MergedIngest(ctx, + totalDocs, + opts.graphqlEndpoint, + transport, + csubClient, + opts.queryVulnOnIngestion, + opts.queryLicenseOnIngestion, + opts.queryEOLOnIngestion, + opts.queryDepsDevOnIngestion, + ) + if err != nil { + stop = true + atomic.StoreInt32(&gotErr, 1) + logger.Errorf("unable to ingest documents: %v", err) + } + totalDocs = []*processor.Document{} + ticker.Reset(tickInterval) + } + case <-ingestionStop: + stop = true + case <-ctx.Done(): + return + } + } + for len(docChan) > 0 { + totalNum += 1 + totalDocs = append(totalDocs, <-docChan) + if len(totalDocs) >= threshold { + err = ingestor.MergedIngest( + ctx, + totalDocs, + opts.graphqlEndpoint, + transport, + csubClient, + opts.queryVulnOnIngestion, + opts.queryLicenseOnIngestion, + opts.queryEOLOnIngestion, + opts.queryDepsDevOnIngestion, + ) + if err != nil { + atomic.StoreInt32(&gotErr, 1) + logger.Errorf("unable to ingest documents: %v", err) + } + totalDocs = []*processor.Document{} + } + } + if len(totalDocs) > 0 { + err = ingestor.MergedIngest( + ctx, + totalDocs, + opts.graphqlEndpoint, + transport, + csubClient, + opts.queryVulnOnIngestion, + opts.queryLicenseOnIngestion, + opts.queryEOLOnIngestion, + opts.queryDepsDevOnIngestion, + ) + if err != nil { + atomic.StoreInt32(&gotErr, 1) + logger.Errorf("unable to ingest documents: %v", err) + } + } + } + wg.Add(1) + go ingestion() + + emit := func(d *processor.Document) error { + docChan <- d + return nil + } + + errHandler := func(err error) bool { + if err != nil { + logger.Errorf("certifier ended with error: %v", err) + atomic.StoreInt32(&gotErr, 1) + } + return true + } + + done := make(chan bool, 1) + wg.Add(1) + go func() { + defer wg.Done() + if err := certify.Certify(ctx, packageQuery, emit, errHandler, opts.poll, opts.interval); err != nil { + logger.Errorf("Unhandled error in the certifier: %s", err) + } + done <- true + }() + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM) + select { + case s := <-sigs: + logger.Infof("Signal received: %s, shutting down gracefully\n", s.String()) + cf() + case <-done: + logger.Infof("All certifiers completed") + } + ingestionStop <- true + wg.Wait() + cf() + + if atomic.LoadInt32(&gotErr) == 1 { + logger.Errorf("completed ingestion with errors") + } else { + logger.Infof("completed ingesting %v documents", totalNum) + } + }, +} + +func validateVEXHubFlags( + graphqlEndpoint, + headerFile, + interval, + csubAddr string, + poll, + csubTls, + csubTlsSkipVerify bool, + queryVulnIngestion bool, + queryLicenseIngestion bool, + queryEOLIngestion bool, + queryDepsDevIngestion bool, + certifierLatencyStr string, + batchSize int, lastScan int, + enableOtel bool, + manifestURL string, +) (vexHubOptions, error) { + var opts vexHubOptions + opts.graphqlEndpoint = graphqlEndpoint + opts.headerFile = headerFile + opts.poll = poll + i, err := time.ParseDuration(interval) + if err != nil { + return opts, err + } + opts.interval = i + opts.enableOtel = enableOtel + opts.manifestURL = manifestURL + + if certifierLatencyStr != "" { + addedLatency, err := time.ParseDuration(certifierLatencyStr) + if err != nil { + return opts, fmt.Errorf("failed to parser duration with error: %w", err) + } + opts.addedLatency = &addedLatency + } else { + opts.addedLatency = nil + } + + opts.batchSize = batchSize + + if lastScan != 0 { + opts.lastScan = &lastScan + } + + csubOpts, err := csub_client.ValidateCsubClientFlags(csubAddr, csubTls, csubTlsSkipVerify) + if err != nil { + return opts, fmt.Errorf("unable to validate csub client flags: %w", err) + } + opts.csubClientOptions = csubOpts + opts.queryVulnOnIngestion = queryVulnIngestion + opts.queryLicenseOnIngestion = queryLicenseIngestion + opts.queryEOLOnIngestion = queryEOLIngestion + opts.queryDepsDevOnIngestion = queryDepsDevIngestion + + return opts, nil +} + +func init() { + set, err := cli.BuildFlags([]string{"certifier-latency", + "certifier-batch-size", "last-scan"}) + if err != nil { + fmt.Fprintf(os.Stderr, "failed to setup flag: %v", err) + os.Exit(1) + } + vexHubCmd.Flags().String("vexhub-manifest-url", vexhub.DefaultManifestURL, + "URL of the VEX repository manifest (vex-repository.json)") + vexHubCmd.PersistentFlags().AddFlagSet(set) + if err := viper.BindPFlags(vexHubCmd.PersistentFlags()); err != nil { + fmt.Fprintf(os.Stderr, "failed to bind flags: %v", err) + os.Exit(1) + } + if err := viper.BindPFlags(vexHubCmd.Flags()); err != nil { + fmt.Fprintf(os.Stderr, "failed to bind flags: %v", err) + os.Exit(1) + } + certifierCmd.AddCommand(vexHubCmd) +} diff --git a/pkg/certifier/certifier.go b/pkg/certifier/certifier.go index 35e47ea99e..ee40ad11fb 100644 --- a/pkg/certifier/certifier.go +++ b/pkg/certifier/certifier.go @@ -51,4 +51,5 @@ const ( CertifierClearlyDefined CertifierType = "CD" CertifierScorecard CertifierType = "scorecard" CertifierEOL CertifierType = "EOL" + CertifierVEXHub CertifierType = "vexhub" ) diff --git a/pkg/certifier/vexhub/vexhub.go b/pkg/certifier/vexhub/vexhub.go new file mode 100644 index 0000000000..4f7b34ae12 --- /dev/null +++ b/pkg/certifier/vexhub/vexhub.go @@ -0,0 +1,359 @@ +// +// Copyright 2024 The GUAC 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 vexhub + +import ( + "archive/tar" + "compress/gzip" + "context" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" + + "github.com/guacsec/guac/pkg/certifier" + "github.com/guacsec/guac/pkg/certifier/components/root_package" + "github.com/guacsec/guac/pkg/clients" + "github.com/guacsec/guac/pkg/events" + "github.com/guacsec/guac/pkg/handler/processor" + "github.com/guacsec/guac/pkg/logging" + "github.com/guacsec/guac/pkg/version" + + jsoniter "github.com/json-iterator/go" + "golang.org/x/time/rate" +) + +var json = jsoniter.ConfigCompatibleWithStandardLibrary + +const ( + DefaultManifestURL = "https://raw.githubusercontent.com/aquasecurity/vexhub/main/vex-repository.json" + VEXHubCollector = "vexhub_certifier" + rateLimit = 100 +) + +var rateLimitInterval = time.Second + +var ErrComponentTypeMismatch = errors.New("rootComponent type is not []*root_package.PackageNode") + +// Manifest represents a vex-repository.json file. +type Manifest struct { + Name string `json:"name"` + Versions []ManifestVersion `json:"versions"` +} + +// ManifestVersion holds the spec version and locations of a VEX repo. +type ManifestVersion struct { + SpecVersion string `json:"spec_version"` + Locations []ManifestLocation `json:"locations"` + UpdateInterval string `json:"update_interval"` +} + +// ManifestLocation holds the URL for a VEX repo archive. +type ManifestLocation struct { + URL string `json:"url"` +} + +// Index represents the index.json file inside the VEX repo archive. +type Index struct { + UpdatedAt string `json:"updated_at"` + Packages []IndexPackage `json:"packages"` +} + +// IndexPackage maps a PURL to its VEX document location. +type IndexPackage struct { + ID string `json:"id"` + Location string `json:"location"` + Format string `json:"format,omitempty"` +} + +// vexHubCertifier queries VEX repositories for VEX statements. +type vexHubCertifier struct { + httpClient *http.Client + manifestURL string +} + +// NewVEXHubCertifier creates a new VEX Hub certifier. +func NewVEXHubCertifier(manifestURL string) certifier.Certifier { + limiter := rate.NewLimiter(rate.Every(rateLimitInterval), rateLimit) + transport := clients.NewRateLimitedTransport(version.UATransport, limiter) + client := &http.Client{Transport: transport} + if manifestURL == "" { + manifestURL = DefaultManifestURL + } + return &vexHubCertifier{ + httpClient: client, + manifestURL: manifestURL, + } +} + +// CertifyComponent fetches VEX documents from the VEX Hub for the given packages. +func (v *vexHubCertifier) CertifyComponent(ctx context.Context, rootComponent interface{}, docChannel chan<- *processor.Document) error { + logger := logging.FromContext(ctx) + + packageNodes, ok := rootComponent.([]*root_package.PackageNode) + if !ok { + return ErrComponentTypeMismatch + } + + var purls []string + for _, node := range packageNodes { + purls = append(purls, node.Purl) + } + + if len(purls) == 0 { + return nil + } + + // Fetch the manifest to get the archive URL. + manifest, err := fetchManifest(ctx, v.httpClient, v.manifestURL) + if err != nil { + return fmt.Errorf("failed to fetch VEX Hub manifest: %w", err) + } + + archiveURL, subdir := getArchiveURL(manifest) + if archiveURL == "" { + logger.Infof("no archive location found in VEX Hub manifest") + return nil + } + + // Download and extract the archive, building a PURL→VEX doc map. + vexDocs, err := downloadAndIndex(ctx, v.httpClient, archiveURL, subdir) + if err != nil { + return fmt.Errorf("failed to download VEX Hub archive: %w", err) + } + + logger.Infof("VEX Hub: indexed %d packages from archive", len(vexDocs)) + + // Look up each PURL and emit matching VEX documents. + if _, err := emitVEXDocuments(purls, vexDocs, docChannel); err != nil { + return fmt.Errorf("failed to emit VEX documents: %w", err) + } + + return nil +} + +// fetchManifest downloads and parses the vex-repository.json manifest. +func fetchManifest(ctx context.Context, client *http.Client, url string) (*Manifest, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, fmt.Errorf("creating request: %w", err) + } + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("fetching manifest: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("manifest fetch returned status %d", resp.StatusCode) + } + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, fmt.Errorf("reading manifest body: %w", err) + } + + var manifest Manifest + if err := json.Unmarshal(body, &manifest); err != nil { + return nil, fmt.Errorf("parsing manifest: %w", err) + } + return &manifest, nil +} + +// getArchiveURL extracts the archive URL and optional subdirectory from the manifest. +// The subdirectory is specified after "//" in the URL (per VEX Repo Spec). +// Example: "https://example.com/archive.tar.gz//vexhub-main" → URL + subdir "vexhub-main" +func getArchiveURL(manifest *Manifest) (archiveURL, subdir string) { + if len(manifest.Versions) == 0 { + return "", "" + } + // Use the first version with a location. + for _, v := range manifest.Versions { + if len(v.Locations) > 0 { + rawURL := v.Locations[0].URL + // Look for "//" that is NOT part of the scheme (e.g., "https://"). + // The subdirectory separator always appears after the host/path portion. + schemeEnd := strings.Index(rawURL, "://") + searchStart := 0 + if schemeEnd >= 0 { + searchStart = schemeEnd + 3 + } + rest := rawURL[searchStart:] + if idx := strings.Index(rest, "//"); idx >= 0 { + archiveURL = rawURL[:searchStart+idx] + subdir = rest[idx+2:] + } else { + archiveURL = rawURL + } + return archiveURL, subdir + } + } + return "", "" +} + +// downloadAndIndex downloads a tar.gz archive, extracts it, parses index.json, +// and returns a map of PURL→VEX document bytes. +func downloadAndIndex(ctx context.Context, client *http.Client, archiveURL, subdir string) (map[string][]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, archiveURL, nil) + if err != nil { + return nil, fmt.Errorf("creating archive request: %w", err) + } + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("downloading archive: %w", err) + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("archive download returned status %d", resp.StatusCode) + } + + gz, err := gzip.NewReader(resp.Body) + if err != nil { + return nil, fmt.Errorf("creating gzip reader: %w", err) + } + defer func() { _ = gz.Close() }() + + tr := tar.NewReader(gz) + + // First pass: extract all files into memory. + files := make(map[string][]byte) + for { + header, err := tr.Next() + if err == io.EOF { + break + } + if err != nil { + return nil, fmt.Errorf("reading tar entry: %w", err) + } + if header.Typeflag != tar.TypeReg { + continue + } + + data, err := io.ReadAll(tr) + if err != nil { + return nil, fmt.Errorf("reading file %s: %w", header.Name, err) + } + + // Normalize path: strip the subdir prefix if present. + name := header.Name + if subdir != "" { + if after, found := strings.CutPrefix(name, subdir+"/"); found { + name = after + } else if after, found := strings.CutPrefix(name, subdir); found { + name = after + } + } + // Also strip leading directory component from tar (e.g., "vexhub-main/") + if idx := strings.Index(name, "/"); idx >= 0 { + // Keep the path after the first directory component only if subdir wasn't already stripped. + if subdir == "" { + name = name[idx+1:] + } + } + if name != "" { + files[name] = data + } + } + + // Parse index.json. + indexData, ok := files["index.json"] + if !ok { + return nil, fmt.Errorf("index.json not found in archive") + } + + var index Index + if err := json.Unmarshal(indexData, &index); err != nil { + return nil, fmt.Errorf("parsing index.json: %w", err) + } + + // Build PURL→VEX doc map. + vexDocs := make(map[string][]byte, len(index.Packages)) + for _, pkg := range index.Packages { + docData, ok := files[pkg.Location] + if !ok { + continue + } + vexDocs[pkg.ID] = docData + } + + return vexDocs, nil +} + +// emitVEXDocuments looks up each PURL in the VEX index and emits matching documents. +func emitVEXDocuments(purls []string, vexDocs map[string][]byte, docChannel chan<- *processor.Document) ([]*processor.Document, error) { + var emitted []*processor.Document + seen := make(map[string]bool) + + for _, purl := range purls { + if strings.Contains(purl, "pkg:guac") { + continue + } + + // Try exact PURL match first, then strip version for lookup. + lookupKey := purl + if _, ok := vexDocs[lookupKey]; !ok { + lookupKey = stripPurlVersion(purl) + } + + docData, ok := vexDocs[lookupKey] + if !ok { + continue + } + + if seen[lookupKey] { + continue + } + seen[lookupKey] = true + + doc := &processor.Document{ + Blob: docData, + Type: processor.DocumentOpenVEX, + Format: processor.FormatJSON, + SourceInformation: processor.SourceInformation{ + Collector: VEXHubCollector, + Source: VEXHubCollector, + DocumentRef: events.GetDocRef(docData), + }, + } + if docChannel != nil { + docChannel <- doc + } + emitted = append(emitted, doc) + } + + return emitted, nil +} + +// stripPurlVersion removes the version, qualifiers, and subpath from a PURL. +// e.g. "pkg:npm/lodash@4.17.21" → "pkg:npm/lodash" +func stripPurlVersion(purl string) string { + // Remove subpath (after last #) + if idx := strings.Index(purl, "#"); idx >= 0 { + purl = purl[:idx] + } + // Remove qualifiers (after ?) + if idx := strings.Index(purl, "?"); idx >= 0 { + purl = purl[:idx] + } + // Remove version (after @) + if idx := strings.LastIndex(purl, "@"); idx >= 0 { + purl = purl[:idx] + } + return purl +} diff --git a/pkg/certifier/vexhub/vexhub_test.go b/pkg/certifier/vexhub/vexhub_test.go new file mode 100644 index 0000000000..4e75648e51 --- /dev/null +++ b/pkg/certifier/vexhub/vexhub_test.go @@ -0,0 +1,263 @@ +// +// Copyright 2024 The GUAC 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 vexhub + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/guacsec/guac/pkg/certifier/components/root_package" + "github.com/guacsec/guac/pkg/handler/processor" +) + +// buildTestArchive creates a tar.gz archive with index.json and a VEX document. +func buildTestArchive(t *testing.T, indexJSON string, vexFiles map[string]string) []byte { + t.Helper() + var buf bytes.Buffer + gw := gzip.NewWriter(&buf) + tw := tar.NewWriter(gw) + + // Write index.json under a root dir prefix (simulating GitHub archive). + writeFile := func(name, content string) { + hdr := &tar.Header{ + Name: "vexhub-main/" + name, + Mode: 0644, + Size: int64(len(content)), + } + if err := tw.WriteHeader(hdr); err != nil { + t.Fatal(err) + } + if _, err := tw.Write([]byte(content)); err != nil { + t.Fatal(err) + } + } + + writeFile("index.json", indexJSON) + for path, content := range vexFiles { + writeFile(path, content) + } + + if err := tw.Close(); err != nil { + t.Fatal(err) + } + if err := gw.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() +} + +func TestStripPurlVersion(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"pkg:npm/lodash@4.17.21", "pkg:npm/lodash"}, + {"pkg:maven/org.apache/log4j@2.0?type=jar", "pkg:maven/org.apache/log4j"}, + {"pkg:npm/lodash@4.17.21#sub/path", "pkg:npm/lodash"}, + {"pkg:deb/debian/curl", "pkg:deb/debian/curl"}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := stripPurlVersion(tt.input) + if got != tt.want { + t.Errorf("stripPurlVersion(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestGetArchiveURL(t *testing.T) { + tests := []struct { + name string + manifest Manifest + wantURL string + wantSubdir string + }{ + { + name: "simple URL", + manifest: Manifest{ + Versions: []ManifestVersion{{ + Locations: []ManifestLocation{{URL: "https://example.com/vex.tar.gz"}}, + }}, + }, + wantURL: "https://example.com/vex.tar.gz", + wantSubdir: "", + }, + { + name: "empty manifest", + manifest: Manifest{}, + wantURL: "", + wantSubdir: "", + }, + { + name: "URL with subdirectory", + manifest: Manifest{ + Versions: []ManifestVersion{{ + Locations: []ManifestLocation{{URL: "https://github.com/org/repo/archive/refs/heads/main.tar.gz//vexhub-main"}}, + }}, + }, + wantURL: "https://github.com/org/repo/archive/refs/heads/main.tar.gz", + wantSubdir: "vexhub-main", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + gotURL, gotSubdir := getArchiveURL(&tt.manifest) + if gotURL != tt.wantURL { + t.Errorf("URL = %q, want %q", gotURL, tt.wantURL) + } + if gotSubdir != tt.wantSubdir { + t.Errorf("subdir = %q, want %q", gotSubdir, tt.wantSubdir) + } + }) + } +} + +func TestEmitVEXDocuments(t *testing.T) { + vexDoc := []byte(`{"@context": "https://openvex.dev/ns/v0.2.0", "@id": "test"}`) + vexDocs := map[string][]byte{ + "pkg:npm/lodash": vexDoc, + } + + t.Run("matching purl emits document", func(t *testing.T) { + docChan := make(chan *processor.Document, 10) + docs, err := emitVEXDocuments( + []string{"pkg:npm/lodash@4.17.21"}, + vexDocs, + docChan, + ) + if err != nil { + t.Fatal(err) + } + if len(docs) != 1 { + t.Fatalf("expected 1 doc, got %d", len(docs)) + } + if docs[0].Type != processor.DocumentOpenVEX { + t.Errorf("expected type %v, got %v", processor.DocumentOpenVEX, docs[0].Type) + } + }) + + t.Run("no match emits nothing", func(t *testing.T) { + docs, err := emitVEXDocuments( + []string{"pkg:npm/express@1.0.0"}, + vexDocs, + nil, + ) + if err != nil { + t.Fatal(err) + } + if len(docs) != 0 { + t.Fatalf("expected 0 docs, got %d", len(docs)) + } + }) + + t.Run("guac purls are skipped", func(t *testing.T) { + docs, err := emitVEXDocuments( + []string{"pkg:guac/test@1.0"}, + vexDocs, + nil, + ) + if err != nil { + t.Fatal(err) + } + if len(docs) != 0 { + t.Fatalf("expected 0 docs for guac purl, got %d", len(docs)) + } + }) + + t.Run("duplicate purls only emit once", func(t *testing.T) { + docChan := make(chan *processor.Document, 10) + docs, err := emitVEXDocuments( + []string{"pkg:npm/lodash@4.17.21", "pkg:npm/lodash@4.17.20"}, + vexDocs, + docChan, + ) + if err != nil { + t.Fatal(err) + } + if len(docs) != 1 { + t.Fatalf("expected 1 doc for duplicate purls, got %d", len(docs)) + } + }) +} + +func TestCertifyComponentTypeMismatch(t *testing.T) { + c := &vexHubCertifier{httpClient: http.DefaultClient, manifestURL: "http://example.com"} + err := c.CertifyComponent(context.Background(), "wrong type", nil) + if err == nil || err != ErrComponentTypeMismatch { + t.Errorf("expected ErrComponentTypeMismatch, got %v", err) + } +} + +func TestCertifyComponentEndToEnd(t *testing.T) { + vexDoc := `{"@context":"https://openvex.dev/ns/v0.2.0","@id":"test-vex","statements":[{"vulnerability":{"name":"CVE-2021-44228"},"products":[{"@id":"pkg:npm/lodash"}],"status":"not_affected"}]}` + indexJSON := `{"updated_at":"2024-01-01T00:00:00Z","packages":[{"id":"pkg:npm/lodash","location":"pkg/npm/lodash/vex.json"}]}` + + archive := buildTestArchive(t, indexJSON, map[string]string{ + "pkg/npm/lodash/vex.json": vexDoc, + }) + + // Serve the manifest and archive. + mux := http.NewServeMux() + mux.HandleFunc("/vex-repository.json", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprintf(w, `{"name":"test","versions":[{"spec_version":"0.1","locations":[{"url":"%s/archive.tar.gz"}]}]}`, "http://"+r.Host) + }) + mux.HandleFunc("/archive.tar.gz", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/gzip") + _, _ = w.Write(archive) + }) + server := httptest.NewServer(mux) + defer server.Close() + + certifier := &vexHubCertifier{ + httpClient: server.Client(), + manifestURL: server.URL + "/vex-repository.json", + } + + docChan := make(chan *processor.Document, 10) + nodes := []*root_package.PackageNode{ + {Purl: "pkg:npm/lodash@4.17.21"}, + {Purl: "pkg:npm/express@1.0.0"}, + } + + err := certifier.CertifyComponent(context.Background(), nodes, docChan) + if err != nil { + t.Fatalf("CertifyComponent failed: %v", err) + } + + close(docChan) + var docs []*processor.Document + for doc := range docChan { + docs = append(docs, doc) + } + + if len(docs) != 1 { + t.Fatalf("expected 1 document, got %d", len(docs)) + } + if docs[0].Type != processor.DocumentOpenVEX { + t.Errorf("expected type %v, got %v", processor.DocumentOpenVEX, docs[0].Type) + } + if docs[0].SourceInformation.Collector != VEXHubCollector { + t.Errorf("expected collector %q, got %q", VEXHubCollector, docs[0].SourceInformation.Collector) + } +} From dca21a87e1053af18c37ce90d797e302c40bee54 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Mon, 4 May 2026 05:11:38 +0530 Subject: [PATCH 2/2] fix: address mentor review feedback for VEX Hub certifier Signed-off-by: Abhishek --- cmd/guaccollect/cmd/vexhub.go | 2 +- cmd/guacone/cmd/vexhub.go | 2 +- pkg/certifier/vexhub/vexhub.go | 276 +++++++++++++++++++--------- pkg/certifier/vexhub/vexhub_test.go | 275 +++++++++++++++++++++++---- 4 files changed, 433 insertions(+), 122 deletions(-) diff --git a/cmd/guaccollect/cmd/vexhub.go b/cmd/guaccollect/cmd/vexhub.go index e77d6bedd7..05fc1527b5 100644 --- a/cmd/guaccollect/cmd/vexhub.go +++ b/cmd/guaccollect/cmd/vexhub.go @@ -1,5 +1,5 @@ // -// Copyright 2024 The GUAC Authors. +// Copyright 2026 The GUAC Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. diff --git a/cmd/guacone/cmd/vexhub.go b/cmd/guacone/cmd/vexhub.go index 8a359d70ee..9c94b08279 100644 --- a/cmd/guacone/cmd/vexhub.go +++ b/cmd/guacone/cmd/vexhub.go @@ -1,5 +1,5 @@ // -// Copyright 2024 The GUAC Authors. +// Copyright 2026 The GUAC Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. diff --git a/pkg/certifier/vexhub/vexhub.go b/pkg/certifier/vexhub/vexhub.go index 4f7b34ae12..76c38e0fa8 100644 --- a/pkg/certifier/vexhub/vexhub.go +++ b/pkg/certifier/vexhub/vexhub.go @@ -1,5 +1,5 @@ // -// Copyright 2024 The GUAC Authors. +// Copyright 2026 The GUAC Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package vexhub import ( "archive/tar" + "bytes" "compress/gzip" "context" "errors" @@ -35,6 +36,7 @@ import ( "github.com/guacsec/guac/pkg/version" jsoniter "github.com/json-iterator/go" + packageurl "github.com/package-url/packageurl-go" "golang.org/x/time/rate" ) @@ -44,6 +46,21 @@ const ( DefaultManifestURL = "https://raw.githubusercontent.com/aquasecurity/vexhub/main/vex-repository.json" VEXHubCollector = "vexhub_certifier" rateLimit = 100 + + // httpTimeout is the deadline for any single HTTP request made by the certifier. + // This prevents goroutine pinning when the archive endpoint stalls. + httpTimeout = 5 * time.Minute + + // maxArchiveBytes is the total cumulative bytes we will buffer from the archive. + // Prevents OOM when the hub grows large (tar-bomb protection). + maxArchiveBytes = 512 * 1024 * 1024 // 512 MiB + + // maxEntryBytes is the per-file byte limit applied via io.LimitReader. + maxEntryBytes = 32 * 1024 * 1024 // 32 MiB + + // supportedSpecVersion is the only spec version we know how to parse. + // Forward-incompatible versions are logged and skipped. + supportedSpecVersion = "0.1" ) var rateLimitInterval = time.Second @@ -91,7 +108,12 @@ type vexHubCertifier struct { func NewVEXHubCertifier(manifestURL string) certifier.Certifier { limiter := rate.NewLimiter(rate.Every(rateLimitInterval), rateLimit) transport := clients.NewRateLimitedTransport(version.UATransport, limiter) - client := &http.Client{Transport: transport} + // Set an explicit Timeout so a stalled archive endpoint cannot pin the + // certifier goroutine indefinitely (P0 fix #2). + client := &http.Client{ + Transport: transport, + Timeout: httpTimeout, + } if manifestURL == "" { manifestURL = DefaultManifestURL } @@ -125,13 +147,13 @@ func (v *vexHubCertifier) CertifyComponent(ctx context.Context, rootComponent in return fmt.Errorf("failed to fetch VEX Hub manifest: %w", err) } - archiveURL, subdir := getArchiveURL(manifest) + archiveURL, subdir := getArchiveURL(logger, manifest) if archiveURL == "" { - logger.Infof("no archive location found in VEX Hub manifest") + logger.Infof("no compatible archive location found in VEX Hub manifest") return nil } - // Download and extract the archive, building a PURL→VEX doc map. + // Download and extract the archive, building a canonical-PURL→VEX doc map. vexDocs, err := downloadAndIndex(ctx, v.httpClient, archiveURL, subdir) if err != nil { return fmt.Errorf("failed to download VEX Hub archive: %w", err) @@ -175,39 +197,68 @@ func fetchManifest(ctx context.Context, client *http.Client, url string) (*Manif return &manifest, nil } -// getArchiveURL extracts the archive URL and optional subdirectory from the manifest. +// getArchiveURL extracts the archive URL and optional subdirectory from the manifest, +// selecting only entries whose spec_version matches supportedSpecVersion. // The subdirectory is specified after "//" in the URL (per VEX Repo Spec). // Example: "https://example.com/archive.tar.gz//vexhub-main" → URL + subdir "vexhub-main" -func getArchiveURL(manifest *Manifest) (archiveURL, subdir string) { +// +// Forward-incompatible spec versions are logged and skipped to avoid silently +// downloading archives that this parser cannot handle (P1 fix #2). +func getArchiveURL(logger interface{ Infof(string, ...interface{}) }, manifest *Manifest) (archiveURL, subdir string) { if len(manifest.Versions) == 0 { return "", "" } - // Use the first version with a location. for _, v := range manifest.Versions { - if len(v.Locations) > 0 { - rawURL := v.Locations[0].URL - // Look for "//" that is NOT part of the scheme (e.g., "https://"). - // The subdirectory separator always appears after the host/path portion. - schemeEnd := strings.Index(rawURL, "://") - searchStart := 0 - if schemeEnd >= 0 { - searchStart = schemeEnd + 3 - } - rest := rawURL[searchStart:] - if idx := strings.Index(rest, "//"); idx >= 0 { - archiveURL = rawURL[:searchStart+idx] - subdir = rest[idx+2:] - } else { - archiveURL = rawURL - } - return archiveURL, subdir + if v.SpecVersion != supportedSpecVersion { + logger.Infof("VEX Hub: skipping unsupported spec_version %q (supported: %q)", v.SpecVersion, supportedSpecVersion) + continue + } + if len(v.Locations) == 0 { + continue } + rawURL := v.Locations[0].URL + // Look for "//" that is NOT part of the scheme (e.g., "https://"). + // The subdirectory separator always appears after the host/path portion. + schemeEnd := strings.Index(rawURL, "://") + searchStart := 0 + if schemeEnd >= 0 { + searchStart = schemeEnd + 3 + } + rest := rawURL[searchStart:] + if idx := strings.Index(rest, "//"); idx >= 0 { + archiveURL = rawURL[:searchStart+idx] + subdir = rest[idx+2:] + } else { + archiveURL = rawURL + } + return archiveURL, subdir } return "", "" } -// downloadAndIndex downloads a tar.gz archive, extracts it, parses index.json, -// and returns a map of PURL→VEX document bytes. +// canonicalizePURL parses a PURL with packageurl-go and re-stringifies it so +// that qualifier ordering, namespace casing, and percent-encoding are all +// normalised before map keying (P1 fix #1). +// Returns the original string unchanged if parsing fails. +func canonicalizePURL(purl string) string { + p, err := packageurl.FromString(purl) + if err != nil { + return purl + } + return p.ToString() +} + +// downloadAndIndex downloads a tar.gz archive, uses a two-pass approach to +// safely extract only the files referenced in index.json, and returns a map +// of canonical-PURL → VEX document bytes. +// +// Two-pass design (P0 fix #1): +// 1. Stream the archive once, buffering only index.json and recording the +// names of files referenced by index.Packages. +// 2. Stream the archive a second time (from the in-memory copy), reading +// only those referenced files, each wrapped in io.LimitReader. +// +// This prevents unbounded memory growth (tar-bomb / OOM) as the hub scales. func downloadAndIndex(ctx context.Context, client *http.Client, archiveURL, subdir string) (map[string][]byte, error) { req, err := http.NewRequestWithContext(ctx, http.MethodGet, archiveURL, nil) if err != nil { @@ -223,35 +274,20 @@ func downloadAndIndex(ctx context.Context, client *http.Client, archiveURL, subd return nil, fmt.Errorf("archive download returned status %d", resp.StatusCode) } - gz, err := gzip.NewReader(resp.Body) + // Buffer the entire compressed archive so we can make two passes over it. + // We cap the raw download at maxArchiveBytes to avoid OOM from a huge tarball. + limitedBody := io.LimitReader(resp.Body, maxArchiveBytes+1) + archiveData, err := io.ReadAll(limitedBody) if err != nil { - return nil, fmt.Errorf("creating gzip reader: %w", err) + return nil, fmt.Errorf("buffering archive: %w", err) + } + if int64(len(archiveData)) > maxArchiveBytes { + return nil, fmt.Errorf("archive exceeds maximum allowed size of %d bytes", maxArchiveBytes) } - defer func() { _ = gz.Close() }() - - tr := tar.NewReader(gz) - - // First pass: extract all files into memory. - files := make(map[string][]byte) - for { - header, err := tr.Next() - if err == io.EOF { - break - } - if err != nil { - return nil, fmt.Errorf("reading tar entry: %w", err) - } - if header.Typeflag != tar.TypeReg { - continue - } - - data, err := io.ReadAll(tr) - if err != nil { - return nil, fmt.Errorf("reading file %s: %w", header.Name, err) - } - // Normalize path: strip the subdir prefix if present. - name := header.Name + // normalizeName strips the subdir prefix and the leading top-level directory + // component (e.g. "vexhub-main/") from a tar entry name. + normalizeName := func(name string) string { if subdir != "" { if after, found := strings.CutPrefix(name, subdir+"/"); found { name = after @@ -259,21 +295,49 @@ func downloadAndIndex(ctx context.Context, client *http.Client, archiveURL, subd name = after } } - // Also strip leading directory component from tar (e.g., "vexhub-main/") - if idx := strings.Index(name, "/"); idx >= 0 { - // Keep the path after the first directory component only if subdir wasn't already stripped. - if subdir == "" { + if subdir == "" { + if idx := strings.Index(name, "/"); idx >= 0 { name = name[idx+1:] } } - if name != "" { - files[name] = data + return name + } + + // openTar returns a fresh *tar.Reader over the buffered archive bytes. + openTar := func() (*tar.Reader, error) { + gz, err := gzip.NewReader(bytes.NewReader(archiveData)) + if err != nil { + return nil, fmt.Errorf("creating gzip reader: %w", err) } + return tar.NewReader(gz), nil } - // Parse index.json. - indexData, ok := files["index.json"] - if !ok { + // ── Pass 1: find and parse index.json only ──────────────────────────────── + tr1, err := openTar() + if err != nil { + return nil, err + } + var indexData []byte + for { + header, err := tr1.Next() + if err == io.EOF { + break + } + if err != nil { + return nil, fmt.Errorf("reading tar entry (pass 1): %w", err) + } + if header.Typeflag != tar.TypeReg { + continue + } + if normalizeName(header.Name) == "index.json" { + indexData, err = io.ReadAll(io.LimitReader(tr1, maxEntryBytes)) + if err != nil { + return nil, fmt.Errorf("reading index.json: %w", err) + } + break + } + } + if indexData == nil { return nil, fmt.Errorf("index.json not found in archive") } @@ -282,20 +346,62 @@ func downloadAndIndex(ctx context.Context, client *http.Client, archiveURL, subd return nil, fmt.Errorf("parsing index.json: %w", err) } - // Build PURL→VEX doc map. + // Build the set of file paths we actually need. + needed := make(map[string]bool, len(index.Packages)) + for _, pkg := range index.Packages { + needed[pkg.Location] = true + } + + // ── Pass 2: read only the referenced files ──────────────────────────────── + tr2, err := openTar() + if err != nil { + return nil, err + } + fileContents := make(map[string][]byte, len(needed)) + var cumulativeBytes int64 + for { + header, err := tr2.Next() + if err == io.EOF { + break + } + if err != nil { + return nil, fmt.Errorf("reading tar entry (pass 2): %w", err) + } + if header.Typeflag != tar.TypeReg { + continue + } + name := normalizeName(header.Name) + if !needed[name] { + continue + } + data, err := io.ReadAll(io.LimitReader(tr2, maxEntryBytes)) + if err != nil { + return nil, fmt.Errorf("reading file %s: %w", header.Name, err) + } + cumulativeBytes += int64(len(data)) + if cumulativeBytes > maxArchiveBytes { + return nil, fmt.Errorf("extracted content exceeds maximum allowed size of %d bytes", maxArchiveBytes) + } + fileContents[name] = data + } + + // Build canonical-PURL → VEX doc map. vexDocs := make(map[string][]byte, len(index.Packages)) for _, pkg := range index.Packages { - docData, ok := files[pkg.Location] + docData, ok := fileContents[pkg.Location] if !ok { continue } - vexDocs[pkg.ID] = docData + // Canonicalize the PURL from the index so lookups are consistent. + vexDocs[canonicalizePURL(pkg.ID)] = docData } return vexDocs, nil } // emitVEXDocuments looks up each PURL in the VEX index and emits matching documents. +// Both the query PURLs and the index keys are canonicalized via packageurl-go +// before comparison so that PURL spelling variants do not cause misses (P1 fix #1). func emitVEXDocuments(purls []string, vexDocs map[string][]byte, docChannel chan<- *processor.Document) ([]*processor.Document, error) { var emitted []*processor.Document seen := make(map[string]bool) @@ -305,13 +411,25 @@ func emitVEXDocuments(purls []string, vexDocs map[string][]byte, docChannel chan continue } - // Try exact PURL match first, then strip version for lookup. - lookupKey := purl - if _, ok := vexDocs[lookupKey]; !ok { - lookupKey = stripPurlVersion(purl) - } + // Canonicalize the query PURL so qualifier order, namespace casing, and + // percent-encoding all match the canonicalized index keys built in + // downloadAndIndex. + lookupKey := canonicalizePURL(purl) docData, ok := vexDocs[lookupKey] + if !ok { + // If an exact (canonicalized) match is not found, try without version + // by zeroing the Version field and re-stringifying. + p, err := packageurl.FromString(purl) + if err == nil { + p.Version = "" + p.Qualifiers = packageurl.Qualifiers{} + p.Subpath = "" + lookupKey = p.ToString() + docData, ok = vexDocs[lookupKey] + } + } + if !ok { continue } @@ -339,21 +457,3 @@ func emitVEXDocuments(purls []string, vexDocs map[string][]byte, docChannel chan return emitted, nil } - -// stripPurlVersion removes the version, qualifiers, and subpath from a PURL. -// e.g. "pkg:npm/lodash@4.17.21" → "pkg:npm/lodash" -func stripPurlVersion(purl string) string { - // Remove subpath (after last #) - if idx := strings.Index(purl, "#"); idx >= 0 { - purl = purl[:idx] - } - // Remove qualifiers (after ?) - if idx := strings.Index(purl, "?"); idx >= 0 { - purl = purl[:idx] - } - // Remove version (after @) - if idx := strings.LastIndex(purl, "@"); idx >= 0 { - purl = purl[:idx] - } - return purl -} diff --git a/pkg/certifier/vexhub/vexhub_test.go b/pkg/certifier/vexhub/vexhub_test.go index 4e75648e51..0adbf92393 100644 --- a/pkg/certifier/vexhub/vexhub_test.go +++ b/pkg/certifier/vexhub/vexhub_test.go @@ -1,5 +1,5 @@ // -// Copyright 2024 The GUAC Authors. +// Copyright 2026 The GUAC Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -23,20 +23,27 @@ import ( "fmt" "net/http" "net/http/httptest" + "strings" "testing" "github.com/guacsec/guac/pkg/certifier/components/root_package" "github.com/guacsec/guac/pkg/handler/processor" ) -// buildTestArchive creates a tar.gz archive with index.json and a VEX document. +// noopLogger satisfies the logger interface used by getArchiveURL in tests. +type noopLogger struct{} + +func (n noopLogger) Infof(_ string, _ ...interface{}) {} + +// buildTestArchive creates a tar.gz archive with index.json and optional VEX files. +// All entries are nested under a "vexhub-main/" top-level directory (simulating a +// GitHub archive download). func buildTestArchive(t *testing.T, indexJSON string, vexFiles map[string]string) []byte { t.Helper() var buf bytes.Buffer gw := gzip.NewWriter(&buf) tw := tar.NewWriter(gw) - // Write index.json under a root dir prefix (simulating GitHub archive). writeFile := func(name, content string) { hdr := &tar.Header{ Name: "vexhub-main/" + name, @@ -65,63 +72,102 @@ func buildTestArchive(t *testing.T, indexJSON string, vexFiles map[string]string return buf.Bytes() } -func TestStripPurlVersion(t *testing.T) { - tests := []struct { - input string - want string - }{ - {"pkg:npm/lodash@4.17.21", "pkg:npm/lodash"}, - {"pkg:maven/org.apache/log4j@2.0?type=jar", "pkg:maven/org.apache/log4j"}, - {"pkg:npm/lodash@4.17.21#sub/path", "pkg:npm/lodash"}, - {"pkg:deb/debian/curl", "pkg:deb/debian/curl"}, +// buildArchiveWithoutIndex creates a tar.gz that intentionally omits index.json. +func buildArchiveWithoutIndex(t *testing.T) []byte { + t.Helper() + var buf bytes.Buffer + gw := gzip.NewWriter(&buf) + tw := tar.NewWriter(gw) + hdr := &tar.Header{Name: "vexhub-main/other.json", Mode: 0644, Size: 2} + if err := tw.WriteHeader(hdr); err != nil { + t.Fatal(err) } - for _, tt := range tests { - t.Run(tt.input, func(t *testing.T) { - got := stripPurlVersion(tt.input) - if got != tt.want { - t.Errorf("stripPurlVersion(%q) = %q, want %q", tt.input, got, tt.want) - } - }) + if _, err := tw.Write([]byte("{}")); err != nil { + t.Fatal(err) } + if err := tw.Close(); err != nil { + t.Fatal(err) + } + if err := gw.Close(); err != nil { + t.Fatal(err) + } + return buf.Bytes() } +// TestGetArchiveURL verifies that getArchiveURL correctly filters spec versions +// and parses the // subdirectory separator. func TestGetArchiveURL(t *testing.T) { + logger := noopLogger{} tests := []struct { - name string - manifest Manifest - wantURL string + name string + manifest Manifest + wantURL string wantSubdir string }{ { - name: "simple URL", + name: "supported spec version simple URL", manifest: Manifest{ Versions: []ManifestVersion{{ - Locations: []ManifestLocation{{URL: "https://example.com/vex.tar.gz"}}, + SpecVersion: "0.1", + Locations: []ManifestLocation{{URL: "https://example.com/vex.tar.gz"}}, }}, }, wantURL: "https://example.com/vex.tar.gz", wantSubdir: "", }, { - name: "empty manifest", - manifest: Manifest{}, - wantURL: "", + name: "empty manifest", + manifest: Manifest{}, + wantURL: "", wantSubdir: "", }, { name: "URL with subdirectory", manifest: Manifest{ Versions: []ManifestVersion{{ - Locations: []ManifestLocation{{URL: "https://github.com/org/repo/archive/refs/heads/main.tar.gz//vexhub-main"}}, + SpecVersion: "0.1", + Locations: []ManifestLocation{{URL: "https://github.com/org/repo/archive/refs/heads/main.tar.gz//vexhub-main"}}, }}, }, wantURL: "https://github.com/org/repo/archive/refs/heads/main.tar.gz", wantSubdir: "vexhub-main", }, + { + // P1 fix: unknown spec version must be skipped; no URL returned. + name: "unsupported spec version is skipped", + manifest: Manifest{ + Versions: []ManifestVersion{ + { + SpecVersion: "99.0", + Locations: []ManifestLocation{{URL: "https://example.com/new.tar.gz"}}, + }, + }, + }, + wantURL: "", + wantSubdir: "", + }, + { + // First entry is unsupported, second entry is supported — should pick the second. + name: "multi-version manifest picks first compatible", + manifest: Manifest{ + Versions: []ManifestVersion{ + { + SpecVersion: "99.0", + Locations: []ManifestLocation{{URL: "https://example.com/new.tar.gz"}}, + }, + { + SpecVersion: "0.1", + Locations: []ManifestLocation{{URL: "https://example.com/old.tar.gz"}}, + }, + }, + }, + wantURL: "https://example.com/old.tar.gz", + wantSubdir: "", + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - gotURL, gotSubdir := getArchiveURL(&tt.manifest) + gotURL, gotSubdir := getArchiveURL(logger, &tt.manifest) if gotURL != tt.wantURL { t.Errorf("URL = %q, want %q", gotURL, tt.wantURL) } @@ -132,13 +178,17 @@ func TestGetArchiveURL(t *testing.T) { } } +// TestEmitVEXDocuments exercises PURL canonicalization, deduplication, guac +// skipping, and the version-stripped fallback. func TestEmitVEXDocuments(t *testing.T) { vexDoc := []byte(`{"@context": "https://openvex.dev/ns/v0.2.0", "@id": "test"}`) + + // Index keyed with canonical PURL (no version). vexDocs := map[string][]byte{ - "pkg:npm/lodash": vexDoc, + canonicalizePURL("pkg:npm/lodash"): vexDoc, } - t.Run("matching purl emits document", func(t *testing.T) { + t.Run("exact versioned purl matches via version-stripped fallback", func(t *testing.T) { docChan := make(chan *processor.Document, 10) docs, err := emitVEXDocuments( []string{"pkg:npm/lodash@4.17.21"}, @@ -198,8 +248,25 @@ func TestEmitVEXDocuments(t *testing.T) { t.Fatalf("expected 1 doc for duplicate purls, got %d", len(docs)) } }) + + t.Run("stripped-version-still-misses for unknown package", func(t *testing.T) { + // A versioned PURL that, even after stripping, has no index entry. + docs, err := emitVEXDocuments( + []string{"pkg:npm/totally-unknown@1.2.3"}, + vexDocs, + nil, + ) + if err != nil { + t.Fatal(err) + } + if len(docs) != 0 { + t.Fatalf("expected 0 docs for unknown package, got %d", len(docs)) + } + }) } +// TestCertifyComponentTypeMismatch checks that a wrong component type returns +// the expected sentinel error. func TestCertifyComponentTypeMismatch(t *testing.T) { c := &vexHubCertifier{httpClient: http.DefaultClient, manifestURL: "http://example.com"} err := c.CertifyComponent(context.Background(), "wrong type", nil) @@ -208,6 +275,8 @@ func TestCertifyComponentTypeMismatch(t *testing.T) { } } +// TestCertifyComponentEndToEnd runs the full certifier pipeline against an +// in-process httptest server serving a manifest and archive. func TestCertifyComponentEndToEnd(t *testing.T) { vexDoc := `{"@context":"https://openvex.dev/ns/v0.2.0","@id":"test-vex","statements":[{"vulnerability":{"name":"CVE-2021-44228"},"products":[{"@id":"pkg:npm/lodash"}],"status":"not_affected"}]}` indexJSON := `{"updated_at":"2024-01-01T00:00:00Z","packages":[{"id":"pkg:npm/lodash","location":"pkg/npm/lodash/vex.json"}]}` @@ -216,7 +285,6 @@ func TestCertifyComponentEndToEnd(t *testing.T) { "pkg/npm/lodash/vex.json": vexDoc, }) - // Serve the manifest and archive. mux := http.NewServeMux() mux.HandleFunc("/vex-repository.json", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") @@ -261,3 +329,146 @@ func TestCertifyComponentEndToEnd(t *testing.T) { t.Errorf("expected collector %q, got %q", VEXHubCollector, docs[0].SourceInformation.Collector) } } + +// TestMalformedManifest verifies that a non-JSON manifest body returns an error. +func TestMalformedManifest(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte("not valid json {{{{")) + })) + defer server.Close() + + c := &vexHubCertifier{ + httpClient: server.Client(), + manifestURL: server.URL + "/vex-repository.json", + } + err := c.CertifyComponent(context.Background(), []*root_package.PackageNode{{Purl: "pkg:npm/lodash@1.0"}}, nil) + if err == nil { + t.Fatal("expected error for malformed manifest, got nil") + } +} + +// TestMissingIndexJSON verifies that an archive without index.json returns an error. +func TestMissingIndexJSON(t *testing.T) { + archive := buildArchiveWithoutIndex(t) + + mux := http.NewServeMux() + mux.HandleFunc("/vex-repository.json", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprintf(w, `{"name":"test","versions":[{"spec_version":"0.1","locations":[{"url":"%s/archive.tar.gz"}]}]}`, "http://"+r.Host) + }) + mux.HandleFunc("/archive.tar.gz", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(archive) + }) + server := httptest.NewServer(mux) + defer server.Close() + + c := &vexHubCertifier{ + httpClient: server.Client(), + manifestURL: server.URL + "/vex-repository.json", + } + err := c.CertifyComponent(context.Background(), []*root_package.PackageNode{{Purl: "pkg:npm/lodash@1.0"}}, nil) + if err == nil || !strings.Contains(err.Error(), "index.json") { + t.Fatalf("expected index.json error, got: %v", err) + } +} + +// TestOversizedArchive verifies that an archive exceeding maxArchiveBytes is rejected. +func TestOversizedArchive(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/vex-repository.json", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprintf(w, `{"name":"test","versions":[{"spec_version":"0.1","locations":[{"url":"%s/archive.tar.gz"}]}]}`, "http://"+r.Host) + }) + mux.HandleFunc("/archive.tar.gz", func(w http.ResponseWriter, r *http.Request) { + // Send more than maxArchiveBytes of zeros (uncompressed, but enough to trigger the limit check). + chunk := make([]byte, 1024*1024) // 1 MiB chunk + for i := 0; i <= int(maxArchiveBytes/int64(len(chunk)))+1; i++ { + _, _ = w.Write(chunk) + } + }) + server := httptest.NewServer(mux) + defer server.Close() + + c := &vexHubCertifier{ + httpClient: server.Client(), + manifestURL: server.URL + "/vex-repository.json", + } + err := c.CertifyComponent(context.Background(), []*root_package.PackageNode{{Purl: "pkg:npm/lodash@1.0"}}, nil) + if err == nil { + t.Fatal("expected error for oversized archive, got nil") + } +} + +// TestNetworkError verifies that a network failure during archive download +// is surfaced as an error from CertifyComponent. +func TestNetworkError(t *testing.T) { + mux := http.NewServeMux() + mux.HandleFunc("/vex-repository.json", func(w http.ResponseWriter, r *http.Request) { + _, _ = fmt.Fprintf(w, `{"name":"test","versions":[{"spec_version":"0.1","locations":[{"url":"%s/archive.tar.gz"}]}]}`, "http://"+r.Host) + }) + mux.HandleFunc("/archive.tar.gz", func(w http.ResponseWriter, r *http.Request) { + // Hijack the connection to force a network error mid-response. + hj, ok := w.(http.Hijacker) + if !ok { + http.Error(w, "no hijack", http.StatusInternalServerError) + return + } + conn, _, _ := hj.Hijack() + _ = conn.Close() + }) + server := httptest.NewServer(mux) + defer server.Close() + + c := &vexHubCertifier{ + httpClient: server.Client(), + manifestURL: server.URL + "/vex-repository.json", + } + err := c.CertifyComponent(context.Background(), []*root_package.PackageNode{{Purl: "pkg:npm/lodash@1.0"}}, nil) + if err == nil { + t.Fatal("expected error for network failure, got nil") + } +} + +// TestMultiVersionManifest checks that when the manifest has multiple versions +// the certifier picks the first one with the supported spec_version. +func TestMultiVersionManifest(t *testing.T) { + vexDoc := `{"@context":"https://openvex.dev/ns/v0.2.0","@id":"multi-test"}` + indexJSON := `{"updated_at":"2024-01-01T00:00:00Z","packages":[{"id":"pkg:npm/lodash","location":"pkg/npm/lodash/vex.json"}]}` + archive := buildTestArchive(t, indexJSON, map[string]string{ + "pkg/npm/lodash/vex.json": vexDoc, + }) + + mux := http.NewServeMux() + mux.HandleFunc("/vex-repository.json", func(w http.ResponseWriter, r *http.Request) { + // Serve an unsupported version first; the certifier should skip it. + _, _ = fmt.Fprintf(w, `{"name":"test","versions":[ + {"spec_version":"99.0","locations":[{"url":"%s/should-not-be-used.tar.gz"}]}, + {"spec_version":"0.1","locations":[{"url":"%s/archive.tar.gz"}]} + ]}`, "http://"+r.Host, "http://"+r.Host) + }) + mux.HandleFunc("/archive.tar.gz", func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(archive) + }) + // Crash if the wrong archive is requested. + mux.HandleFunc("/should-not-be-used.tar.gz", func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "wrong version selected", http.StatusBadRequest) + }) + server := httptest.NewServer(mux) + defer server.Close() + + c := &vexHubCertifier{ + httpClient: server.Client(), + manifestURL: server.URL + "/vex-repository.json", + } + docChan := make(chan *processor.Document, 10) + err := c.CertifyComponent(context.Background(), []*root_package.PackageNode{{Purl: "pkg:npm/lodash@4.17.21"}}, docChan) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + close(docChan) + var docs []*processor.Document + for doc := range docChan { + docs = append(docs, doc) + } + if len(docs) != 1 { + t.Fatalf("expected 1 document, got %d", len(docs)) + } +}