From 20b241bc068e57cb0ff7b11b6bfaa6a8ea0bee72 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tymoteusz=20G=C5=82uch?= Date: Mon, 24 Aug 2026 22:46:52 +0200 Subject: [PATCH 1/2] fix: Db structure fixes --- batch.json | 0 cmd/api-service/main.go | 3 + cmd/replicator-service/main.go | 9 ++- internal/adaptive-worker/constants.go | 2 +- internal/adaptive-worker/controller.go | 13 ++- internal/adaptive-worker/job-controler.go | 6 ++ internal/adaptive-worker/models.go | 2 +- internal/npm/constants.go | 3 + internal/npm/http.go | 40 ++++++++++ internal/npm/models.go | 12 +++ internal/npm/service.go | 22 ++++- internal/npm/sql/insert_packages_batch.sql | 6 +- internal/osv/constants.go | 6 ++ internal/osv/http.go | 55 +++++++++++++ internal/osv/models.go | 80 +++++++++++++++++++ internal/packages-data-collector/collector.go | 1 + internal/packages-data-collector/models.go | 1 + .../packages-data-collector/repository.go | 1 + internal/packages-data-collector/service.go | 1 + .../0004_create_package_data_table.down.sql | 1 + .../0004_create_package_data_table.up.sql | 9 +++ ...005_update_vulnerabilities_tables.down.sql | 33 ++++++++ .../0005_update_vulnerabilities_tables.up.sql | 64 +++++++++++++++ ...ust_tables_to_multiple_ecosystems.down.sql | 47 +++++++++++ ...djust_tables_to_multiple_ecosystems.up.sql | 43 ++++++++++ 25 files changed, 448 insertions(+), 12 deletions(-) delete mode 100644 batch.json create mode 100644 cmd/api-service/main.go create mode 100644 internal/adaptive-worker/job-controler.go create mode 100644 internal/osv/constants.go create mode 100644 internal/osv/http.go create mode 100644 internal/osv/models.go create mode 100644 internal/packages-data-collector/collector.go create mode 100644 internal/packages-data-collector/models.go create mode 100644 internal/packages-data-collector/repository.go create mode 100644 internal/packages-data-collector/service.go create mode 100644 migrations/0004_create_package_data_table.down.sql create mode 100644 migrations/0004_create_package_data_table.up.sql create mode 100644 migrations/0005_update_vulnerabilities_tables.down.sql create mode 100644 migrations/0005_update_vulnerabilities_tables.up.sql create mode 100644 migrations/0006_adjust_tables_to_multiple_ecosystems.down.sql create mode 100644 migrations/0006_adjust_tables_to_multiple_ecosystems.up.sql diff --git a/batch.json b/batch.json deleted file mode 100644 index e69de29..0000000 diff --git a/cmd/api-service/main.go b/cmd/api-service/main.go new file mode 100644 index 0000000..38dd16d --- /dev/null +++ b/cmd/api-service/main.go @@ -0,0 +1,3 @@ +package main + +func main() {} diff --git a/cmd/replicator-service/main.go b/cmd/replicator-service/main.go index 23d9a30..d1b2768 100644 --- a/cmd/replicator-service/main.go +++ b/cmd/replicator-service/main.go @@ -19,11 +19,12 @@ func main() { worker, startWorker := adaptiveWorker.New(ctx) worker.ScheduleJob(adaptiveWorker.Job{ - Handler: func(concurrency int) { - npm.Pipeline(concurrency) + Handler: func(concurrency int, controller *adaptiveWorker.JobController) { + npm.Pipeline(concurrency, controller) + }, - SkippedCycles: 0, - ReservationRatio: 1, + SkippedCycles: 29, + ReservationRatio: 0, }) startWorker() diff --git a/internal/adaptive-worker/constants.go b/internal/adaptive-worker/constants.go index 7823187..3fb1aa5 100644 --- a/internal/adaptive-worker/constants.go +++ b/internal/adaptive-worker/constants.go @@ -6,7 +6,7 @@ const ( maxWorkerHeadroom = 0.2 - cycleTime = "@every 30s" + cycleTime = "@every 1s" skipCount = 2 diff --git a/internal/adaptive-worker/controller.go b/internal/adaptive-worker/controller.go index 7af0a59..6cdae06 100644 --- a/internal/adaptive-worker/controller.go +++ b/internal/adaptive-worker/controller.go @@ -40,7 +40,7 @@ func (controller *Controller) calculateWorkers() error { fmt.Println("cpu percent: ", resourceUsage.cpuPercent) fmt.Println("loadAvg: ", resourceUsage.loadAvg) fmt.Println("load percent: ", resourceUsage.loadPercent) - fmt.Println("rem percent: ", resourceUsage.ramPercent) + fmt.Println("ram percent: ", resourceUsage.ramPercent) fmt.Println("resources usage - END-----------------------<") maxSimultaneousWorkers := getMaxWorkers(resourceUsage.cpuCount) @@ -108,7 +108,16 @@ func (controller *Controller) runCycle() { inferMinSimultaneousWorkers(controller.currentCycle, controller.jobs), ) - currentJob.scheduledJob.Handler(currentJob.concurrency) + if currentJob.canRun { + currentJob.scheduledJob.Handler(currentJob.concurrency, &JobController{ + AdjustReservationRatioHandler: func(reservationRatio float64) { + currentJob.scheduledJob.ReservationRatio = reservationRatio + }, + AdjustSkippedCyclesHandler: func(skippedCycles int) { + currentJob.scheduledJob.SkippedCycles = skippedCycles + }, + }) + } } } diff --git a/internal/adaptive-worker/job-controler.go b/internal/adaptive-worker/job-controler.go new file mode 100644 index 0000000..ea3faa3 --- /dev/null +++ b/internal/adaptive-worker/job-controler.go @@ -0,0 +1,6 @@ +package adaptiveWorker + +type JobController struct { + AdjustReservationRatioHandler func(float64) + AdjustSkippedCyclesHandler func(int) +} diff --git a/internal/adaptive-worker/models.go b/internal/adaptive-worker/models.go index d93e05c..f38c040 100644 --- a/internal/adaptive-worker/models.go +++ b/internal/adaptive-worker/models.go @@ -14,7 +14,7 @@ type calculatedResourcesUsage struct { ramPercent float64 } -type jobHandler func(concurrency int) +type jobHandler func(concurrency int, controller *JobController) type Job struct { Handler jobHandler diff --git a/internal/npm/constants.go b/internal/npm/constants.go index 36ac6dd..328fbff 100644 --- a/internal/npm/constants.go +++ b/internal/npm/constants.go @@ -3,4 +3,7 @@ package npm const ( npmReplicationUrl = "https://replicate.npmjs.com/registry/_changes" limit = 10000 + + npmDownloadUrl = "https://api.npmjs.org/downloads/point" + defaultNpmDownloadPeriod = "last-month" ) diff --git a/internal/npm/http.go b/internal/npm/http.go index 127286b..e0074c7 100644 --- a/internal/npm/http.go +++ b/internal/npm/http.go @@ -2,6 +2,7 @@ package npm import ( "encoding/json" + "fmt" "net/http" "strconv" ) @@ -31,3 +32,42 @@ func fetchNpmPackages(since int) (*NpmChangesResponse, error) { return &packagesData, nil } + +func FetchNpmPackageDownloads(payload FetchNpmPackageDownloadsPayload) (*FetchNpmPackageDownloadsResponse, error) { + var since string + if payload.since == nil { + since = defaultNpmDownloadPeriod + } else { + since = *payload.since + } + + url := npmDownloadUrl + "/" + since + "/" + payload.packageName + + request, err := http.NewRequest(http.MethodGet, url, nil) + if err != nil { + return nil, err + } + + client := &http.Client{} + + response, err := client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + + if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf( + "npm download API returned status %s", + response.Status, + ) + } + + var packageData FetchNpmPackageDownloadsResponse + + if err := json.NewDecoder(response.Body).Decode(&packageData); err != nil { + return nil, err + } + + return &packageData, nil +} diff --git a/internal/npm/models.go b/internal/npm/models.go index 62a3130..7181cf0 100644 --- a/internal/npm/models.go +++ b/internal/npm/models.go @@ -23,3 +23,15 @@ type dbDropPackagesBatchPayload struct { type dbInsertPackagesBatchPayload struct { packages []string } + +type FetchNpmPackageDownloadsPayload struct { + packageName string + since *string +} + +type FetchNpmPackageDownloadsResponse struct { + Downloads int `json:"downloads"` + End string `json:"end"` + Package string `json:"package"` + Start string `json:"start"` +} diff --git a/internal/npm/service.go b/internal/npm/service.go index 2f5c868..a0fad0f 100644 --- a/internal/npm/service.go +++ b/internal/npm/service.go @@ -2,12 +2,16 @@ package npm import ( "context" + "encoding/json" + "fmt" + "os" "sync" "github.com/trustpkg/trustpkg-api/db" + adaptiveWorker "github.com/trustpkg/trustpkg-api/internal/adaptive-worker" ) -func Pipeline(concurrency int) error { +func Pipeline(concurrency int, JobController *adaptiveWorker.JobController) error { ctx := context.Background() if concurrency < 1 { @@ -39,12 +43,28 @@ func Pipeline(concurrency int) error { if err := savePackagesPage(ctx, changes); err != nil { errs <- err } + + fmt.Print("---------------------->\n") + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + enc.Encode(response.Results) + + fmt.Println("response", response.LastSeq) + fmt.Print("----------------------<\n") }(response.Results) } waitGroup.Wait() close(errs) + data, err := FetchNpmPackageDownloads(FetchNpmPackageDownloadsPayload{ + packageName: "next", + }) + + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + enc.Encode(data) + for err := range errs { if err != nil { return err diff --git a/internal/npm/sql/insert_packages_batch.sql b/internal/npm/sql/insert_packages_batch.sql index fc96bf7..c7be98b 100644 --- a/internal/npm/sql/insert_packages_batch.sql +++ b/internal/npm/sql/insert_packages_batch.sql @@ -1,3 +1,3 @@ -INSERT INTO packages (name) -SELECT unnest($1::text[]) -ON CONFLICT (name) DO NOTHING; \ No newline at end of file +INSERT INTO packages (name, ecosystem) +SELECT unnest($1::text[]), 'npm' +ON CONFLICT (ecosystem, name) DO NOTHING; \ No newline at end of file diff --git a/internal/osv/constants.go b/internal/osv/constants.go new file mode 100644 index 0000000..d07ffe3 --- /dev/null +++ b/internal/osv/constants.go @@ -0,0 +1,6 @@ +package osv + +const ( + osvPackageVulnerabilitiesUrl = "https://api.osv.dev/v1/query" + defaultOsvEcosystem = "npm" +) diff --git a/internal/osv/http.go b/internal/osv/http.go new file mode 100644 index 0000000..beaef2e --- /dev/null +++ b/internal/osv/http.go @@ -0,0 +1,55 @@ +package osv + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" +) + +func FetchPackageVulnerabilities(payload FetchPackageVulnerabilitiesPayload) (*FetchPackageVulnerabilitiesResponse, error) { + var ecosystem string + if payload.Ecosystem != nil { + ecosystem = *payload.Ecosystem + } else { + ecosystem = defaultOsvEcosystem + } + + url := osvPackageVulnerabilitiesUrl + requestPayload := OsvQuery{ + OsvPackage: OsvQueryPackage{ + Name: payload.Name, + Ecosystem: ecosystem, + }, + } + + body, err := json.Marshal(requestPayload) + if err != nil { + return nil, fmt.Errorf("marshal request: %w", err) + } + + request, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return nil, err + } + + request.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(request) + if err != nil { + return nil, fmt.Errorf("send request: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("OSV returned %s", resp.Status) + } + + var result FetchPackageVulnerabilitiesResponse + + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("decode response: %w", err) + } + + return &result, nil +} diff --git a/internal/osv/models.go b/internal/osv/models.go new file mode 100644 index 0000000..9686fbd --- /dev/null +++ b/internal/osv/models.go @@ -0,0 +1,80 @@ +package osv + +import "time" + +type FetchPackageVulnerabilitiesPayload struct { + Name string + Ecosystem *string +} + +type OsvVuln struct { + Affected []OsvAffected `json:"affected"` + Aliases []string `json:"aliases"` + DatabaseSpecific OsvDatabaseSpecific `json:"database_specific"` + Details string `json:"details"` + ID string `json:"id"` + Modified time.Time `json:"modified"` + Published time.Time `json:"published"` + References []OsvReference `json:"references"` + SchemaVersion string `json:"schema_version"` + Severity []OsvSeverity `json:"severity"` + Summary string `json:"summary"` +} + +type OsvAffected struct { + DatabaseSpecific OsvAffectedDatabaseSpecific `json:"database_specific"` + Package OsvPackage `json:"package"` + Ranges []OsvRange `json:"ranges"` +} + +type OsvAffectedDatabaseSpecific struct { + Source string `json:"source"` +} + +type OsvPackage struct { + Ecosystem string `json:"ecosystem"` + Name string `json:"name"` + PURL string `json:"purl"` +} + +type OsvRange struct { + Events []OsvEvent `json:"events"` + Type string `json:"type"` +} + +type OsvEvent struct { + Introduced string `json:"introduced,omitempty"` + Fixed string `json:"fixed,omitempty"` + LastAffected string `json:"last_affected,omitempty"` +} + +type OsvDatabaseSpecific struct { + CWEIDs []string `json:"cwe_ids"` + GitHubReviewed bool `json:"github_reviewed"` + GitHubReviewedAt *time.Time `json:"github_reviewed_at,omitempty"` + NVDPublishedAt *time.Time `json:"nvd_published_at,omitempty"` + Severity string `json:"severity"` +} + +type OsvReference struct { + Type string `json:"type"` + URL string `json:"url"` +} + +type OsvSeverity struct { + Score string `json:"score"` + Type string `json:"type"` +} + +type FetchPackageVulnerabilitiesResponse struct { + Vulns any `json:"vulns"` +} + +type OsvQueryPackage struct { + Name string `json:"name"` + Ecosystem string `json:"ecosystem"` +} + +type OsvQuery struct { + OsvPackage OsvQueryPackage `json:"package"` +} diff --git a/internal/packages-data-collector/collector.go b/internal/packages-data-collector/collector.go new file mode 100644 index 0000000..07e10f4 --- /dev/null +++ b/internal/packages-data-collector/collector.go @@ -0,0 +1 @@ +package packagesDataCollector \ No newline at end of file diff --git a/internal/packages-data-collector/models.go b/internal/packages-data-collector/models.go new file mode 100644 index 0000000..43c9366 --- /dev/null +++ b/internal/packages-data-collector/models.go @@ -0,0 +1 @@ +package packagesDataCollector diff --git a/internal/packages-data-collector/repository.go b/internal/packages-data-collector/repository.go new file mode 100644 index 0000000..43c9366 --- /dev/null +++ b/internal/packages-data-collector/repository.go @@ -0,0 +1 @@ +package packagesDataCollector diff --git a/internal/packages-data-collector/service.go b/internal/packages-data-collector/service.go new file mode 100644 index 0000000..07e10f4 --- /dev/null +++ b/internal/packages-data-collector/service.go @@ -0,0 +1 @@ +package packagesDataCollector \ No newline at end of file diff --git a/migrations/0004_create_package_data_table.down.sql b/migrations/0004_create_package_data_table.down.sql new file mode 100644 index 0000000..8c98bde --- /dev/null +++ b/migrations/0004_create_package_data_table.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS packages_data; \ No newline at end of file diff --git a/migrations/0004_create_package_data_table.up.sql b/migrations/0004_create_package_data_table.up.sql new file mode 100644 index 0000000..2a7b8ca --- /dev/null +++ b/migrations/0004_create_package_data_table.up.sql @@ -0,0 +1,9 @@ +CREATE TABLE IF NOT EXISTS packages_data ( + package_id BIGINT PRIMARY KEY + REFERENCES packages(id) + ON DELETE CASCADE, + is_deprecated BOOLEAN NOT NULL DEFAULT FALSE, + need_update BOOLEAN NOT NULL DEFAULT FALSE, + popularity INTEGER NOT NULL DEFAULT 0, + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); \ No newline at end of file diff --git a/migrations/0005_update_vulnerabilities_tables.down.sql b/migrations/0005_update_vulnerabilities_tables.down.sql new file mode 100644 index 0000000..adc0069 --- /dev/null +++ b/migrations/0005_update_vulnerabilities_tables.down.sql @@ -0,0 +1,33 @@ +BEGIN; + +DROP TABLE IF EXISTS vulnerability_references; +DROP TABLE IF EXISTS vulnerability_affected; +DROP TABLE IF EXISTS package_vulnerabilities; +DROP TABLE IF EXISTS vulnerabilities; + +CREATE TABLE IF NOT EXISTS vulnerabilities ( + id BIGSERIAL PRIMARY KEY, + osv_id TEXT NOT NULL UNIQUE, + cve_id TEXT, + summary TEXT, + description TEXT, + severity TEXT, + cvss_score NUMERIC(3, 1), + cvss_vector TEXT, + affected_ranges TEXT[], + fixed_versions TEXT[], + published_at TIMESTAMP, + modified_at TIMESTAMP, + reference_urls TEXT[] +); + +CREATE TABLE IF NOT EXISTS package_vulnerabilities ( + package_id BIGINT NOT NULL + REFERENCES packages(id) ON DELETE CASCADE, + + vulnerability_id BIGINT NOT NULL + REFERENCES vulnerabilities(id) ON DELETE CASCADE, + + PRIMARY KEY (package_id, vulnerability_id) +); +COMMIT; \ No newline at end of file diff --git a/migrations/0005_update_vulnerabilities_tables.up.sql b/migrations/0005_update_vulnerabilities_tables.up.sql new file mode 100644 index 0000000..4b2447a --- /dev/null +++ b/migrations/0005_update_vulnerabilities_tables.up.sql @@ -0,0 +1,64 @@ +BEGIN; + +DROP TABLE IF EXISTS package_vulnerabilities; +DROP TABLE IF EXISTS vulnerabilities; + +CREATE TABLE vulnerabilities ( + id BIGSERIAL PRIMARY KEY, + osv_id TEXT NOT NULL UNIQUE, + cve_id TEXT, + summary TEXT, + description TEXT, + severity TEXT, + cvss_score NUMERIC(3, 1), + cvss_vector TEXT, + published_at TIMESTAMP, + modified_at TIMESTAMP +); + +CREATE TABLE vulnerability_affected ( + id BIGSERIAL PRIMARY KEY, + + vulnerability_id BIGINT NOT NULL + REFERENCES vulnerabilities(id) + ON DELETE CASCADE, + + ecosystem TEXT NOT NULL, + package_name TEXT NOT NULL, + + introduced_version TEXT, + fixed_version TEXT, + last_affected_version TEXT, + + UNIQUE ( + vulnerability_id, + ecosystem, + package_name, + introduced_version, + fixed_version, + last_affected_version + ) +); + +CREATE INDEX idx_vulnerability_affected_package + ON vulnerability_affected (ecosystem, package_name); + +CREATE INDEX idx_vulnerability_affected_vulnerability + ON vulnerability_affected (vulnerability_id); + +CREATE TABLE vulnerability_references ( + id BIGSERIAL PRIMARY KEY, + + vulnerability_id BIGINT NOT NULL + REFERENCES vulnerabilities(id) + ON DELETE CASCADE, + + url TEXT NOT NULL, + + UNIQUE (vulnerability_id, url) +); + +CREATE INDEX idx_vulnerability_references_vulnerability + ON vulnerability_references (vulnerability_id); + +COMMIT; \ No newline at end of file diff --git a/migrations/0006_adjust_tables_to_multiple_ecosystems.down.sql b/migrations/0006_adjust_tables_to_multiple_ecosystems.down.sql new file mode 100644 index 0000000..5764dc5 --- /dev/null +++ b/migrations/0006_adjust_tables_to_multiple_ecosystems.down.sql @@ -0,0 +1,47 @@ +BEGIN; + +ALTER TABLE vulnerability_affected + DROP CONSTRAINT vulnerability_affected_unique; + +ALTER TABLE vulnerability_affected + ADD COLUMN ecosystem TEXT NOT NULL DEFAULT 'npm'; + +ALTER TABLE vulnerability_affected + ADD COLUMN package_name TEXT NOT NULL DEFAULT ''; + +ALTER TABLE vulnerability_affected + DROP CONSTRAINT vulnerability_affected_package_id_fkey; + +DROP INDEX IF EXISTS idx_vulnerability_affected_package_id; + +ALTER TABLE vulnerability_affected + DROP COLUMN package_id; + +ALTER TABLE vulnerability_affected + ADD CONSTRAINT vulnerability_affected_vulnerability_id_ecosystem_package_name_introduced_version_fixed_version_last_affected_version_key + UNIQUE ( + vulnerability_id, + ecosystem, + package_name, + introduced_version, + fixed_version, + last_affected_version + ); + +ALTER TABLE vulnerability_affected + ALTER COLUMN ecosystem DROP DEFAULT; + +ALTER TABLE vulnerability_affected + ALTER COLUMN package_name DROP DEFAULT; + +ALTER TABLE packages + DROP CONSTRAINT packages_ecosystem_name_key; + +ALTER TABLE packages + ADD CONSTRAINT packages_name_key + UNIQUE (name); + +ALTER TABLE packages + DROP COLUMN ecosystem; + +COMMIT; \ No newline at end of file diff --git a/migrations/0006_adjust_tables_to_multiple_ecosystems.up.sql b/migrations/0006_adjust_tables_to_multiple_ecosystems.up.sql new file mode 100644 index 0000000..6a57ac9 --- /dev/null +++ b/migrations/0006_adjust_tables_to_multiple_ecosystems.up.sql @@ -0,0 +1,43 @@ +BEGIN; + +ALTER TABLE packages + ADD COLUMN ecosystem TEXT NOT NULL DEFAULT 'npm'; + +ALTER TABLE packages + DROP CONSTRAINT packages_name_key; + +ALTER TABLE packages + ADD CONSTRAINT packages_ecosystem_name_key + UNIQUE (ecosystem, name); + +ALTER TABLE packages + ALTER COLUMN ecosystem DROP DEFAULT; + + +ALTER TABLE vulnerability_affected + ADD COLUMN package_id BIGINT + REFERENCES packages(id) + ON DELETE CASCADE; + +CREATE INDEX idx_vulnerability_affected_package_id + ON vulnerability_affected(package_id); + +ALTER TABLE vulnerability_affected + DROP CONSTRAINT IF EXISTS + vulnerability_affected_vulnerability_id_ecosystem_package_n_key; + +ALTER TABLE vulnerability_affected + DROP COLUMN ecosystem, + DROP COLUMN package_name; + +ALTER TABLE vulnerability_affected + ADD CONSTRAINT vulnerability_affected_unique + UNIQUE ( + vulnerability_id, + package_id, + introduced_version, + fixed_version, + last_affected_version + ); + +COMMIT; \ No newline at end of file From 96f99f6f3d9fe5e98d92c374b7c500f416ad4298 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Tymoteusz=20G=C5=82uch?= Date: Wed, 26 Aug 2026 23:39:50 +0200 Subject: [PATCH 2/2] feat: Add init osv data script --- Makefile | 4 + cmd/first-osv-init-service/main.go | 19 +++ internal/osv/constants.go | 3 + internal/osv/helpers.go | 130 +++++++++++++++++ internal/osv/helpers_test.go | 51 +++++++ internal/osv/http.go | 39 ++++++ internal/osv/models.go | 4 + internal/osv/progress.go | 132 ++++++++++++++++++ internal/osv/replicator.go | 103 ++++++++++++++ internal/osv/repository.go | 70 ++++++++++ .../osv/sql/delete_vulnerability_children.sql | 2 + .../sql/delete_vulnerability_references.sql | 2 + internal/osv/sql/insert_affected.sql | 14 ++ internal/osv/sql/insert_reference.sql | 3 + internal/osv/sql/upsert_package.sql | 4 + internal/osv/sql/upsert_vulnerability.sql | 15 ++ 16 files changed, 595 insertions(+) create mode 100644 cmd/first-osv-init-service/main.go create mode 100644 internal/osv/helpers.go create mode 100644 internal/osv/helpers_test.go create mode 100644 internal/osv/progress.go create mode 100644 internal/osv/replicator.go create mode 100644 internal/osv/repository.go create mode 100644 internal/osv/sql/delete_vulnerability_children.sql create mode 100644 internal/osv/sql/delete_vulnerability_references.sql create mode 100644 internal/osv/sql/insert_affected.sql create mode 100644 internal/osv/sql/insert_reference.sql create mode 100644 internal/osv/sql/upsert_package.sql create mode 100644 internal/osv/sql/upsert_vulnerability.sql diff --git a/Makefile b/Makefile index e6b0d7d..705973c 100644 --- a/Makefile +++ b/Makefile @@ -16,6 +16,10 @@ dropBin: replicator-service: air -c configs/air/replicator-service.toml +.PHONY: init-osv-replication +init-osv-replication: + go run ./cmd/first-osv-init-service/main.go + .PHONY: migrate-up migrate-up: migrate -path migrations -database $(DB_URL) up diff --git a/cmd/first-osv-init-service/main.go b/cmd/first-osv-init-service/main.go new file mode 100644 index 0000000..1ba7e54 --- /dev/null +++ b/cmd/first-osv-init-service/main.go @@ -0,0 +1,19 @@ +package main + +import ( + "context" + "log" + + "github.com/trustpkg/trustpkg-api/db" + "github.com/trustpkg/trustpkg-api/internal/osv" +) + +func main() { + db.ConnectDb() + + if err := osv.ReplicateEcosystem(context.Background(), osv.FetchOSVEcosystemDumpPayload{ + Ecosystem: "npm", + }); err != nil { + log.Printf("OSV replication failed: %v", err) + } +} diff --git a/internal/osv/constants.go b/internal/osv/constants.go index d07ffe3..9f9f7b3 100644 --- a/internal/osv/constants.go +++ b/internal/osv/constants.go @@ -3,4 +3,7 @@ package osv const ( osvPackageVulnerabilitiesUrl = "https://api.osv.dev/v1/query" defaultOsvEcosystem = "npm" + + npmOsvDumpUrl = "https://storage.googleapis.com/osv-vulnerabilities" + npmOsvDumpUrlSuffix = "all.zip" ) diff --git a/internal/osv/helpers.go b/internal/osv/helpers.go new file mode 100644 index 0000000..3bbcbd6 --- /dev/null +++ b/internal/osv/helpers.go @@ -0,0 +1,130 @@ +package osv + +import ( + "strconv" + "strings" + "time" +) + +type ParsedVulnerability struct { + OSVID string + CVEID *string + Summary *string + Description *string + Severity *string + CVSSScore *float64 + CVSSVector *string + PublishedAt *time.Time + ModifiedAt *time.Time + + Affected []AffectedPackage + References []string +} + +type AffectedPackage struct { + Ecosystem string + PackageName string + IntroducedVersion *string + FixedVersion *string + LastAffectedVersion *string +} + +func parseVulnerability(raw OsvVuln) ParsedVulnerability { + parsed := ParsedVulnerability{OSVID: raw.ID, Summary: optionalString(raw.Summary), Description: optionalString(raw.Details), PublishedAt: optionalTime(raw.Published), ModifiedAt: optionalTime(raw.Modified)} + for _, alias := range raw.Aliases { + if strings.HasPrefix(alias, "CVE-") { + parsed.CVEID = optionalString(alias) + break + } + } + for _, severity := range raw.Severity { + if parsed.CVSSVector == nil && severity.Score != "" { + parsed.CVSSVector = optionalString(severity.Score) + } + if score, err := strconv.ParseFloat(severity.Score, 64); err == nil { + parsed.CVSSScore = &score + } + } + if raw.DatabaseSpecific.Severity != "" { + parsed.Severity = optionalString(raw.DatabaseSpecific.Severity) + } + for _, affected := range raw.Affected { + for _, versionRange := range affected.Ranges { + for _, version := range parseAffectedRange(versionRange.Events) { + parsed.Affected = append(parsed.Affected, AffectedPackage{ + Ecosystem: affected.Package.Ecosystem, PackageName: affected.Package.Name, + IntroducedVersion: version.introduced, + FixedVersion: version.fixed, + LastAffectedVersion: version.lastAffected, + }) + } + } + } + for _, reference := range raw.References { + if reference.URL != "" { + parsed.References = append(parsed.References, reference.URL) + } + } + return parsed +} + +type affectedVersion struct { + introduced *string + fixed *string + lastAffected *string +} + +func parseAffectedRange(events []OsvEvent) []affectedVersion { + versions := make([]affectedVersion, 0, len(events)) + var introduced *string + + for _, event := range events { + if event.Introduced != "" { + if introduced != nil { + versions = append(versions, affectedVersion{introduced: introduced}) + } + introduced = optionalString(event.Introduced) + } + + if event.Fixed != "" { + if introduced == nil { + introduced = optionalString("0") + } + versions = append(versions, affectedVersion{ + introduced: introduced, + fixed: optionalString(event.Fixed), + }) + introduced = nil + } + + if event.LastAffected != "" { + if introduced == nil { + introduced = optionalString("0") + } + versions = append(versions, affectedVersion{ + introduced: introduced, + lastAffected: optionalString(event.LastAffected), + }) + introduced = nil + } + } + + if introduced != nil { + versions = append(versions, affectedVersion{introduced: introduced}) + } + return versions +} + +func optionalString(value string) *string { + if value == "" { + return nil + } + return &value +} + +func optionalTime(value time.Time) *time.Time { + if value.IsZero() { + return nil + } + return &value +} diff --git a/internal/osv/helpers_test.go b/internal/osv/helpers_test.go new file mode 100644 index 0000000..edf90d4 --- /dev/null +++ b/internal/osv/helpers_test.go @@ -0,0 +1,51 @@ +package osv + +import "testing" + +func TestParseAffectedRangeBuildsIntervals(t *testing.T) { + versions := parseAffectedRange([]OsvEvent{ + {Introduced: "1.2.0"}, + {Fixed: "1.5.0"}, + {Introduced: "2.0.0"}, + {LastAffected: "2.3.0"}, + }) + + if len(versions) != 2 { + t.Fatalf("expected 2 intervals, got %d", len(versions)) + } + assertAffectedVersion(t, versions[0], "1.2.0", "1.5.0", "") + assertAffectedVersion(t, versions[1], "2.0.0", "", "2.3.0") +} + +func TestParseAffectedRangeKeepsOpenInterval(t *testing.T) { + versions := parseAffectedRange([]OsvEvent{{Introduced: "0"}}) + + if len(versions) != 1 { + t.Fatalf("expected 1 interval, got %d", len(versions)) + } + assertAffectedVersion(t, versions[0], "0", "", "") +} + +func TestParseAffectedRangeDefaultsMissingIntroducedToZero(t *testing.T) { + versions := parseAffectedRange([]OsvEvent{{Fixed: "1.5.0"}, {LastAffected: "2.0.0"}}) + + if len(versions) != 2 { + t.Fatalf("expected 2 intervals, got %d", len(versions)) + } + assertAffectedVersion(t, versions[0], "0", "1.5.0", "") + assertAffectedVersion(t, versions[1], "0", "", "2.0.0") +} + +func assertAffectedVersion(t *testing.T, version affectedVersion, introduced, fixed, lastAffected string) { + t.Helper() + if optionalValue(version.introduced) != introduced || optionalValue(version.fixed) != fixed || optionalValue(version.lastAffected) != lastAffected { + t.Fatalf("got introduced=%q fixed=%q last_affected=%q", optionalValue(version.introduced), optionalValue(version.fixed), optionalValue(version.lastAffected)) + } +} + +func optionalValue(value *string) string { + if value == nil { + return "" + } + return *value +} diff --git a/internal/osv/http.go b/internal/osv/http.go index beaef2e..78cbfbb 100644 --- a/internal/osv/http.go +++ b/internal/osv/http.go @@ -4,7 +4,9 @@ import ( "bytes" "encoding/json" "fmt" + "io" "net/http" + "time" ) func FetchPackageVulnerabilities(payload FetchPackageVulnerabilitiesPayload) (*FetchPackageVulnerabilitiesResponse, error) { @@ -53,3 +55,40 @@ func FetchPackageVulnerabilities(payload FetchPackageVulnerabilitiesPayload) (*F return &result, nil } + +func FetchOSVEcosystemDump(payload FetchOSVEcosystemDumpPayload) (io.ReadCloser, error) { + dumpURL := fmt.Sprintf( + "%s/%s/%s", + npmOsvDumpUrl, + payload.Ecosystem, + npmOsvDumpUrlSuffix, + ) + + request, err := http.NewRequest(http.MethodGet, dumpURL, nil) + if err != nil { + return nil, err + } + + response, err := http.DefaultClient.Do(request) + if err != nil { + return nil, err + } + + if response.StatusCode != http.StatusOK { + response.Body.Close() + + return nil, fmt.Errorf( + "OSV returned status %s", + response.Status, + ) + } + + progress := &progressReader{ + reader: response.Body, + total: response.ContentLength, + start: time.Now(), + reporter: newStatusReporter("OSV download", "MB", response.ContentLength, 1024*1024), + closer: response.Body, + } + return progress, nil +} diff --git a/internal/osv/models.go b/internal/osv/models.go index 9686fbd..59a57ee 100644 --- a/internal/osv/models.go +++ b/internal/osv/models.go @@ -78,3 +78,7 @@ type OsvQueryPackage struct { type OsvQuery struct { OsvPackage OsvQueryPackage `json:"package"` } + +type FetchOSVEcosystemDumpPayload struct { + Ecosystem string +} diff --git a/internal/osv/progress.go b/internal/osv/progress.go new file mode 100644 index 0000000..947862a --- /dev/null +++ b/internal/osv/progress.go @@ -0,0 +1,132 @@ +package osv + +import ( + "fmt" + "io" + "log" + "os" + "sync" + "sync/atomic" + "time" +) + +const progressLogInterval = 100 * time.Millisecond + +type statusReporter struct { + mu sync.Mutex + phase string + current int64 + total int64 + divisor int64 + unit string + started time.Time + stop chan struct{} + done chan struct{} + lastLine int + lastPrint time.Time + printMu sync.Mutex + finished atomic.Bool +} + +func newStatusReporter(phase, unit string, total, divisor int64) *statusReporter { + reporter := &statusReporter{ + phase: phase, unit: unit, total: total, divisor: divisor, started: time.Now(), + stop: make(chan struct{}), done: make(chan struct{}), + } + reporter.print(false) + go reporter.run() + return reporter +} + +func (reporter *statusReporter) run() { + ticker := time.NewTicker(progressLogInterval) + defer ticker.Stop() + defer close(reporter.done) + for { + select { + case <-ticker.C: + reporter.print(false) + case <-reporter.stop: + return + } + } +} + +func (reporter *statusReporter) update(current int64) { + reporter.mu.Lock() + reporter.current = current + shouldPrint := time.Since(reporter.lastPrint) >= progressLogInterval + reporter.mu.Unlock() + if shouldPrint { + reporter.print(false) + } +} + +func (reporter *statusReporter) finish(message string) { + if !reporter.finished.CompareAndSwap(false, true) { + return + } + close(reporter.stop) + <-reporter.done + reporter.print(true) + log.Printf("%s", message) +} + +func (reporter *statusReporter) print(final bool) { + reporter.mu.Lock() + if !final && time.Since(reporter.lastPrint) < progressLogInterval { + reporter.mu.Unlock() + return + } + current, total, phase, unit, started := reporter.current, reporter.total, reporter.phase, reporter.unit, reporter.started + reporter.lastPrint = time.Now() + reporter.mu.Unlock() + + percent := float64(0) + if total > 0 { + percent = float64(current) / float64(total) * 100 + } + perSecond := float64(current) / time.Since(started).Seconds() + displayedCurrent := float64(current) / float64(reporter.divisor) + displayedTotal := float64(total) / float64(reporter.divisor) + line := fmt.Sprintf("%s: %.1f%% | %.2f/%.2f %s | %.2f %s/s", phase, percent, displayedCurrent, displayedTotal, unit, perSecond/float64(reporter.divisor), unit) + if total <= 0 { + line = fmt.Sprintf("%s: %.2f %s | %.2f %s/s", phase, displayedCurrent, unit, perSecond/float64(reporter.divisor), unit) + } + + reporter.printMu.Lock() + defer reporter.printMu.Unlock() + // Clear the previous line before writing the next state. + padding := "" + if len(line) < reporter.lastLine { + padding = fmt.Sprintf("%*s", reporter.lastLine-len(line), "") + } + fmt.Fprintf(os.Stdout, "\r\033[2K%s%s", line, padding) + reporter.lastLine = len(line) + if final { + fmt.Fprintln(os.Stdout) + } +} + +type progressReader struct { + reader io.Reader + total int64 + downloaded int64 + start time.Time + reporter *statusReporter + closer io.Closer +} + +func (reader *progressReader) Read(p []byte) (int, error) { + n, err := reader.reader.Read(p) + reader.downloaded += int64(n) + if n > 0 { + reader.reporter.update(reader.downloaded) + } + return n, err +} + +func (reader *progressReader) Close() error { + reader.reporter.finish(fmt.Sprintf("OSV download completed: %.2f MB", float64(reader.downloaded)/(1024*1024))) + return reader.closer.Close() +} diff --git a/internal/osv/replicator.go b/internal/osv/replicator.go new file mode 100644 index 0000000..9c4a05f --- /dev/null +++ b/internal/osv/replicator.go @@ -0,0 +1,103 @@ +package osv + +import ( + "archive/zip" + "context" + "encoding/json" + "fmt" + "io" + "log" + "sync" + "sync/atomic" +) + +const maxOSVWorkers = 12 + +func ReplicateEcosystem(ctx context.Context, payload FetchOSVEcosystemDumpPayload) error { + dump, err := FetchOSVEcosystemDump(payload) + if err != nil { + return fmt.Errorf("fetch OSV dump: %w", err) + } + archiveData, err := io.ReadAll(dump) + dump.Close() + if err != nil { + return fmt.Errorf("read OSV dump: %w", err) + } + archive, err := zip.NewReader(byteReaderAt(archiveData), int64(len(archiveData))) + if err != nil { + return fmt.Errorf("open OSV ZIP: %w", err) + } + + workerContext, cancel := context.WithCancel(ctx) + defer cancel() + processingReporter := newStatusReporter("OSV PostgreSQL", "records", int64(len(archive.File)), 1) + jobs := make(chan ParsedVulnerability, maxOSVWorkers) + var workers sync.WaitGroup + var saved atomic.Int64 + var firstErr error + var errorOnce sync.Once + for worker := 0; worker < maxOSVWorkers; worker++ { + workers.Add(1) + go func() { + defer workers.Done() + for vulnerability := range jobs { + if err := saveVulnerability(workerContext, vulnerability); err != nil { + errorOnce.Do(func() { + firstErr = err + cancel() + }) + } else { + processingReporter.update(saved.Add(1)) + } + } + }() + } +fileLoop: + for _, file := range archive.File { + if workerContext.Err() != nil { + break + } + if file.FileInfo().IsDir() || file.Name == "" || len(file.Name) < 5 || file.Name[len(file.Name)-5:] != ".json" { + continue + } + reader, openErr := file.Open() + if openErr != nil { + errorOnce.Do(func() { firstErr = openErr }) + break + } + var raw OsvVuln + decodeErr := json.NewDecoder(reader).Decode(&raw) + reader.Close() + if decodeErr != nil { + errorOnce.Do(func() { firstErr = fmt.Errorf("decode %s: %w", file.Name, decodeErr) }) + break + } + select { + case jobs <- parseVulnerability(raw): + case <-workerContext.Done(): + break fileLoop + } + } + close(jobs) + workers.Wait() + if firstErr != nil { + processingReporter.finish("OSV PostgreSQL stage failed") + return firstErr + } + processingReporter.finish("OSV PostgreSQL stage completed") + log.Printf("OSV replication completed: ecosystem=%s records=%d", payload.Ecosystem, saved.Load()) + return ctx.Err() +} + +type byteReaderAt []byte + +func (data byteReaderAt) ReadAt(target []byte, offset int64) (int, error) { + if offset >= int64(len(data)) { + return 0, io.EOF + } + read := copy(target, data[offset:]) + if read < len(target) { + return read, io.EOF + } + return read, nil +} diff --git a/internal/osv/repository.go b/internal/osv/repository.go new file mode 100644 index 0000000..cd2957a --- /dev/null +++ b/internal/osv/repository.go @@ -0,0 +1,70 @@ +package osv + +import ( + "context" + "embed" + + "github.com/trustpkg/trustpkg-api/db" +) + +//go:embed sql/*.sql +var sqlFiles embed.FS + +const ( + queryUpsertVulnerability = "sql/upsert_vulnerability.sql" + queryUpsertPackage = "sql/upsert_package.sql" + queryDeleteAffected = "sql/delete_vulnerability_children.sql" + queryDeleteReferences = "sql/delete_vulnerability_references.sql" + queryInsertAffected = "sql/insert_affected.sql" + queryInsertReference = "sql/insert_reference.sql" +) + +func saveVulnerability(ctx context.Context, vulnerability ParsedVulnerability) error { + transaction, err := db.Pool.Begin(ctx) + if err != nil { + return err + } + defer transaction.Rollback(ctx) + + var vulnerabilityID int64 + err = transaction.QueryRow(ctx, readSQL(queryUpsertVulnerability), + vulnerability.OSVID, vulnerability.CVEID, vulnerability.Summary, + vulnerability.Description, vulnerability.Severity, vulnerability.CVSSScore, + vulnerability.CVSSVector, vulnerability.PublishedAt, vulnerability.ModifiedAt, + ).Scan(&vulnerabilityID) + if err != nil { + return err + } + if _, err = transaction.Exec(ctx, readSQL(queryDeleteAffected), vulnerabilityID); err != nil { + return err + } + if _, err = transaction.Exec(ctx, readSQL(queryDeleteReferences), vulnerabilityID); err != nil { + return err + } + for _, affected := range vulnerability.Affected { + var packageID int64 + err = transaction.QueryRow(ctx, readSQL(queryUpsertPackage), affected.PackageName, affected.Ecosystem).Scan(&packageID) + if err != nil { + return err + } + if _, err = transaction.Exec(ctx, readSQL(queryInsertAffected), vulnerabilityID, packageID, + affected.IntroducedVersion, affected.FixedVersion, affected.LastAffectedVersion); err != nil { + return err + } + } + for _, reference := range vulnerability.References { + if _, err = transaction.Exec(ctx, readSQL(queryInsertReference), vulnerabilityID, reference); err != nil { + return err + } + } + + return transaction.Commit(ctx) +} + +func readSQL(name string) string { + contents, err := sqlFiles.ReadFile(name) + if err != nil { + panic(err) + } + return string(contents) +} diff --git a/internal/osv/sql/delete_vulnerability_children.sql b/internal/osv/sql/delete_vulnerability_children.sql new file mode 100644 index 0000000..c25666b --- /dev/null +++ b/internal/osv/sql/delete_vulnerability_children.sql @@ -0,0 +1,2 @@ +DELETE FROM vulnerability_affected +WHERE vulnerability_id = $1; diff --git a/internal/osv/sql/delete_vulnerability_references.sql b/internal/osv/sql/delete_vulnerability_references.sql new file mode 100644 index 0000000..7d75fd5 --- /dev/null +++ b/internal/osv/sql/delete_vulnerability_references.sql @@ -0,0 +1,2 @@ +DELETE FROM vulnerability_references +WHERE vulnerability_id = $1; diff --git a/internal/osv/sql/insert_affected.sql b/internal/osv/sql/insert_affected.sql new file mode 100644 index 0000000..78df410 --- /dev/null +++ b/internal/osv/sql/insert_affected.sql @@ -0,0 +1,14 @@ +INSERT INTO vulnerability_affected ( + vulnerability_id, package_id, introduced_version, + fixed_version, last_affected_version +) +SELECT $1, $2, $3, $4, $5 +WHERE NOT EXISTS ( + SELECT 1 + FROM vulnerability_affected + WHERE vulnerability_id = $1 + AND package_id = $2 + AND introduced_version IS NOT DISTINCT FROM $3 + AND fixed_version IS NOT DISTINCT FROM $4 + AND last_affected_version IS NOT DISTINCT FROM $5 +); diff --git a/internal/osv/sql/insert_reference.sql b/internal/osv/sql/insert_reference.sql new file mode 100644 index 0000000..fef1342 --- /dev/null +++ b/internal/osv/sql/insert_reference.sql @@ -0,0 +1,3 @@ +INSERT INTO vulnerability_references (vulnerability_id, url) +VALUES ($1, $2) +ON CONFLICT (vulnerability_id, url) DO NOTHING; diff --git a/internal/osv/sql/upsert_package.sql b/internal/osv/sql/upsert_package.sql new file mode 100644 index 0000000..879ee05 --- /dev/null +++ b/internal/osv/sql/upsert_package.sql @@ -0,0 +1,4 @@ +INSERT INTO packages (name, ecosystem) +VALUES ($1, $2) +ON CONFLICT (ecosystem, name) DO UPDATE SET name = EXCLUDED.name +RETURNING id; diff --git a/internal/osv/sql/upsert_vulnerability.sql b/internal/osv/sql/upsert_vulnerability.sql new file mode 100644 index 0000000..29da7f4 --- /dev/null +++ b/internal/osv/sql/upsert_vulnerability.sql @@ -0,0 +1,15 @@ +INSERT INTO vulnerabilities ( + osv_id, cve_id, summary, description, severity, cvss_score, + cvss_vector, published_at, modified_at +) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) +ON CONFLICT (osv_id) DO UPDATE SET + cve_id = EXCLUDED.cve_id, + summary = EXCLUDED.summary, + description = EXCLUDED.description, + severity = EXCLUDED.severity, + cvss_score = EXCLUDED.cvss_score, + cvss_vector = EXCLUDED.cvss_vector, + published_at = EXCLUDED.published_at, + modified_at = EXCLUDED.modified_at +RETURNING id;