From d13b8be0ae9a583c7a037fc866622d3f1f69a3f0 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 20 Oct 2026 19:39:14 +0200 Subject: [PATCH 1/5] Host generic files and share one push path across hosted ecosystems Hosted ecosystems are now entries in one table: each defines its package names, how the stored filename is chosen, how the manifest is read and which namespaces are served from pushes alone. npm and Composer move there unchanged, and generic is added as a third entry. Generic files are pushed to /api/hosted/generic/{name}/{version}/{file} and served without a token from /generic/hosted/{name}/{version}/{file}, with the sha256 as ETag. They have no manifest, a version can hold several files, and they never come from an upstream. "hosted" is reserved as an upstream.generic name. DELETE with a file removes only that file. For all hosted ecosystems: - A second push of the same file is refused with 409. Clients pin pushed files by version and often by hash, so a published file is no longer silently replaced. Delete it first to push it again. - PUT takes an optional ?sha256= and refuses bytes that don't match. proxy hosted push always sends it. Versions of generic files carry no version-level integrity, since it could only describe one of their files; each file is checked against its own sha256. Refs #133, #134, #535 --- cmd/proxy/hosted.go | 47 +++++-- cmd/proxy/hosted_test.go | 36 ++++- docs/configuration.md | 33 ++++- docs/swagger/docs.go | 18 ++- docs/swagger/swagger.json | 18 ++- internal/config/config.go | 7 + internal/config/config_test.go | 7 + internal/database/local_artifacts.go | 27 ++-- internal/database/local_artifacts_test.go | 4 +- internal/handler/composer_hosted_test.go | 4 +- internal/handler/generic.go | 41 ++++++ internal/handler/generic_test.go | 1 + internal/handler/hosted.go | 164 +++++++++++++++++----- internal/handler/hosted_manifest.go | 13 +- internal/handler/hosted_test.go | 33 ++++- internal/handler/npm_hosted_test.go | 4 +- internal/server/hosted_api.go | 95 ++++++++++--- internal/server/hosted_api_test.go | 122 ++++++++++++++++ 18 files changed, 569 insertions(+), 105 deletions(-) diff --git a/cmd/proxy/hosted.go b/cmd/proxy/hosted.go index bbc947df..ec61b3ab 100644 --- a/cmd/proxy/hosted.go +++ b/cmd/proxy/hosted.go @@ -1,6 +1,8 @@ package main import ( + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "flag" @@ -9,9 +11,12 @@ import ( "net/http" "net/url" "os" + "path/filepath" "strings" "text/tabwriter" "time" + + "github.com/git-pkgs/proxy/internal/handler" ) const ( @@ -21,7 +26,7 @@ const ( pushArgCount = 4 // ecosystem, name, version, file listArgCount = 2 // ecosystem, name - deleteArgCount = 3 // ecosystem, name, version + deleteArgCount = 3 // ecosystem, name, version[, file] tablePadding = 2 ) @@ -69,15 +74,17 @@ func newHostedCommand(name, usage, summary string, getenv func(string) string) h return cmd } -// parse parses flags and checks for exactly want positional arguments. -func (c hostedCommand) parse(args []string, want int) ([]string, error) { +// parse parses flags and checks for exactly want positional arguments, or +// one more when optional is set. +func (c hostedCommand) parse(args []string, want int, optional ...bool) ([]string, error) { if err := c.flags.Parse(args); err != nil { if errors.Is(err, flag.ErrHelp) { c.flags.Usage() } return nil, fmt.Errorf("%w\nusage: %s", err, c.usage) } - if c.flags.NArg() != want { + n := c.flags.NArg() + if n != want && (len(optional) == 0 || !optional[0] || n != want+1) { return nil, fmt.Errorf("usage: %s", c.usage) } return c.flags.Args(), nil @@ -121,7 +128,22 @@ func runPush(args []string, getenv func(string) string, stdout io.Writer, client } defer func() { _ = file.Close() }() - req, err := http.NewRequest(http.MethodPut, cmd.endpoint(ecosystem, name, version), file) + // The proxy checks the upload against this, so a truncated or corrupted + // transfer is refused rather than stored. + digest := sha256.New() + if _, err := io.Copy(digest, file); err != nil { + return err + } + if _, err := file.Seek(0, io.SeekStart); err != nil { + return err + } + + segments := []string{version} + if handler.HostedFileInPath(ecosystem) { + segments = append(segments, filepath.Base(path)) + } + endpoint := cmd.endpoint(ecosystem, name, segments...) + "?sha256=" + hex.EncodeToString(digest.Sum(nil)) + req, err := http.NewRequest(http.MethodPut, endpoint, file) if err != nil { return err } @@ -161,8 +183,9 @@ func runList(args []string, getenv func(string) string, stdout io.Writer, client } func runDelete(args []string, getenv func(string) string, stdout io.Writer, client *http.Client) error { - cmd := newHostedCommand("delete", "proxy hosted delete [-url URL] ", "Delete a pushed version", getenv) - positional, err := cmd.parse(args, deleteArgCount) + cmd := newHostedCommand("delete", "proxy hosted delete [-url URL] [file]", + "Delete a pushed version, or one generic file of it", getenv) + positional, err := cmd.parse(args, deleteArgCount, true) if err != nil { return err } @@ -170,7 +193,7 @@ func runDelete(args []string, getenv func(string) string, stdout io.Writer, clie if err != nil { return err } - req, err := http.NewRequest(http.MethodDelete, cmd.endpoint(positional[0], positional[1], positional[2]), nil) + req, err := http.NewRequest(http.MethodDelete, cmd.endpoint(positional[0], positional[1], positional[2:]...), nil) if err != nil { return err } @@ -179,7 +202,11 @@ func runDelete(args []string, getenv func(string) string, stdout io.Writer, clie if err := doHosted(client, req, http.StatusNoContent, nil); err != nil { return err } - _, err = fmt.Fprintf(stdout, "deleted %s %s@%s\n", positional[0], positional[1], positional[2]) + deleted := positional[1] + "@" + positional[2] + if len(positional) > deleteArgCount { + deleted += " " + positional[deleteArgCount] + } + _, err = fmt.Fprintf(stdout, "deleted %s %s\n", positional[0], deleted) return err } @@ -210,7 +237,7 @@ func doHosted(client *http.Client, req *http.Request, wantStatus int, out any) e const hostedUsage = "usage: proxy hosted [flags] ...\n\n" + " push Push a package archive to the proxy\n" + " list List pushed package archives\n" + - " delete Delete a pushed version" + " delete Delete a pushed version, or one generic file of it" type hostedSubcommand func([]string, func(string) string, io.Writer, *http.Client) error diff --git a/cmd/proxy/hosted_test.go b/cmd/proxy/hosted_test.go index fef71608..34f9a7b3 100644 --- a/cmd/proxy/hosted_test.go +++ b/cmd/proxy/hosted_test.go @@ -62,7 +62,9 @@ func TestRunPush(t *testing.T) { if recorded.method != http.MethodPut || recorded.path != "/api/hosted/npm/@types/node/20.1.0" { t.Errorf("request = %s %s", recorded.method, recorded.path) } - if recorded.authorization != "Bearer tok" || recorded.body != "tarball" || recorded.query != "" { + // sha256 of "tarball", so the proxy can check the upload. + const tarballSHA256 = "db4b4d0d1cb480bf9aeea253771c00febe627f236765fa37d6a5614f079a3aa0" + if recorded.authorization != "Bearer tok" || recorded.body != "tarball" || recorded.query != "sha256="+tarballSHA256 { t.Errorf("authorization/body/query = %q/%q/%q", recorded.authorization, recorded.body, recorded.query) } if out := stdout.String(); !strings.Contains(out, "node-20.1.0.tgz") || !strings.Contains(out, "abc123") { @@ -70,6 +72,21 @@ func TestRunPush(t *testing.T) { } } +func TestRunPushGenericNamesTheFile(t *testing.T) { + srv, recorded := hostedTestServer(t, http.StatusCreated, + `{"ecosystem":"generic","name":"tools","version":"1.2.0","filename":"tool-linux-arm64.tar.gz","sha256":"abc123","size":7}`) + archive := writeArchive(t, "tool-linux-arm64.tar.gz", "tarball") + + err := runPush([]string{"-url", srv.URL, "generic", "tools", "1.2.0", archive}, + hostedEnv(map[string]string{"PROXY_HOSTED_TOKEN": "tok"}), io.Discard, srv.Client()) + if err != nil { + t.Fatalf("runPush: %v", err) + } + if recorded.path != "/api/hosted/generic/tools/1.2.0/tool-linux-arm64.tar.gz" { + t.Errorf("path = %q, want the file name appended", recorded.path) + } +} + func TestRunPushErrors(t *testing.T) { srv, _ := hostedTestServer(t, http.StatusBadRequest, `{"code":"BAD_REQUEST","message":"pushing packages is not supported for this ecosystem: cargo"}`) archive := writeArchive(t, "serde-1.0.200.crate", "crate") @@ -141,6 +158,23 @@ func TestRunDelete(t *testing.T) { } } +func TestRunDeleteOneFile(t *testing.T) { + srv, recorded := hostedTestServer(t, http.StatusNoContent, "") + var stdout bytes.Buffer + + err := runDelete([]string{"-url", srv.URL, "generic", "tools", "1.2.0", "tool-linux-arm64.tar.gz"}, + hostedEnv(map[string]string{"PROXY_HOSTED_TOKEN": "tok"}), &stdout, srv.Client()) + if err != nil { + t.Fatalf("runDelete: %v", err) + } + if recorded.path != "/api/hosted/generic/tools/1.2.0/tool-linux-arm64.tar.gz" { + t.Errorf("path = %q", recorded.path) + } + if out := stdout.String(); out != "deleted generic tools@1.2.0 tool-linux-arm64.tar.gz\n" { + t.Errorf("output = %q", out) + } +} + func TestRunHostedDispatchesSubcommands(t *testing.T) { srv, recorded := hostedTestServer(t, http.StatusOK, `{"ecosystem":"npm","name":"request","artifacts":[]}`) diff --git a/docs/configuration.md b/docs/configuration.md index 507cad59..a08ec4c4 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1076,7 +1076,7 @@ Set to `"0"` to disable the timeout entirely (requests then rely only on the ser ## Hosted Packages -Operators can push a package archive into the proxy under a name and version. The proxy serves it in place of the upstream archive with the same version, or in addition to the upstream's versions. Use it to ship a patched build of a dependency without running a separate registry. +Operators can push a package archive into the proxy under a name and version. The proxy serves it in place of the upstream archive with the same version, or in addition to the upstream's versions. Use it to ship a patched build of a dependency without running a separate registry. Files that aren't packages, such as prebuilt binaries other builds download, can be pushed as [generic files](#generic-files). The `/api/hosted` endpoints are disabled by default. Because a pushed version replaces the upstream one for every client, enabling them requires a token: @@ -1093,7 +1093,7 @@ hosted: |--------|-------------|-------------| | `hosted.enabled` | `PROXY_HOSTED_ENABLED` | Register the `/api/hosted` endpoints (default `false`) | | `hosted.token` | `PROXY_HOSTED_TOKEN` | Bearer token required to push and delete; required when enabled, supports `${VAR}` | -| `hosted.max_upload_size` | `PROXY_HOSTED_MAX_UPLOAD_SIZE` | Maximum size of one pushed archive (default `100MB`) | +| `hosted.max_upload_size` | `PROXY_HOSTED_MAX_UPLOAD_SIZE` | Maximum size of one pushed archive or file (default `100MB`) | | `hosted.npm_scopes` | `PROXY_HOSTED_NPM_SCOPES` | npm scopes served from pushed packages only (comma-separated in the environment) | | `hosted.composer_vendors` | `PROXY_HOSTED_COMPOSER_VENDORS` | Composer vendors served from pushed packages only | @@ -1114,13 +1114,17 @@ hosted: ```bash curl -X PUT --data-binary @request-2.88.3.tgz \ -H "Authorization: Bearer $PROXY_HOSTED_TOKEN" \ - https://proxy.example.com/api/hosted/npm/request/2.88.3 + "https://proxy.example.com/api/hosted/npm/request/2.88.3?sha256=$(sha256sum request-2.88.3.tgz | cut -d' ' -f1)" ``` + The `sha256` query parameter is optional. When it is set, a push whose bytes don't match is refused with 400. `proxy hosted push` always sends it. + 3. Install it as usual, e.g. `npm install request@2.88.3`. The version is listed in the package metadata, so ranges like `^2.88.0` resolve to it too. 4. Check what is pushed: `proxy hosted list npm request` or `GET /api/hosted/npm/request`. 5. Remove it again: `proxy hosted delete npm request 2.88.3` or `DELETE /api/hosted/npm/request/2.88.3`. If the upstream has the version, clients get the upstream archive again. +A pushed file is never replaced: pushing the same version again answers 409. Clients that pinned the file by its hash would otherwise get different bytes under the same version. To push it again, delete it first. + ### Supported ecosystems Download handlers look archives up by the filename clients request, so a push is stored under that name: @@ -1129,10 +1133,11 @@ Download handlers look archives up by the filename clients request, so a push is |-----------|------|----------| | `npm` | `{name}` or `@{scope}/{name}` | `{name}-{version}.tgz`, without the scope | | `composer` | `{vendor}/{package}` | `{package}-{version}.zip` | +| `generic` | `{name}`, one path segment | named in the push path, see [Generic files](#generic-files) | Other ecosystems are not supported yet. -The archive must contain the package manifest: `package.json` in the top directory of an npm tarball, or `composer.json` at the root or in the single top directory of a Composer zip. Its name, and its version when it has one, must match the push. Otherwise the push is refused with 400. +npm and Composer archives must contain the package manifest: `package.json` in the top directory of an npm tarball, or `composer.json` at the root or in the single top directory of a Composer zip. Its name, and its version when it has one, must match the push. Otherwise the push is refused with 400. ### Package metadata @@ -1157,6 +1162,26 @@ For private packages: Other packages, including ones in other scopes and vendors, are proxied as usual. +### Generic files + +Files that aren't packages, such as a prebuilt binary that other builds download and check against a pinned sha256, are pushed to the `generic` ecosystem. They have no manifest; the name, version and filename come from the path, and a version can hold several files: + +```bash +proxy hosted push generic tools 1.2.0 ./tool-linux-arm64.tar.gz +# or +curl -X PUT --data-binary @tool-linux-arm64.tar.gz \ + -H "Authorization: Bearer $PROXY_HOSTED_TOKEN" \ + https://proxy.example.com/api/hosted/generic/tools/1.2.0/tool-linux-arm64.tar.gz +``` + +They are downloaded without a token from `/generic/hosted/{name}/{version}/{file}`, with the file's sha256 as `ETag`: + +```bash +curl -O https://proxy.example.com/generic/hosted/tools/1.2.0/tool-linux-arm64.tar.gz +``` + +Generic files never come from an upstream: anything not pushed answers 404, and `hosted` can't be used as an `upstream.generic` name. `proxy hosted list generic tools` lists all versions and files. `proxy hosted delete generic tools 1.2.0` removes a version with all its files, and `proxy hosted delete generic tools 1.2.0 tool-linux-arm64.tar.gz` removes only that file. + ### Storage and eviction Pushed archives are stored like cached ones but marked as local. The cache size limit and storage retention never remove them; only a delete does. Scanning, when enabled, checks pushed archives like fetched ones. diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go index 7132ffb0..e35a86f3 100644 --- a/docs/swagger/docs.go +++ b/docs/swagger/docs.go @@ -230,7 +230,7 @@ const docTemplate = `{ "BearerAuth": [] } ], - "description": "Stores the request body as the archive for name@version. It is served in place of the upstream archive with the same version and is never evicted.", + "description": "Stores the request body as the archive for name@version. It is served in place of the upstream archive with the same version and is never evicted. Generic files are pushed to /api/hosted/generic/{name}/{version}/{file} and served at /generic/hosted/{name}/{version}/{file}. A file that was already pushed is refused with 409; delete it first to push it again.", "consumes": [ "application/octet-stream" ], @@ -244,7 +244,7 @@ const docTemplate = `{ "parameters": [ { "type": "string", - "description": "Ecosystem (npm or composer)", + "description": "Ecosystem (npm, composer or generic)", "name": "ecosystem", "in": "path", "required": true @@ -262,6 +262,12 @@ const docTemplate = `{ "name": "version", "in": "path", "required": true + }, + { + "type": "string", + "description": "Expected sha256 of the body, hex-encoded; a mismatch is refused", + "name": "sha256", + "in": "query" } ], "responses": { @@ -289,6 +295,12 @@ const docTemplate = `{ "$ref": "#/definitions/server.ErrorResponse" } }, + "409": { + "description": "Conflict", + "schema": { + "$ref": "#/definitions/server.ErrorResponse" + } + }, "413": { "description": "Request Entity Too Large", "schema": { @@ -303,7 +315,7 @@ const docTemplate = `{ "BearerAuth": [] } ], - "description": "Removes the archives pushed for name@version. If the upstream has the version, clients get the upstream archive again.", + "description": "Removes the archives pushed for name@version. If the upstream has the version, clients get the upstream archive again. For generic files, /api/hosted/generic/{name}/{version}/{file} removes only that file.", "tags": [ "hosted" ], diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json index fcb4123e..09e4f11e 100644 --- a/docs/swagger/swagger.json +++ b/docs/swagger/swagger.json @@ -223,7 +223,7 @@ "BearerAuth": [] } ], - "description": "Stores the request body as the archive for name@version. It is served in place of the upstream archive with the same version and is never evicted.", + "description": "Stores the request body as the archive for name@version. It is served in place of the upstream archive with the same version and is never evicted. Generic files are pushed to /api/hosted/generic/{name}/{version}/{file} and served at /generic/hosted/{name}/{version}/{file}. A file that was already pushed is refused with 409; delete it first to push it again.", "consumes": [ "application/octet-stream" ], @@ -237,7 +237,7 @@ "parameters": [ { "type": "string", - "description": "Ecosystem (npm or composer)", + "description": "Ecosystem (npm, composer or generic)", "name": "ecosystem", "in": "path", "required": true @@ -255,6 +255,12 @@ "name": "version", "in": "path", "required": true + }, + { + "type": "string", + "description": "Expected sha256 of the body, hex-encoded; a mismatch is refused", + "name": "sha256", + "in": "query" } ], "responses": { @@ -282,6 +288,12 @@ "$ref": "#/definitions/server.ErrorResponse" } }, + "409": { + "description": "Conflict", + "schema": { + "$ref": "#/definitions/server.ErrorResponse" + } + }, "413": { "description": "Request Entity Too Large", "schema": { @@ -296,7 +308,7 @@ "BearerAuth": [] } ], - "description": "Removes the archives pushed for name@version. If the upstream has the version, clients get the upstream archive again.", + "description": "Removes the archives pushed for name@version. If the upstream has the version, clients get the upstream archive again. For generic files, /api/hosted/generic/{name}/{version}/{file} removes only that file.", "tags": [ "hosted" ], diff --git a/internal/config/config.go b/internal/config/config.go index e7851c91..e788f8d9 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -840,6 +840,9 @@ func (u *UpstreamConfig) Validate() error { if err := validateOCIMirrors(u.OCIMirrors, u.OCI); err != nil { return err } + if _, reserved := u.Generic[HostedGenericRepository]; reserved { + return fmt.Errorf("invalid upstream.generic name %q: reserved for pushed files at /generic/%s/", HostedGenericRepository, HostedGenericRepository) + } if err := validateNamedUpstreams("upstream.generic", u.Generic); err != nil { return err } @@ -894,6 +897,10 @@ func (c *Config) validateUpstream() error { return nil } +// HostedGenericRepository is the /generic/{name}/ path that serves files +// pushed through /api/hosted/generic. No upstream.generic entry may use it. +const HostedGenericRepository = "hosted" + // CompileGenericArtifactPattern compiles an upstream.generic_artifacts pattern // so that it only matches a whole path, and checks it has the `version` and // `file` groups. diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 48ae1c79..b21c3480 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1455,6 +1455,13 @@ func TestValidateNamedUpstreams(t *testing.T) { }, wantErr: true, }, + { + name: "generic upstream name reserved for pushed files", + modify: func(cfg *Config) { + cfg.Upstream.Generic = map[string]string{"hosted": "https://example.com"} + }, + wantErr: true, + }, { name: "generic artifact patterns", modify: func(cfg *Config) { diff --git a/internal/database/local_artifacts.go b/internal/database/local_artifacts.go index 83197768..1bb391fd 100644 --- a/internal/database/local_artifacts.go +++ b/internal/database/local_artifacts.go @@ -31,27 +31,32 @@ func (db *DB) GetLocalArtifacts(packagePURL string) ([]LocalArtifact, error) { return artifacts, nil } -// DeleteLocalArtifacts removes the pushed artifacts of a version and queues -// their stored objects for deletion in the same transaction, so a failed -// queue keeps the records instead of orphaning the objects. Artifacts cached -// from an upstream are left alone. It returns how many records it removed. -func (db *DB) DeleteLocalArtifacts(versionPURL string) (int, error) { +// DeleteLocalArtifacts removes the pushed artifacts of a version, or only the +// one named filename when it is set, and queues their stored objects for +// deletion in the same transaction, so a failed queue keeps the records +// instead of orphaning the objects. Artifacts cached from an upstream are left +// alone. It returns how many records it removed. +func (db *DB) DeleteLocalArtifacts(versionPURL, filename string) (int, error) { tx, err := db.Beginx() if err != nil { return 0, err } defer func() { _ = tx.Rollback() }() + where := "version_purl = ? AND source = 'local'" + args := []any{versionPURL} + if filename != "" { + where += " AND filename = ?" + args = append(args, filename) + } + var paths []string - selectQuery := db.Rebind(` - SELECT storage_path FROM artifacts - WHERE version_purl = ? AND source = 'local' AND storage_path IS NOT NULL - `) - if err := tx.Select(&paths, selectQuery, versionPURL); err != nil { + selectQuery := db.Rebind(`SELECT storage_path FROM artifacts WHERE ` + where + ` AND storage_path IS NOT NULL`) + if err := tx.Select(&paths, selectQuery, args...); err != nil { return 0, fmt.Errorf("reading local artifacts: %w", err) } - res, err := tx.Exec(db.Rebind(`DELETE FROM artifacts WHERE version_purl = ? AND source = 'local'`), versionPURL) + res, err := tx.Exec(db.Rebind(`DELETE FROM artifacts WHERE `+where), args...) if err != nil { return 0, fmt.Errorf("deleting local artifacts: %w", err) } diff --git a/internal/database/local_artifacts_test.go b/internal/database/local_artifacts_test.go index 81d5ec8b..02335da3 100644 --- a/internal/database/local_artifacts_test.go +++ b/internal/database/local_artifacts_test.go @@ -183,7 +183,7 @@ func TestDeleteLocalArtifacts(t *testing.T) { seedSourcedArtifact(t, db, "express", SourceUpstream, 100, time.Now()) const pushedPath = "npm/request/1.0.0/request-1.0.0.tgz" - deleted, err := db.DeleteLocalArtifacts("pkg:npm/request@1.0.0") + deleted, err := db.DeleteLocalArtifacts("pkg:npm/request@1.0.0", "") if err != nil || deleted != 1 { t.Fatalf("DeleteLocalArtifacts(request) = %d, %v, want 1, nil", deleted, err) } @@ -196,7 +196,7 @@ func TestDeleteLocalArtifacts(t *testing.T) { t.Errorf("queued deletes for %s = %d, %v, want 1", pushedPath, queued, err) } - deleted, err = db.DeleteLocalArtifacts("pkg:npm/express@1.0.0") + deleted, err = db.DeleteLocalArtifacts("pkg:npm/express@1.0.0", "") if err != nil || deleted != 0 { t.Errorf("DeleteLocalArtifacts(upstream) = %d, %v, want 0, nil", deleted, err) } diff --git a/internal/handler/composer_hosted_test.go b/internal/handler/composer_hosted_test.go index 048c291c..539e999f 100644 --- a/internal/handler/composer_hosted_test.go +++ b/internal/handler/composer_hosted_test.go @@ -58,11 +58,11 @@ func newComposerHostedFixture(t *testing.T, upstream map[string]string) composer func (f composerHostedFixture) push(t *testing.T, name, version, manifest string) HostedArtifact { t.Helper() - target, err := HostedTargetFor("composer", name, version) + target, err := HostedTargetFor("composer", name, version, "") if err != nil { t.Fatalf("HostedTargetFor: %v", err) } - pushed, err := f.proxy.PushLocalArtifact(context.Background(), "composer", target, bytes.NewReader(composerZip("", manifest)), "") + pushed, err := f.proxy.PushLocalArtifact(context.Background(), "composer", target, bytes.NewReader(composerZip("", manifest)), "", "") if err != nil { t.Fatalf("PushLocalArtifact: %v", err) } diff --git a/internal/handler/generic.go b/internal/handler/generic.go index 33a6ed20..f4114307 100644 --- a/internal/handler/generic.go +++ b/internal/handler/generic.go @@ -126,6 +126,10 @@ func (h *GenericHandler) Routes() http.Handler { } repository, rest, ok := strings.Cut(path, "/") + if repository == HostedGenericRepository { + h.handleHosted(w, r, rest) + return + } upstreamURL, found := h.repositories[repository] if !ok || rest == "" || !found { http.NotFound(w, r) @@ -141,6 +145,38 @@ func (h *GenericHandler) Routes() http.Handler { }) } +// HostedGenericRepository is the /generic/{name}/ path that serves pushed +// generic files instead of an upstream. upstream.generic can't use the name. +const HostedGenericRepository = config.HostedGenericRepository + +// handleHosted serves a pushed file at {name}/{version}/{file}. Pushed files +// never come from an upstream, so anything not pushed is a 404. +func (h *GenericHandler) handleHosted(w http.ResponseWriter, r *http.Request, path string) { + segments := strings.Split(path, "/") + if !h.proxy.HostedEnabled || len(segments) != hostedGenericPathSegments { + http.NotFound(w, r) + return + } + target, err := HostedTargetFor(genericEcosystem, segments[0], segments[1], segments[2]) + if err != nil { + http.NotFound(w, r) + return + } + result, err := h.proxy.GetLocalArtifact(r.Context(), genericEcosystem, target.Name, target.Version, target.Filename) + if err != nil { + h.proxy.serveArtifactError(w, err, "failed to read pushed file") + return + } + if result == nil { + http.NotFound(w, r) + return + } + serveGenericArtifact(w, r, result) +} + +// hostedGenericPathSegments is {name}/{version}/{file}. +const hostedGenericPathSegments = 3 + // genericArtifact is the identity of a version-pinned download. type genericArtifact struct { name string @@ -234,7 +270,12 @@ func (h *GenericHandler) handleArtifact(w http.ResponseWriter, r *http.Request, h.proxy.serveArtifactError(w, err, "failed to fetch artifact") return } + serveGenericArtifact(w, r, result) +} +// serveGenericArtifact serves a cached or pushed file, as a plain download +// when its media type is unknown. +func serveGenericArtifact(w http.ResponseWriter, r *http.Request, result *CacheResult) { if result.Artifact.MediaType == "" { result.Artifact.MediaType = "application/octet-stream" } diff --git a/internal/handler/generic_test.go b/internal/handler/generic_test.go index 9fac379b..6948b862 100644 --- a/internal/handler/generic_test.go +++ b/internal/handler/generic_test.go @@ -312,6 +312,7 @@ func TestGenericHandler_RejectsUnknownUpstreamAndBadPaths(t *testing.T) { {"traversal", http.MethodGet, "/github/../etc/passwd", http.StatusBadRequest}, {"encoded traversal", http.MethodGet, "/github/%2e%2e/etc/passwd", http.StatusBadRequest}, {"post", http.MethodPost, "/github/owner/repo/releases/download/v1/asset", http.StatusMethodNotAllowed}, + {"hosted files with hosting off", http.MethodGet, "/hosted/tools/1.2.0/tool.tar.gz", http.StatusNotFound}, } for _, tt := range tests { diff --git a/internal/handler/hosted.go b/internal/handler/hosted.go index 41ccb930..3e15868f 100644 --- a/internal/handler/hosted.go +++ b/internal/handler/hosted.go @@ -23,6 +23,16 @@ var ( ErrHostedUnsupported = errors.New("pushing packages is not supported for this ecosystem") ErrHostedInvalidName = errors.New("invalid package name or version") ErrHostedVersionNotFound = errors.New("no pushed artifacts for this version") + ErrHostedFileNotFound = errors.New("no pushed file of this name for this version") + + // ErrHostedExists refuses a second push of the same file. Clients pin + // pushed files by version and often by hash, so a published file is never + // silently replaced; delete it first to push it again. + ErrHostedExists = errors.New("this file was already pushed; delete it first to push it again") + + // ErrHostedChecksumMismatch reports a push whose bytes don't match the + // sha256 the client sent along. + ErrHostedChecksumMismatch = errors.New("pushed bytes don't match the expected sha256") // ErrHostedArtifactUnavailable and ErrHostedArtifactMissing report a // pushed artifact the cache can't serve. They stop the caller from @@ -50,53 +60,105 @@ type HostedArtifact struct { PushedAt time.Time `json:"pushed_at"` } -// hostedNaming describes how an ecosystem names pushed archives: which -// package names it has, split on "/", and the filename its clients request. -type hostedNaming struct { +// hostedEcosystem is what an ecosystem defines for pushed packages. Storing, +// listing, deleting and the push rules are shared; adding an ecosystem means +// adding an entry here and, if its clients read version lists, listing pushed +// versions in its metadata. +type hostedEcosystem struct { + // validName reports whether a package name, split on "/", is one of the + // ecosystem's. validName func(segments []string) bool - filename func(name, version string) string + + // filename is the name clients request the archive by, derived from the + // package name and version. When nil, the filename is the last segment of + // the push path, and the package name must be a single segment so the + // path is unambiguous. + filename func(name, version string) string + + // readManifest reads the package manifest out of a pushed archive and + // checks that it describes target. When nil, pushes carry no manifest. + readManifest func(archive io.Reader, target HostedTarget) (json.RawMessage, error) + + // hostedOnly reports whether a namespace (an npm scope, a Composer + // vendor) is served from pushed packages alone. When nil, the ecosystem + // has no such namespaces. + hostedOnly func(p *Proxy, namespace string) bool } -var hostedNamings = map[string]hostedNaming{ +var hostedEcosystems = map[string]hostedEcosystem{ "npm": { validName: func(segments []string) bool { return len(segments) == 1 || (len(segments) == 2 && strings.HasPrefix(segments[0], "@")) }, - filename: func(name, version string) string { return lastSegment(name) + "-" + version + ".tgz" }, + filename: func(name, version string) string { return lastSegment(name) + "-" + version + ".tgz" }, + readManifest: readNPMManifest, + hostedOnly: func(p *Proxy, scope string) bool { + return slices.Contains(p.HostedNPMScopes, scope) + }, }, "composer": { - validName: func(segments []string) bool { return len(segments) == vendorPackageParts }, - filename: func(name, version string) string { return lastSegment(name) + "-" + version + ".zip" }, + validName: func(segments []string) bool { return len(segments) == vendorPackageParts }, + filename: func(name, version string) string { return lastSegment(name) + "-" + version + ".zip" }, + readManifest: readComposerManifest, + hostedOnly: func(p *Proxy, vendor string) bool { + return slices.ContainsFunc(p.HostedComposerVendors, func(v string) bool { + return strings.EqualFold(v, vendor) + }) + }, + }, + // generic holds files that aren't packages, such as prebuilt binaries. + // They are served at /generic/hosted/{name}/{version}/{file} and never + // come from an upstream. + genericEcosystem: { + validName: func(segments []string) bool { return len(segments) == 1 }, }, } // HostedSupported reports whether packages can be pushed for ecosystem. func HostedSupported(ecosystem string) bool { - _, ok := hostedNamings[ecosystem] + _, ok := hostedEcosystems[ecosystem] return ok } +// HostedFileInPath reports whether pushes for ecosystem name the file in the +// path, {name}/{version}/{file}, rather than deriving it from the name and +// version. +func HostedFileInPath(ecosystem string) bool { + eco, ok := hostedEcosystems[ecosystem] + return ok && eco.filename == nil +} + func lastSegment(name string) string { return name[strings.LastIndex(name, "/")+1:] } // HostedTargetFor validates a push of name@version to ecosystem and returns -// where it is stored. -func HostedTargetFor(ecosystem, name, version string) (HostedTarget, error) { - naming, ok := hostedNamings[ecosystem] +// where it is stored. filename is required for ecosystems whose pushes name +// the file (see HostedFileInPath) and must be empty for the others. +func HostedTargetFor(ecosystem, name, version, filename string) (HostedTarget, error) { + eco, ok := hostedEcosystems[ecosystem] if !ok { return HostedTarget{}, fmt.Errorf("%w: %s", ErrHostedUnsupported, ecosystem) } segments := strings.Split(name, "/") - if !naming.validName(segments) { + if !eco.validName(segments) { return HostedTarget{}, ErrHostedInvalidName } - for _, segment := range append(segments, version) { + checked := append(slices.Clone(segments), version) + if eco.filename == nil { + checked = append(checked, filename) + } else if filename != "" { + return HostedTarget{}, ErrHostedInvalidName + } + for _, segment := range checked { if !validHostedSegment(segment) { return HostedTarget{}, ErrHostedInvalidName } } - return HostedTarget{Name: name, Version: version, Filename: naming.filename(name, version)}, nil + if eco.filename != nil { + filename = eco.filename(name, version) + } + return HostedTarget{Name: name, Version: version, Filename: filename}, nil } func validHostedSegment(s string) bool { @@ -114,9 +176,11 @@ func validHostedSegment(s string) bool { // PushLocalArtifact stores body as the artifact for target and records it as // local, replacing any cached upstream copy. Download handlers serve it from // the cache without contacting the upstream, and eviction never removes it. -// The archive's manifest must name target; it is kept so the ecosystem's -// metadata can list the version with its dependencies. -func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target HostedTarget, body io.Reader, contentType string) (HostedArtifact, error) { +// A file that was already pushed is refused with ErrHostedExists. When +// wantSHA256 is set, the pushed bytes must hash to it. If the ecosystem has +// manifests, the archive's manifest must name target; it is kept so the +// ecosystem's metadata can list the version with its dependencies. +func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target HostedTarget, body io.Reader, contentType, wantSHA256 string) (HostedArtifact, error) { pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, target.Name, target.Version) if err != nil { return HostedArtifact{}, err @@ -124,6 +188,9 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target if p.versionDenied(ecosystem, target.Name, target.Version) { return HostedArtifact{}, fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) } + if err := p.checkNotPushed(versionPURL, target.Filename); err != nil { + return HostedArtifact{}, err + } push := storedPush{ storagePath: storage.FetchPath(ecosystem, target.Name, target.Version, storage.NewFetchID(), target.Filename), @@ -140,11 +207,23 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target } push.integrity = "sha512-" + base64.StdEncoding.EncodeToString(sha512Hash.Sum(nil)) + if wantSHA256 != "" && !strings.EqualFold(wantSHA256, push.sha256) { + p.discardStored(ctx, push.storagePath) + return HostedArtifact{}, fmt.Errorf("%w: got %s", ErrHostedChecksumMismatch, push.sha256) + } + if err := p.checkPush(ctx, ecosystem, target, versionPURL, &push); err != nil { p.discardStored(ctx, push.storagePath) return HostedArtifact{}, err } + // A concurrent push of the same file may have finished while this one + // was being stored and checked. + if err := p.checkNotPushed(versionPURL, target.Filename); err != nil { + p.discardStored(ctx, push.storagePath) + return HostedArtifact{}, err + } + push.pushedAt = time.Now() if err := p.recordLocalArtifact(ecosystem, target, pkgPURL, versionPURL, push); err != nil { p.discardStored(ctx, push.storagePath) @@ -160,6 +239,19 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target }, nil } +// checkNotPushed refuses a push of a file that was already pushed. A cached +// upstream copy under the same name is fine to replace. +func (p *Proxy) checkNotPushed(versionPURL, filename string) error { + record, err := p.DB.GetArtifact(versionPURL, filename) + if err != nil { + return fmt.Errorf("checking existing artifact: %w", err) + } + if record != nil && record.Source == database.SourceLocal { + return ErrHostedExists + } + return nil +} + // storedPush is a pushed archive once it is in storage. type storedPush struct { storagePath string @@ -210,10 +302,14 @@ func (p *Proxy) recordLocalArtifact(ecosystem string, target HostedTarget, pkgPU } } + // Ecosystems that name the file in the push path hold several files per + // version, and the version's integrity can only describe one of them. + // Their files are checked against their own sha256 instead. + oneFilePerVersion := hostedEcosystems[ecosystem].filename != nil if err := p.DB.UpsertVersion(&database.Version{ PURL: versionPURL, PackagePURL: pkgPURL, - Integrity: sql.NullString{String: push.integrity, Valid: true}, + Integrity: sql.NullString{String: push.integrity, Valid: oneFilePerVersion}, }); err != nil { return fmt.Errorf("upserting version: %w", err) } @@ -227,7 +323,7 @@ func (p *Proxy) recordLocalArtifact(ecosystem string, target HostedTarget, pkgPU ContentType: sql.NullString{String: push.contentType, Valid: push.contentType != ""}, FetchedAt: sql.NullTime{Time: push.pushedAt, Valid: true}, Source: database.SourceLocal, - Manifest: sql.NullString{String: string(push.manifest), Valid: true}, + Manifest: sql.NullString{String: string(push.manifest), Valid: push.manifest != nil}, }); err != nil { return fmt.Errorf("upserting artifact: %w", err) } @@ -266,19 +362,8 @@ func (p *Proxy) hostedOnly(ecosystem, name string) bool { return false } namespace, _, namespaced := strings.Cut(name, "/") - if !namespaced { - return false - } - switch ecosystem { - case "npm": - return slices.Contains(p.HostedNPMScopes, namespace) - case "composer": - return slices.ContainsFunc(p.HostedComposerVendors, func(vendor string) bool { - return strings.EqualFold(vendor, namespace) - }) - default: - return false - } + eco := hostedEcosystems[ecosystem] + return namespaced && eco.hostedOnly != nil && eco.hostedOnly(p, namespace) } // localVersion is a pushed version as ecosystem metadata lists it. @@ -328,16 +413,25 @@ func (p *Proxy) localVersions(ecosystem, name string) ([]localVersion, error) { // DeleteLocalVersion removes the archives pushed for name@version. If the // upstream has the version, clients get the upstream archive again. func (p *Proxy) DeleteLocalVersion(ecosystem, name, version string) error { + return p.deleteLocal(ecosystem, name, version, "", ErrHostedVersionNotFound) +} + +// DeleteLocalFile removes one pushed file of name@version. +func (p *Proxy) DeleteLocalFile(ecosystem, name, version, filename string) error { + return p.deleteLocal(ecosystem, name, version, filename, ErrHostedFileNotFound) +} + +func (p *Proxy) deleteLocal(ecosystem, name, version, filename string, notFound error) error { _, versionPURL, err := packagePURLStrings(ecosystem, name, version) if err != nil { return err } - deleted, err := p.DB.DeleteLocalArtifacts(versionPURL) + deleted, err := p.DB.DeleteLocalArtifacts(versionPURL, filename) if err != nil { return fmt.Errorf("deleting local artifacts: %w", err) } if deleted == 0 { - return ErrHostedVersionNotFound + return notFound } return nil } diff --git a/internal/handler/hosted_manifest.go b/internal/handler/hosted_manifest.go index 3bd9b38a..cf90dcb0 100644 --- a/internal/handler/hosted_manifest.go +++ b/internal/handler/hosted_manifest.go @@ -26,15 +26,16 @@ const maxManifestSize = 1 << 20 // readHostedManifest reads the package manifest out of a pushed archive and // checks that it describes target. The manifest is what the ecosystem's // metadata lists the version with, so its dependencies reach the client. +// Ecosystems without manifests return nil. func readHostedManifest(ecosystem string, archive io.Reader, target HostedTarget) (json.RawMessage, error) { - switch ecosystem { - case "npm": - return readNPMManifest(archive, target) - case "composer": - return readComposerManifest(archive, target) - default: + eco, ok := hostedEcosystems[ecosystem] + if !ok { return nil, fmt.Errorf("%w: %s", ErrHostedUnsupported, ecosystem) } + if eco.readManifest == nil { + return nil, nil + } + return eco.readManifest(archive, target) } // readNPMManifest reads package.json from the top directory of an npm diff --git a/internal/handler/hosted_test.go b/internal/handler/hosted_test.go index a4e7145e..252d94b6 100644 --- a/internal/handler/hosted_test.go +++ b/internal/handler/hosted_test.go @@ -15,6 +15,27 @@ import ( "github.com/git-pkgs/registries/fetch" ) +func TestHostedTargetForFileInPath(t *testing.T) { + got, err := HostedTargetFor("generic", "tools", "1.2.0", "tool.tar.gz") + if err != nil || got != (HostedTarget{Name: "tools", Version: "1.2.0", Filename: "tool.tar.gz"}) { + t.Errorf("generic target = %+v, %v", got, err) + } + for _, tt := range []struct{ ecosystem, name, version, file string }{ + {"generic", "tools", "1.2.0", ""}, + {"generic", "tools", "1.2.0", ".."}, + {"generic", "acme/tools", "1.2.0", "tool.tar.gz"}, + {"npm", "request", "2.88.3", "request-2.88.3.tgz"}, + } { + if _, err := HostedTargetFor(tt.ecosystem, tt.name, tt.version, tt.file); !errors.Is(err, ErrHostedInvalidName) { + t.Errorf("HostedTargetFor(%q, %q, %q, %q) error = %v, want ErrHostedInvalidName", + tt.ecosystem, tt.name, tt.version, tt.file, err) + } + } + if !HostedFileInPath("generic") || HostedFileInPath("npm") || HostedFileInPath("cargo") { + t.Error("HostedFileInPath should hold for generic only") + } +} + func TestHostedTargetFor(t *testing.T) { tests := []struct { name string @@ -49,7 +70,7 @@ func TestHostedTargetFor(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got, err := HostedTargetFor(tt.ecosystem, tt.pkg, tt.version) + got, err := HostedTargetFor(tt.ecosystem, tt.pkg, tt.version, "") if tt.wantErr != nil { if !errors.Is(err, tt.wantErr) { t.Fatalf("HostedTargetFor() error = %v, want %v", err, tt.wantErr) @@ -114,11 +135,11 @@ var pushedTarball = string(npmTarball(requestManifest)) func pushRequest(t *testing.T, proxy *Proxy) HostedArtifact { t.Helper() - target, err := HostedTargetFor("npm", "request", "2.88.3") + target, err := HostedTargetFor("npm", "request", "2.88.3", "") if err != nil { t.Fatalf("HostedTargetFor: %v", err) } - pushed, err := proxy.PushLocalArtifact(context.Background(), "npm", target, strings.NewReader(pushedTarball), "application/gzip") + pushed, err := proxy.PushLocalArtifact(context.Background(), "npm", target, strings.NewReader(pushedTarball), "application/gzip", "") if err != nil { t.Fatalf("PushLocalArtifact: %v", err) } @@ -290,11 +311,11 @@ func TestPushLocalArtifactChecksManifest(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { proxy, _, store, _ := setupTestProxy(t) - target, err := HostedTargetFor(tt.ecosystem, tt.pkg, tt.version) + target, err := HostedTargetFor(tt.ecosystem, tt.pkg, tt.version, "") if err != nil { t.Fatalf("HostedTargetFor: %v", err) } - _, err = proxy.PushLocalArtifact(context.Background(), tt.ecosystem, target, bytes.NewReader(tt.archive), "") + _, err = proxy.PushLocalArtifact(context.Background(), tt.ecosystem, target, bytes.NewReader(tt.archive), "", "") if tt.wantErr == nil { if err != nil { t.Fatalf("PushLocalArtifact: %v", err) @@ -401,7 +422,7 @@ func TestComposerManifestLookup(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - target, err := HostedTargetFor("composer", "monolog/monolog", "3.5.0") + target, err := HostedTargetFor("composer", "monolog/monolog", "3.5.0", "") if err != nil { t.Fatalf("HostedTargetFor: %v", err) } diff --git a/internal/handler/npm_hosted_test.go b/internal/handler/npm_hosted_test.go index 7102046c..75c82a11 100644 --- a/internal/handler/npm_hosted_test.go +++ b/internal/handler/npm_hosted_test.go @@ -60,11 +60,11 @@ func newNPMHostedFixture(t *testing.T, upstream map[string]string) npmHostedFixt func (f npmHostedFixture) push(t *testing.T, name, version, manifest string) HostedArtifact { t.Helper() - target, err := HostedTargetFor("npm", name, version) + target, err := HostedTargetFor("npm", name, version, "") if err != nil { t.Fatalf("HostedTargetFor: %v", err) } - pushed, err := f.proxy.PushLocalArtifact(context.Background(), "npm", target, bytes.NewReader(npmTarball(manifest)), "") + pushed, err := f.proxy.PushLocalArtifact(context.Background(), "npm", target, bytes.NewReader(npmTarball(manifest)), "", "") if err != nil { t.Fatalf("PushLocalArtifact: %v", err) } diff --git a/internal/server/hosted_api.go b/internal/server/hosted_api.go index e9033d64..1e4e1a4b 100644 --- a/internal/server/hosted_api.go +++ b/internal/server/hosted_api.go @@ -2,6 +2,8 @@ package server import ( "bufio" + "crypto/sha256" + "encoding/hex" "errors" "io" "log/slog" @@ -50,32 +52,41 @@ func registerHostedRoutes(r chi.Router, h *HostedAPIHandler, token string) { }) } -// HandlePush handles PUT /api/hosted/{ecosystem}/{name}/{version} +// HandlePush handles PUT /api/hosted/{ecosystem}/{name}/{version}, and +// PUT /api/hosted/generic/{name}/{version}/{file} for generic files. // @Summary Push a package archive -// @Description Stores the request body as the archive for name@version. It is served in place of the upstream archive with the same version and is never evicted. +// @Description Stores the request body as the archive for name@version. It is served in place of the upstream archive with the same version and is never evicted. Generic files are pushed to /api/hosted/generic/{name}/{version}/{file} and served at /generic/hosted/{name}/{version}/{file}. A file that was already pushed is refused with 409; delete it first to push it again. // @Tags hosted // @Accept octet-stream // @Produce json // @Security BearerAuth -// @Param ecosystem path string true "Ecosystem (npm or composer)" +// @Param ecosystem path string true "Ecosystem (npm, composer or generic)" // @Param name path string true "Package name, including an npm scope or the Composer vendor" // @Param version path string true "Version" +// @Param sha256 query string false "Expected sha256 of the body, hex-encoded; a mismatch is refused" // @Success 201 {object} HostedPushResponse // @Failure 400 {object} ErrorResponse // @Failure 401 {object} ErrorResponse // @Failure 403 {object} ErrorResponse +// @Failure 409 {object} ErrorResponse // @Failure 413 {object} ErrorResponse // @Router /api/hosted/{ecosystem}/{name}/{version} [put] func (h *HostedAPIHandler) HandlePush(w http.ResponseWriter, r *http.Request) { - ecosystem, name, version, ok := hostedVersionPath(w, r) + path, ok := hostedPath(w, r, true) if !ok { return } - target, err := handler.HostedTargetFor(ecosystem, name, version) + target, err := handler.HostedTargetFor(path.ecosystem, path.name, path.version, path.filename) if err != nil { badRequest(w, err.Error()) return } + wantSHA256 := r.URL.Query().Get("sha256") + if wantSHA256 != "" && !validSHA256(wantSHA256) { + badRequest(w, "sha256 must be 64 hex characters") + return + } + ecosystem := path.ecosystem body := bufio.NewReader(http.MaxBytesReader(w, r.Body, h.maxUploadBytes)) if _, err := body.Peek(1); errors.Is(err, io.EOF) { @@ -83,7 +94,7 @@ func (h *HostedAPIHandler) HandlePush(w http.ResponseWriter, r *http.Request) { return } - pushed, err := h.proxy.PushLocalArtifact(r.Context(), ecosystem, target, body, r.Header.Get("Content-Type")) + pushed, err := h.proxy.PushLocalArtifact(r.Context(), ecosystem, target, body, r.Header.Get("Content-Type"), wantSHA256) if err != nil { h.pushError(w, err) return @@ -102,8 +113,11 @@ func (h *HostedAPIHandler) pushError(w http.ResponseWriter, err error) { switch { case errors.As(err, &tooLarge): writeError(w, http.StatusRequestEntityTooLarge, ErrCodeBadRequest, "archive exceeds hosted.max_upload_size") - case errors.Is(err, handler.ErrHostedInvalidArchive), errors.Is(err, handler.ErrHostedManifestMismatch): + case errors.Is(err, handler.ErrHostedInvalidArchive), errors.Is(err, handler.ErrHostedManifestMismatch), + errors.Is(err, handler.ErrHostedChecksumMismatch): badRequest(w, err.Error()) + case errors.Is(err, handler.ErrHostedExists): + writeError(w, http.StatusConflict, ErrCodeConflict, err.Error()) case errors.Is(err, handler.ErrVersionDenied), errors.Is(err, handler.ErrArtifactBlocked): writeError(w, http.StatusForbidden, ErrCodeBadRequest, err.Error()) default: @@ -143,9 +157,10 @@ func (h *HostedAPIHandler) HandleList(w http.ResponseWriter, r *http.Request) { writeJSON(w, HostedListResponse{Ecosystem: ecosystem, Name: name, Artifacts: listed}) } -// HandleDelete handles DELETE /api/hosted/{ecosystem}/{name}/{version} +// HandleDelete handles DELETE /api/hosted/{ecosystem}/{name}/{version}, and +// DELETE /api/hosted/generic/{name}/{version}/{file} for one generic file. // @Summary Delete a pushed version -// @Description Removes the archives pushed for name@version. If the upstream has the version, clients get the upstream archive again. +// @Description Removes the archives pushed for name@version. If the upstream has the version, clients get the upstream archive again. For generic files, /api/hosted/generic/{name}/{version}/{file} removes only that file. // @Tags hosted // @Security BearerAuth // @Param ecosystem path string true "Ecosystem" @@ -156,19 +171,25 @@ func (h *HostedAPIHandler) HandleList(w http.ResponseWriter, r *http.Request) { // @Failure 404 {object} ErrorResponse // @Router /api/hosted/{ecosystem}/{name}/{version} [delete] func (h *HostedAPIHandler) HandleDelete(w http.ResponseWriter, r *http.Request) { - ecosystem, name, version, ok := hostedVersionPath(w, r) + path, ok := hostedPath(w, r, false) if !ok { return } - err := h.proxy.DeleteLocalVersion(ecosystem, name, version) + var err error + if path.filename != "" { + err = h.proxy.DeleteLocalFile(path.ecosystem, path.name, path.version, path.filename) + } else { + err = h.proxy.DeleteLocalVersion(path.ecosystem, path.name, path.version) + } switch { - case errors.Is(err, handler.ErrHostedVersionNotFound): + case errors.Is(err, handler.ErrHostedVersionNotFound), errors.Is(err, handler.ErrHostedFileNotFound): notFound(w, err.Error()) case err != nil: h.logger.Error("failed to delete pushed version", "error", err) internalError(w, "failed to delete pushed version") default: - h.logger.Info("deleted pushed version", "ecosystem", ecosystem, "name", name, "version", version) + h.logger.Info("deleted pushed version", "ecosystem", path.ecosystem, "name", path.name, + "version", path.version, "filename", path.filename) w.WriteHeader(http.StatusNoContent) } } @@ -176,19 +197,53 @@ func (h *HostedAPIHandler) HandleDelete(w http.ResponseWriter, r *http.Request) // minVersionPathSegments is one name segment plus the version. const minVersionPathSegments = 2 -// hostedVersionPath splits {name}/{version} from the wildcard. The version is -// always the last segment; the name may span several (npm scopes, Composer -// vendors). -func hostedVersionPath(w http.ResponseWriter, r *http.Request) (ecosystem, name, version string, ok bool) { +// hostedPushPath is a parsed /api/hosted/{ecosystem}/... path. +type hostedPushPath struct { + ecosystem string + name string + version string + filename string +} + +// hostedPath splits the wildcard into name, version and, for ecosystems +// whose pushes name the file, the filename. The version is the last segment +// otherwise, and the name may span several (npm scopes, Composer vendors). +// Ecosystems that name the file have single-segment names, so their path is +// {name}/{version}/{file}; requireFile is false for deletes, where +// {name}/{version} removes the whole version. +func hostedPath(w http.ResponseWriter, r *http.Request, requireFile bool) (hostedPushPath, bool) { segments, err := packagePathSegments(r) if err != nil { badRequest(w, err.Error()) - return "", "", "", false + return hostedPushPath{}, false + } + path := hostedPushPath{ecosystem: chi.URLParam(r, "ecosystem")} + if handler.HostedFileInPath(path.ecosystem) { + switch { + case len(segments) == minVersionPathSegments+1: + path.name, path.version, path.filename = segments[0], segments[1], segments[2] + case len(segments) == minVersionPathSegments && !requireFile: + path.name, path.version = segments[0], segments[1] + default: + badRequest(w, "path must be /api/hosted/"+path.ecosystem+"/{name}/{version}/{file}") + return hostedPushPath{}, false + } + return path, true } if len(segments) < minVersionPathSegments { badRequest(w, "path must be /api/hosted/{ecosystem}/{name}/{version}") - return "", "", "", false + return hostedPushPath{}, false } last := len(segments) - 1 - return chi.URLParam(r, "ecosystem"), strings.Join(segments[:last], "/"), segments[last], true + path.name, path.version = strings.Join(segments[:last], "/"), segments[last] + return path, true +} + +// validSHA256 reports whether s is a hex-encoded sha256 digest. +func validSHA256(s string) bool { + if len(s) != sha256.Size*2 { + return false + } + _, err := hex.DecodeString(s) + return err == nil } diff --git a/internal/server/hosted_api_test.go b/internal/server/hosted_api_test.go index 101ff318..2185f3a8 100644 --- a/internal/server/hosted_api_test.go +++ b/internal/server/hosted_api_test.go @@ -2,6 +2,8 @@ package server import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "io" "log/slog" @@ -80,6 +82,7 @@ func newHostedTestServer(t *testing.T) hostedTestServer { r.Mount("/npm", http.StripPrefix("/npm", handler.NewNPMHandler(proxy, "http://proxy.test", upstream.URL).Routes())) composer := handler.NewComposerHandlerWithUpstreams(proxy, "http://proxy.test", upstream.URL, upstream.URL) r.Mount("/composer", http.StripPrefix("/composer", composer.Routes())) + r.Mount("/generic", http.StripPrefix("/generic", handler.NewGenericHandler(proxy, nil).Routes())) return hostedTestServer{router: r, fetches: &calls, upstreamRequests: &upstreamRequests} } @@ -214,6 +217,10 @@ func TestHostedPushRejections(t *testing.T) { {name: "manifest of another version", target: "/api/hosted/npm/request/2.88.4", token: testHostedToken, body: requestTarball, wantStatus: http.StatusBadRequest}, {name: "empty body", target: "/api/hosted/npm/request/2.88.3", token: testHostedToken, wantStatus: http.StatusBadRequest}, {name: "too large", target: "/api/hosted/npm/request/2.88.3", token: testHostedToken, body: strings.Repeat("x", testHostedMaxUpload+1), wantStatus: http.StatusRequestEntityTooLarge}, + {name: "sha256 mismatch", target: "/api/hosted/npm/request/2.88.3?sha256=" + strings.Repeat("0", 64), token: testHostedToken, body: requestTarball, wantStatus: http.StatusBadRequest}, + {name: "malformed sha256", target: "/api/hosted/npm/request/2.88.3?sha256=abc", token: testHostedToken, body: requestTarball, wantStatus: http.StatusBadRequest}, + {name: "generic without file", target: "/api/hosted/generic/request/2.88.3", token: testHostedToken, body: "binary", wantStatus: http.StatusBadRequest}, + {name: "generic name with two segments", target: "/api/hosted/generic/acme/tools/1.0.0/tool.bin", token: testHostedToken, body: "binary", wantStatus: http.StatusBadRequest}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -360,3 +367,118 @@ func TestHostedComposerVersionIsInstallableFromMetadata(t *testing.T) { t.Errorf("download of %s status = %d", distPath, rec.Code) } } + +func TestHostedRepushIsRefused(t *testing.T) { + s := newHostedTestServer(t) + const target = "/api/hosted/npm/request/2.88.3" + + if rec := s.do(t, http.MethodPut, target, testHostedToken, requestTarball); rec.Code != http.StatusCreated { + t.Fatalf("first PUT status = %d, body %s", rec.Code, rec.Body) + } + if rec := s.do(t, http.MethodPut, target, testHostedToken, requestTarball); rec.Code != http.StatusConflict { + t.Fatalf("second PUT status = %d, want 409, body %s", rec.Code, rec.Body) + } + if rec := s.do(t, http.MethodDelete, target, testHostedToken, ""); rec.Code != http.StatusNoContent { + t.Fatalf("DELETE status = %d", rec.Code) + } + if rec := s.do(t, http.MethodPut, target, testHostedToken, requestTarball); rec.Code != http.StatusCreated { + t.Fatalf("PUT after DELETE status = %d, want 201, body %s", rec.Code, rec.Body) + } +} + +func TestHostedPushWithMatchingSHA256(t *testing.T) { + s := newHostedTestServer(t) + digest := sha256.Sum256([]byte(requestTarball)) + target := "/api/hosted/npm/request/2.88.3?sha256=" + strings.ToUpper(hex.EncodeToString(digest[:])) + + if rec := s.do(t, http.MethodPut, target, testHostedToken, requestTarball); rec.Code != http.StatusCreated { + t.Fatalf("PUT status = %d, body %s", rec.Code, rec.Body) + } +} + +func TestHostedGenericFiles(t *testing.T) { + s := newHostedTestServer(t) + const ( + arm64 = "tool-linux-arm64.tar.gz" + amd64 = "tool-linux-amd64.tar.gz" + ) + + for _, file := range []string{arm64, amd64} { + rec := s.do(t, http.MethodPut, "/api/hosted/generic/tools/1.2.0/"+file, testHostedToken, "bytes of "+file) + if rec.Code != http.StatusCreated { + t.Fatalf("PUT %s status = %d, body %s", file, rec.Code, rec.Body) + } + var pushed HostedPushResponse + if err := json.NewDecoder(rec.Body).Decode(&pushed); err != nil { + t.Fatalf("decoding PUT response: %v", err) + } + digest := sha256.Sum256([]byte("bytes of " + file)) + if pushed.Filename != file || pushed.Version != "1.2.0" || pushed.SHA256 != hex.EncodeToString(digest[:]) { + t.Errorf("PUT response = %+v", pushed) + } + } + + rec := s.do(t, http.MethodGet, "/generic/hosted/tools/1.2.0/"+arm64, "", "") + if rec.Code != http.StatusOK || rec.Body.String() != "bytes of "+arm64 { + t.Fatalf("download status = %d, body %q", rec.Code, rec.Body) + } + digest := sha256.Sum256([]byte("bytes of " + arm64)) + if etag := rec.Header().Get("ETag"); etag != `"`+hex.EncodeToString(digest[:])+`"` { + t.Errorf("ETag = %q, want the sha256", etag) + } + if rec := s.do(t, http.MethodHead, "/generic/hosted/tools/1.2.0/"+arm64, "", ""); rec.Code != http.StatusOK { + t.Errorf("HEAD status = %d", rec.Code) + } + + listed := s.do(t, http.MethodGet, "/api/hosted/generic/tools", "", "") + var list HostedListResponse + if err := json.NewDecoder(listed.Body).Decode(&list); err != nil { + t.Fatalf("decoding list: %v", err) + } + if len(list.Artifacts) != 2 { + t.Fatalf("listed %d files, want 2: %+v", len(list.Artifacts), list.Artifacts) + } + + if rec := s.do(t, http.MethodDelete, "/api/hosted/generic/tools/1.2.0/"+arm64, testHostedToken, ""); rec.Code != http.StatusNoContent { + t.Fatalf("DELETE file status = %d, body %s", rec.Code, rec.Body) + } + if rec := s.do(t, http.MethodGet, "/generic/hosted/tools/1.2.0/"+arm64, "", ""); rec.Code != http.StatusNotFound { + t.Errorf("download of deleted file status = %d, want 404", rec.Code) + } + if rec := s.do(t, http.MethodGet, "/generic/hosted/tools/1.2.0/"+amd64, "", ""); rec.Code != http.StatusOK { + t.Errorf("download of the other file status = %d, want 200", rec.Code) + } + if rec := s.do(t, http.MethodDelete, "/api/hosted/generic/tools/1.2.0/"+arm64, testHostedToken, ""); rec.Code != http.StatusNotFound { + t.Errorf("second DELETE status = %d, want 404", rec.Code) + } + + for _, path := range []string{ + "/generic/hosted/tools/1.2.0/missing.bin", + "/generic/hosted/tools/1.2.0", + "/generic/hosted/tools/1.2.0/a/b", + } { + if rec := s.do(t, http.MethodGet, path, "", ""); rec.Code != http.StatusNotFound { + t.Errorf("GET %s status = %d, want 404", path, rec.Code) + } + } + if *s.fetches != 0 || *s.upstreamRequests != 0 { + t.Errorf("hosted generic files went upstream: %d fetches, %d requests", *s.fetches, *s.upstreamRequests) + } +} + +func TestHostedGenericFileIsServedByTheRunningServer(t *testing.T) { + base := startHostedServer(t, config.HostedConfig{Enabled: true, Token: testHostedToken}) + if status := putBody(t, base+"/api/hosted/generic/tools/1.2.0/tool.tar.gz", testHostedToken, "tool bytes"); status != http.StatusCreated { + t.Fatalf("push status = %d, want 201", status) + } + + resp, err := http.Get(base + "/generic/hosted/tools/1.2.0/tool.tar.gz") + if err != nil { + t.Fatalf("GET file: %v", err) + } + defer func() { _ = resp.Body.Close() }() + body, _ := io.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK || string(body) != "tool bytes" { + t.Errorf("download status = %d, body %q", resp.StatusCode, body) + } +} From 5c0120f3050ad7df5f9eaeeab351123bf0e39d65 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 20 Oct 2026 20:03:06 +0200 Subject: [PATCH 2/5] Record pushes atomically and address review findings - Record the package, version and artifact of a push in one transaction that only inserts the artifact if no pushed copy exists. Of two concurrent pushes of the same file exactly one wins, and the loser's version integrity is rolled back with it. - Check for an existing file before reading the request body, and send Expect: 100-continue from proxy hosted push, so a refused re-push isn't uploaded first. - Leave an existing version record alone for multi-file ecosystems, so a generic push doesn't reset a version cached from upstream.generic. - Reserve the upstream.generic name "hosted" only while hosting is enabled, and only intercept /generic/hosted/ then. - Accept a file argument in proxy hosted delete for generic only. - Mark multi-file ecosystems explicitly, and give parse a min/max range. --- cmd/proxy/hosted.go | 34 ++++--- cmd/proxy/hosted_test.go | 21 ++++- docs/configuration.md | 4 +- internal/config/config.go | 10 ++- internal/config/config_test.go | 23 +++-- internal/database/local_artifacts.go | 85 ++++++++++++++++++ internal/database/local_artifacts_test.go | 89 ++++++++++++++++++ internal/handler/generic.go | 6 +- internal/handler/generic_test.go | 17 ++++ internal/handler/hosted.go | 105 +++++++++++----------- internal/handler/hosted_test.go | 53 +++++++++++ internal/server/hosted_api.go | 7 ++ internal/server/hosted_api_test.go | 24 +++++ 13 files changed, 394 insertions(+), 84 deletions(-) diff --git a/cmd/proxy/hosted.go b/cmd/proxy/hosted.go index ec61b3ab..36a798d4 100644 --- a/cmd/proxy/hosted.go +++ b/cmd/proxy/hosted.go @@ -24,10 +24,11 @@ const ( hostedTokenEnv = "PROXY_HOSTED_TOKEN" hostedClientTimeout = 10 * time.Minute - pushArgCount = 4 // ecosystem, name, version, file - listArgCount = 2 // ecosystem, name - deleteArgCount = 3 // ecosystem, name, version[, file] - tablePadding = 2 + pushArgCount = 4 // ecosystem, name, version, file + listArgCount = 2 // ecosystem, name + deleteArgCount = 3 // ecosystem, name, version + deleteMaxArgCount = 4 // ecosystem, name, version, file for generic files + tablePadding = 2 ) // hostedArtifact mirrors the JSON the /api/hosted endpoints return. @@ -74,17 +75,16 @@ func newHostedCommand(name, usage, summary string, getenv func(string) string) h return cmd } -// parse parses flags and checks for exactly want positional arguments, or -// one more when optional is set. -func (c hostedCommand) parse(args []string, want int, optional ...bool) ([]string, error) { +// parse parses flags and checks for between minArgs and maxArgs positional +// arguments. +func (c hostedCommand) parse(args []string, minArgs, maxArgs int) ([]string, error) { if err := c.flags.Parse(args); err != nil { if errors.Is(err, flag.ErrHelp) { c.flags.Usage() } return nil, fmt.Errorf("%w\nusage: %s", err, c.usage) } - n := c.flags.NArg() - if n != want && (len(optional) == 0 || !optional[0] || n != want+1) { + if n := c.flags.NArg(); n < minArgs || n > maxArgs { return nil, fmt.Errorf("usage: %s", c.usage) } return c.flags.Args(), nil @@ -112,7 +112,7 @@ func hostedToken(getenv func(string) string) (string, error) { func runPush(args []string, getenv func(string) string, stdout io.Writer, client *http.Client) error { cmd := newHostedCommand("push", "proxy hosted push [-url URL] ", "Push a package archive to the proxy", getenv) - positional, err := cmd.parse(args, pushArgCount) + positional, err := cmd.parse(args, pushArgCount, pushArgCount) if err != nil { return err } @@ -131,7 +131,8 @@ func runPush(args []string, getenv func(string) string, stdout io.Writer, client // The proxy checks the upload against this, so a truncated or corrupted // transfer is refused rather than stored. digest := sha256.New() - if _, err := io.Copy(digest, file); err != nil { + size, err := io.Copy(digest, file) + if err != nil { return err } if _, err := file.Seek(0, io.SeekStart); err != nil { @@ -147,8 +148,12 @@ func runPush(args []string, getenv func(string) string, stdout io.Writer, client if err != nil { return err } + req.ContentLength = size req.Header.Set("Authorization", "Bearer "+token) req.Header.Set("Content-Type", "application/octet-stream") + // The proxy refuses an existing file or a bad token before reading the + // body, so the file is only uploaded once it would be accepted. + req.Header.Set("Expect", "100-continue") var pushed hostedPushResponse if err := doHosted(client, req, http.StatusCreated, &pushed); err != nil { @@ -161,7 +166,7 @@ func runPush(args []string, getenv func(string) string, stdout io.Writer, client func runList(args []string, getenv func(string) string, stdout io.Writer, client *http.Client) error { cmd := newHostedCommand("list", "proxy hosted list [-url URL] ", "List pushed package archives", getenv) - positional, err := cmd.parse(args, listArgCount) + positional, err := cmd.parse(args, listArgCount, listArgCount) if err != nil { return err } @@ -185,10 +190,13 @@ func runList(args []string, getenv func(string) string, stdout io.Writer, client func runDelete(args []string, getenv func(string) string, stdout io.Writer, client *http.Client) error { cmd := newHostedCommand("delete", "proxy hosted delete [-url URL] [file]", "Delete a pushed version, or one generic file of it", getenv) - positional, err := cmd.parse(args, deleteArgCount, true) + positional, err := cmd.parse(args, deleteArgCount, deleteMaxArgCount) if err != nil { return err } + if len(positional) > deleteArgCount && !handler.HostedFileInPath(positional[0]) { + return fmt.Errorf("only generic files can be deleted one at a time\nusage: %s", cmd.usage) + } token, err := hostedToken(getenv) if err != nil { return err diff --git a/cmd/proxy/hosted_test.go b/cmd/proxy/hosted_test.go index 34f9a7b3..33fe2264 100644 --- a/cmd/proxy/hosted_test.go +++ b/cmd/proxy/hosted_test.go @@ -12,7 +12,8 @@ import ( ) type recordedRequest struct { - method, path, query, authorization, body string + method, path, query, authorization, body, expect string + contentLength int64 } func hostedTestServer(t *testing.T, status int, response string) (*httptest.Server, *recordedRequest) { @@ -26,6 +27,8 @@ func hostedTestServer(t *testing.T, status int, response string) (*httptest.Serv query: r.URL.RawQuery, authorization: r.Header.Get("Authorization"), body: string(body), + expect: r.Header.Get("Expect"), + contentLength: r.ContentLength, } w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) @@ -67,6 +70,9 @@ func TestRunPush(t *testing.T) { if recorded.authorization != "Bearer tok" || recorded.body != "tarball" || recorded.query != "sha256="+tarballSHA256 { t.Errorf("authorization/body/query = %q/%q/%q", recorded.authorization, recorded.body, recorded.query) } + if recorded.expect != "100-continue" || recorded.contentLength != int64(len("tarball")) { + t.Errorf("expect/content length = %q/%d, want 100-continue and the file size", recorded.expect, recorded.contentLength) + } if out := stdout.String(); !strings.Contains(out, "node-20.1.0.tgz") || !strings.Contains(out, "abc123") { t.Errorf("output = %q, want filename and sha256", out) } @@ -175,6 +181,19 @@ func TestRunDeleteOneFile(t *testing.T) { } } +func TestRunDeleteOneFileOnlyForGeneric(t *testing.T) { + srv, recorded := hostedTestServer(t, http.StatusNoContent, "") + + err := runDelete([]string{"-url", srv.URL, "npm", "@acme", "widget", "1.0.0"}, + hostedEnv(map[string]string{"PROXY_HOSTED_TOKEN": "tok"}), io.Discard, srv.Client()) + if err == nil || !strings.Contains(err.Error(), "only generic files") { + t.Errorf("npm delete with a fourth argument error = %v, want a refusal", err) + } + if recorded.method != "" { + t.Errorf("refused delete sent %s %s", recorded.method, recorded.path) + } +} + func TestRunHostedDispatchesSubcommands(t *testing.T) { srv, recorded := hostedTestServer(t, http.StatusOK, `{"ecosystem":"npm","name":"request","artifacts":[]}`) diff --git a/docs/configuration.md b/docs/configuration.md index a08ec4c4..f805387d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1123,7 +1123,7 @@ hosted: 4. Check what is pushed: `proxy hosted list npm request` or `GET /api/hosted/npm/request`. 5. Remove it again: `proxy hosted delete npm request 2.88.3` or `DELETE /api/hosted/npm/request/2.88.3`. If the upstream has the version, clients get the upstream archive again. -A pushed file is never replaced: pushing the same version again answers 409. Clients that pinned the file by its hash would otherwise get different bytes under the same version. To push it again, delete it first. +A pushed file is never replaced: pushing the same version again answers 409, before the body is uploaded when the client sends `Expect: 100-continue` (as `proxy hosted push` does). Clients that pinned the file by its hash would otherwise get different bytes under the same version. To push it again, delete it first. ### Supported ecosystems @@ -1180,7 +1180,7 @@ They are downloaded without a token from `/generic/hosted/{name}/{version}/{file curl -O https://proxy.example.com/generic/hosted/tools/1.2.0/tool-linux-arm64.tar.gz ``` -Generic files never come from an upstream: anything not pushed answers 404, and `hosted` can't be used as an `upstream.generic` name. `proxy hosted list generic tools` lists all versions and files. `proxy hosted delete generic tools 1.2.0` removes a version with all its files, and `proxy hosted delete generic tools 1.2.0 tool-linux-arm64.tar.gz` removes only that file. +Generic files never come from an upstream: anything not pushed answers 404. While hosting is enabled, `hosted` can't be used as an `upstream.generic` name. `proxy hosted list generic tools` lists all versions and files. `proxy hosted delete generic tools 1.2.0` removes a version with all its files, and `proxy hosted delete generic tools 1.2.0 tool-linux-arm64.tar.gz` removes only that file. ### Storage and eviction diff --git a/internal/config/config.go b/internal/config/config.go index e788f8d9..b06122f6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -840,9 +840,6 @@ func (u *UpstreamConfig) Validate() error { if err := validateOCIMirrors(u.OCIMirrors, u.OCI); err != nil { return err } - if _, reserved := u.Generic[HostedGenericRepository]; reserved { - return fmt.Errorf("invalid upstream.generic name %q: reserved for pushed files at /generic/%s/", HostedGenericRepository, HostedGenericRepository) - } if err := validateNamedUpstreams("upstream.generic", u.Generic); err != nil { return err } @@ -898,7 +895,8 @@ func (c *Config) validateUpstream() error { } // HostedGenericRepository is the /generic/{name}/ path that serves files -// pushed through /api/hosted/generic. No upstream.generic entry may use it. +// pushed through /api/hosted/generic. While hosted packages are enabled, no +// upstream.generic entry may use it. const HostedGenericRepository = "hosted" // CompileGenericArtifactPattern compiles an upstream.generic_artifacts pattern @@ -1511,6 +1509,10 @@ func (c *Config) validateComponents() error { if err := c.Hosted.Validate(); err != nil { return err } + if _, reserved := c.Upstream.Generic[HostedGenericRepository]; reserved && c.Hosted.Enabled { + return fmt.Errorf("invalid upstream.generic name %q: reserved for pushed files at /generic/%s/ when hosted.enabled is set", + HostedGenericRepository, HostedGenericRepository) + } if err := c.Mirror.Validate(); err != nil { return err diff --git a/internal/config/config_test.go b/internal/config/config_test.go index b21c3480..224c6517 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1455,13 +1455,6 @@ func TestValidateNamedUpstreams(t *testing.T) { }, wantErr: true, }, - { - name: "generic upstream name reserved for pushed files", - modify: func(cfg *Config) { - cfg.Upstream.Generic = map[string]string{"hosted": "https://example.com"} - }, - wantErr: true, - }, { name: "generic artifact patterns", modify: func(cfg *Config) { @@ -1892,3 +1885,19 @@ func TestMavenRepositoryNames(t *testing.T) { t.Error("maven_repositories with an invalid URL: want an error") } } + +func TestGenericUpstreamNamedHosted(t *testing.T) { + cfg := Default() + cfg.Upstream.Generic = map[string]string{"hosted": "https://example.com"} + if err := cfg.Validate(); err != nil { + t.Errorf("upstream.generic named hosted with hosting off: %v, want it accepted", err) + } + + cfg = Default() + cfg.Upstream.Generic = map[string]string{"hosted": "https://example.com"} + cfg.Hosted.Enabled = true + cfg.Hosted.Token = "secret" + if err := cfg.Validate(); err == nil || !strings.Contains(err.Error(), "reserved") { + t.Errorf("upstream.generic named hosted with hosting on: %v, want it refused", err) + } +} diff --git a/internal/database/local_artifacts.go b/internal/database/local_artifacts.go index 1bb391fd..1bd85552 100644 --- a/internal/database/local_artifacts.go +++ b/internal/database/local_artifacts.go @@ -2,6 +2,7 @@ package database import ( "database/sql" + "errors" "fmt" "time" ) @@ -73,3 +74,87 @@ func (db *DB) DeleteLocalArtifacts(versionPURL, filename string) (int, error) { } return int(deleted), tx.Commit() } + +// ErrLocalArtifactExists is returned by RecordLocalArtifact when the file +// was already pushed. +var ErrLocalArtifactExists = errors.New("local artifact already exists") + +// RecordLocalArtifact records a pushed artifact in one transaction: the +// package if it is missing, the version, and the artifact. An artifact cached +// from an upstream under the same name is replaced and its stored object +// queued for deletion; a pushed one is never replaced, and the whole record +// is refused with ErrLocalArtifactExists, so of two concurrent pushes of the +// same file exactly one wins. +// +// When setIntegrity is false, an existing version record is left as it is. +// Otherwise its integrity is set to v.Integrity. +func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, setIntegrity bool, a *Artifact) error { + tx, err := db.Beginx() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + now := time.Now() + + if _, err := tx.Exec(db.Rebind(` + INSERT INTO packages (purl, ecosystem, name, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(purl) DO NOTHING + `), pkg.PURL, pkg.Ecosystem, pkg.Name, now, now); err != nil { + return fmt.Errorf("inserting package: %w", err) + } + + versionConflict := `DO NOTHING` + if setIntegrity { + versionConflict = `DO UPDATE SET integrity = excluded.integrity, updated_at = excluded.updated_at` + } + if _, err := tx.Exec(db.Rebind(` + INSERT INTO versions (purl, package_purl, integrity, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(purl) `+versionConflict, + ), v.PURL, v.PackagePURL, v.Integrity, now, now); err != nil { + return fmt.Errorf("recording version: %w", err) + } + + var previous sql.NullString + if err := tx.Get(&previous, db.Rebind(`SELECT storage_path FROM artifacts WHERE version_purl = ? AND filename = ?`), + a.VersionPURL, a.Filename); err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("reading artifact: %w", err) + } + + res, err := tx.Exec(db.Rebind(` + INSERT INTO artifacts (version_purl, filename, upstream_url, storage_path, content_hash, + size, content_type, fetched_at, hit_count, created_at, updated_at, + source, manifest) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?) + ON CONFLICT(version_purl, filename) DO UPDATE SET + upstream_url = excluded.upstream_url, + storage_path = excluded.storage_path, + content_hash = excluded.content_hash, + size = excluded.size, + content_type = excluded.content_type, + fetched_at = excluded.fetched_at, + updated_at = excluded.updated_at, + source = excluded.source, + manifest = excluded.manifest + WHERE artifacts.source IS NULL OR artifacts.source <> 'local' + `), a.VersionPURL, a.Filename, a.UpstreamURL, a.StoragePath, a.ContentHash, + a.Size, a.ContentType, inUTC(a.FetchedAt), now, now, SourceLocal, a.Manifest) + if err != nil { + return fmt.Errorf("recording artifact: %w", err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("recording artifact: %w", err) + } + if n == 0 { + return ErrLocalArtifactExists + } + + if previous.Valid && previous.String != a.StoragePath.String { + if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), previous.String, now.UTC()); err != nil { + return fmt.Errorf("queueing replaced artifact: %w", err) + } + } + return tx.Commit() +} diff --git a/internal/database/local_artifacts_test.go b/internal/database/local_artifacts_test.go index 02335da3..08caceae 100644 --- a/internal/database/local_artifacts_test.go +++ b/internal/database/local_artifacts_test.go @@ -2,6 +2,7 @@ package database import ( "database/sql" + "errors" "path/filepath" "testing" "time" @@ -295,3 +296,91 @@ func TestUpsertPackageKeepsSource(t *testing.T) { } }) } + +func localPush(versionPURL, filename, storagePath string) *Artifact { + return &Artifact{ + VersionPURL: versionPURL, + Filename: filename, + StoragePath: sql.NullString{String: storagePath, Valid: true}, + ContentHash: sql.NullString{String: storagePath, Valid: true}, + Size: sql.NullInt64{Int64: 1, Valid: true}, + FetchedAt: sql.NullTime{Time: time.Now(), Valid: true}, + } +} + +func TestRecordLocalArtifactNeverReplacesAPush(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + pkg := &Package{PURL: "pkg:npm/request", Ecosystem: "npm", Name: "request"} + version := func(integrity string) *Version { + return &Version{PURL: "pkg:npm/request@1.0.0", PackagePURL: pkg.PURL, + Integrity: sql.NullString{String: integrity, Valid: true}} + } + + first := localPush("pkg:npm/request@1.0.0", "request-1.0.0.tgz", "npm/request/first") + if err := db.RecordLocalArtifact(pkg, version("sha512-first"), true, first); err != nil { + t.Fatalf("first RecordLocalArtifact: %v", err) + } + second := localPush("pkg:npm/request@1.0.0", "request-1.0.0.tgz", "npm/request/second") + if err := db.RecordLocalArtifact(pkg, version("sha512-second"), true, second); !errors.Is(err, ErrLocalArtifactExists) { + t.Fatalf("second RecordLocalArtifact error = %v, want ErrLocalArtifactExists", err) + } + + art, err := db.GetArtifact("pkg:npm/request@1.0.0", "request-1.0.0.tgz") + if err != nil || art == nil || art.StoragePath.String != "npm/request/first" || art.Source != SourceLocal { + t.Fatalf("artifact after refused push = %+v, %v, want the first push", art, err) + } + v, err := db.GetVersionByPURL("pkg:npm/request@1.0.0") + if err != nil || v == nil || v.Integrity.String != "sha512-first" { + t.Errorf("version integrity after refused push = %+v, %v, want the first push's", v, err) + } + var queued int + if err := db.Get(&queued, db.Rebind(`SELECT COUNT(*) FROM pending_deletes`)); err != nil || queued != 0 { + t.Errorf("queued deletes = %d, %v, want none", queued, err) + } + }) +} + +func TestRecordLocalArtifactReplacesUpstreamCopy(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + seedSourcedArtifact(t, db, "request", SourceUpstream, 100, time.Now()) + pkg := &Package{PURL: "pkg:npm/request", Ecosystem: "npm", Name: "request"} + v := &Version{PURL: "pkg:npm/request@1.0.0", PackagePURL: pkg.PURL} + + pushed := localPush("pkg:npm/request@1.0.0", "request-1.0.0.tgz", "npm/request/pushed") + if err := db.RecordLocalArtifact(pkg, v, true, pushed); err != nil { + t.Fatalf("RecordLocalArtifact: %v", err) + } + art, err := db.GetArtifact("pkg:npm/request@1.0.0", "request-1.0.0.tgz") + if err != nil || art == nil || art.Source != SourceLocal || art.StoragePath.String != "npm/request/pushed" { + t.Fatalf("artifact = %+v, %v, want the push", art, err) + } + const upstreamPath = "npm/request/1.0.0/request-1.0.0.tgz" + var queued int + if err := db.Get(&queued, db.Rebind(`SELECT COUNT(*) FROM pending_deletes WHERE path = ?`), upstreamPath); err != nil || queued != 1 { + t.Errorf("queued deletes for the upstream copy = %d, %v, want 1", queued, err) + } + }) +} + +func TestRecordLocalArtifactCanKeepTheVersionRecord(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + pkg := &Package{PURL: "pkg:generic/tools", Ecosystem: "generic", Name: "tools"} + if err := db.UpsertPackage(pkg); err != nil { + t.Fatalf("UpsertPackage: %v", err) + } + if err := db.UpsertVersion(&Version{PURL: "pkg:generic/tools@1.2.0", PackagePURL: pkg.PURL, + License: sql.NullString{String: "MIT", Valid: true}, + Integrity: sql.NullString{String: "sha256-upstream", Valid: true}}); err != nil { + t.Fatalf("UpsertVersion: %v", err) + } + + pushed := localPush("pkg:generic/tools@1.2.0", "tool.tar.gz", "generic/tools/pushed") + if err := db.RecordLocalArtifact(pkg, &Version{PURL: "pkg:generic/tools@1.2.0", PackagePURL: pkg.PURL}, false, pushed); err != nil { + t.Fatalf("RecordLocalArtifact: %v", err) + } + v, err := db.GetVersionByPURL("pkg:generic/tools@1.2.0") + if err != nil || v == nil || v.Integrity.String != "sha256-upstream" || v.License.String != "MIT" { + t.Errorf("version after push = %+v, %v, want it unchanged", v, err) + } + }) +} diff --git a/internal/handler/generic.go b/internal/handler/generic.go index f4114307..700e53e2 100644 --- a/internal/handler/generic.go +++ b/internal/handler/generic.go @@ -126,7 +126,7 @@ func (h *GenericHandler) Routes() http.Handler { } repository, rest, ok := strings.Cut(path, "/") - if repository == HostedGenericRepository { + if repository == HostedGenericRepository && h.proxy.HostedEnabled { h.handleHosted(w, r, rest) return } @@ -146,14 +146,14 @@ func (h *GenericHandler) Routes() http.Handler { } // HostedGenericRepository is the /generic/{name}/ path that serves pushed -// generic files instead of an upstream. upstream.generic can't use the name. +// generic files instead of an upstream while hosted packages are enabled. const HostedGenericRepository = config.HostedGenericRepository // handleHosted serves a pushed file at {name}/{version}/{file}. Pushed files // never come from an upstream, so anything not pushed is a 404. func (h *GenericHandler) handleHosted(w http.ResponseWriter, r *http.Request, path string) { segments := strings.Split(path, "/") - if !h.proxy.HostedEnabled || len(segments) != hostedGenericPathSegments { + if len(segments) != hostedGenericPathSegments { http.NotFound(w, r) return } diff --git a/internal/handler/generic_test.go b/internal/handler/generic_test.go index 6948b862..424eafb4 100644 --- a/internal/handler/generic_test.go +++ b/internal/handler/generic_test.go @@ -2,6 +2,7 @@ package handler import ( "context" + "io" "net/http" "net/http/httptest" "regexp" @@ -629,3 +630,19 @@ func serveGenericRequest(h *GenericHandler, target string) *httptest.ResponseRec h.Routes().ServeHTTP(w, httptest.NewRequest(http.MethodGet, target, nil)) return w } + +func TestGenericHandler_UpstreamNamedHostedWhileHostingIsOff(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, "from upstream "+r.URL.Path) + })) + t.Cleanup(upstream.Close) + proxy, _, _, _ := setupTestProxy(t) + proxy.HTTPClient = upstream.Client() + h := NewGenericHandler(proxy, map[string]string{"hosted": upstream.URL}) + + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/hosted/index.json", nil)) + if w.Code != http.StatusOK || w.Body.String() != "from upstream /index.json" { + t.Errorf("status = %d, body %q, want the upstream response", w.Code, w.Body) + } +} diff --git a/internal/handler/hosted.go b/internal/handler/hosted.go index 3e15868f..72b39293 100644 --- a/internal/handler/hosted.go +++ b/internal/handler/hosted.go @@ -75,6 +75,10 @@ type hostedEcosystem struct { // path is unambiguous. filename func(name, version string) string + // multiFile is set when a version can hold several files. Such versions + // get no version-level integrity, which could only describe one of them. + multiFile bool + // readManifest reads the package manifest out of a pushed archive and // checks that it describes target. When nil, pushes carry no manifest. readManifest func(archive io.Reader, target HostedTarget) (json.RawMessage, error) @@ -111,6 +115,7 @@ var hostedEcosystems = map[string]hostedEcosystem{ // come from an upstream. genericEcosystem: { validName: func(segments []string) bool { return len(segments) == 1 }, + multiFile: true, }, } @@ -181,14 +186,11 @@ func validHostedSegment(s string) bool { // manifests, the archive's manifest must name target; it is kept so the // ecosystem's metadata can list the version with its dependencies. func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target HostedTarget, body io.Reader, contentType, wantSHA256 string) (HostedArtifact, error) { - pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, target.Name, target.Version) - if err != nil { + if err := p.CheckPush(ecosystem, target); err != nil { return HostedArtifact{}, err } - if p.versionDenied(ecosystem, target.Name, target.Version) { - return HostedArtifact{}, fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) - } - if err := p.checkNotPushed(versionPURL, target.Filename); err != nil { + pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, target.Name, target.Version) + if err != nil { return HostedArtifact{}, err } @@ -217,13 +219,6 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target return HostedArtifact{}, err } - // A concurrent push of the same file may have finished while this one - // was being stored and checked. - if err := p.checkNotPushed(versionPURL, target.Filename); err != nil { - p.discardStored(ctx, push.storagePath) - return HostedArtifact{}, err - } - push.pushedAt = time.Now() if err := p.recordLocalArtifact(ecosystem, target, pkgPURL, versionPURL, push); err != nil { p.discardStored(ctx, push.storagePath) @@ -239,10 +234,19 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target }, nil } -// checkNotPushed refuses a push of a file that was already pushed. A cached -// upstream copy under the same name is fine to replace. -func (p *Proxy) checkNotPushed(versionPURL, filename string) error { - record, err := p.DB.GetArtifact(versionPURL, filename) +// CheckPush refuses a push of a denied version or of a file that was already +// pushed, before any bytes are read, so a client can be told without +// uploading the file. A cached upstream copy under the same name is fine to +// replace. Recording the push checks again atomically, for concurrent pushes. +func (p *Proxy) CheckPush(ecosystem string, target HostedTarget) error { + _, versionPURL, err := packagePURLStrings(ecosystem, target.Name, target.Version) + if err != nil { + return err + } + if p.versionDenied(ecosystem, target.Name, target.Version) { + return fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) + } + record, err := p.DB.GetArtifact(versionPURL, target.Filename) if err != nil { return fmt.Errorf("checking existing artifact: %w", err) } @@ -287,45 +291,38 @@ func (p *Proxy) discardStored(ctx context.Context, storagePath string) { } } -// recordLocalArtifact writes the package, version and artifact records. The -// version's integrity is set to the pushed bytes, so the cache's integrity -// check verifies them rather than an upstream archive's digest. An existing -// package record is kept as it is, since it may describe the upstream package. +// recordLocalArtifact writes the package, version and artifact records in +// one transaction. An existing package record is kept as it is, since it may +// describe the upstream package. For ecosystems with one file per version, +// the version's integrity is set to the pushed bytes, so the cache's +// integrity check verifies them rather than an upstream archive's digest. +// Multi-file versions can't have one integrity, so an existing version record +// is left alone and each file is checked against its own sha256. func (p *Proxy) recordLocalArtifact(ecosystem string, target HostedTarget, pkgPURL, versionPURL string, push storedPush) error { - existing, err := p.DB.GetPackageByPURL(pkgPURL) - if err != nil { - return fmt.Errorf("reading package: %w", err) - } - if existing == nil { - if err := p.DB.UpsertPackage(&database.Package{PURL: pkgPURL, Ecosystem: ecosystem, Name: target.Name}); err != nil { - return fmt.Errorf("upserting package: %w", err) - } + oneFilePerVersion := !hostedEcosystems[ecosystem].multiFile + err := p.DB.RecordLocalArtifact( + &database.Package{PURL: pkgPURL, Ecosystem: ecosystem, Name: target.Name}, + &database.Version{ + PURL: versionPURL, + PackagePURL: pkgPURL, + Integrity: sql.NullString{String: push.integrity, Valid: oneFilePerVersion}, + }, + oneFilePerVersion, + &database.Artifact{ + VersionPURL: versionPURL, + Filename: target.Filename, + StoragePath: sql.NullString{String: push.storagePath, Valid: true}, + ContentHash: sql.NullString{String: push.sha256, Valid: true}, + Size: sql.NullInt64{Int64: push.size, Valid: true}, + ContentType: sql.NullString{String: push.contentType, Valid: push.contentType != ""}, + FetchedAt: sql.NullTime{Time: push.pushedAt, Valid: true}, + Manifest: sql.NullString{String: string(push.manifest), Valid: push.manifest != nil}, + }) + if errors.Is(err, database.ErrLocalArtifactExists) { + return ErrHostedExists } - - // Ecosystems that name the file in the push path hold several files per - // version, and the version's integrity can only describe one of them. - // Their files are checked against their own sha256 instead. - oneFilePerVersion := hostedEcosystems[ecosystem].filename != nil - if err := p.DB.UpsertVersion(&database.Version{ - PURL: versionPURL, - PackagePURL: pkgPURL, - Integrity: sql.NullString{String: push.integrity, Valid: oneFilePerVersion}, - }); err != nil { - return fmt.Errorf("upserting version: %w", err) - } - - if err := p.DB.UpsertArtifact(&database.Artifact{ - VersionPURL: versionPURL, - Filename: target.Filename, - StoragePath: sql.NullString{String: push.storagePath, Valid: true}, - ContentHash: sql.NullString{String: push.sha256, Valid: true}, - Size: sql.NullInt64{Int64: push.size, Valid: true}, - ContentType: sql.NullString{String: push.contentType, Valid: push.contentType != ""}, - FetchedAt: sql.NullTime{Time: push.pushedAt, Valid: true}, - Source: database.SourceLocal, - Manifest: sql.NullString{String: string(push.manifest), Valid: push.manifest != nil}, - }); err != nil { - return fmt.Errorf("upserting artifact: %w", err) + if err != nil { + return fmt.Errorf("recording pushed artifact: %w", err) } return nil } diff --git a/internal/handler/hosted_test.go b/internal/handler/hosted_test.go index 252d94b6..01993de8 100644 --- a/internal/handler/hosted_test.go +++ b/internal/handler/hosted_test.go @@ -7,8 +7,10 @@ import ( "compress/gzip" "context" "errors" + "fmt" "io" "strings" + "sync" "testing" "github.com/git-pkgs/proxy/internal/database" @@ -436,3 +438,54 @@ func TestComposerManifestLookup(t *testing.T) { }) } } + +func TestConcurrentPushesOfTheSameFileHaveOneWinner(t *testing.T) { + proxy, _, _, _ := setupTestProxy(t) + proxy.HostedEnabled = true + target, err := HostedTargetFor("generic", "tools", "1.2.0", "tool.tar.gz") + if err != nil { + t.Fatalf("HostedTargetFor: %v", err) + } + + const pushes = 8 + start := make(chan struct{}) + results := make([]error, pushes) + pushed := make([]HostedArtifact, pushes) + var wg sync.WaitGroup + for i := range pushes { + wg.Add(1) + go func() { + defer wg.Done() + <-start + body := strings.NewReader(fmt.Sprintf("build %d", i)) + pushed[i], results[i] = proxy.PushLocalArtifact(context.Background(), "generic", target, body, "", "") + }() + } + close(start) + wg.Wait() + + winner := -1 + for i, err := range results { + switch { + case err == nil && winner >= 0: + t.Fatalf("pushes %d and %d both succeeded", winner, i) + case err == nil: + winner = i + case !errors.Is(err, ErrHostedExists): + t.Fatalf("push %d error = %v, want ErrHostedExists", i, err) + } + } + if winner < 0 { + t.Fatal("no push succeeded") + } + + result, err := proxy.GetLocalArtifact(context.Background(), "generic", "tools", "1.2.0", "tool.tar.gz") + if err != nil || result == nil { + t.Fatalf("GetLocalArtifact = %v, %v", result, err) + } + served, _ := io.ReadAll(result.Reader) + _ = result.Reader.Close() + if string(served) != fmt.Sprintf("build %d", winner) { + t.Errorf("served %q, want the winning push %q", served, fmt.Sprintf("build %d", winner)) + } +} diff --git a/internal/server/hosted_api.go b/internal/server/hosted_api.go index 1e4e1a4b..ff4efeb4 100644 --- a/internal/server/hosted_api.go +++ b/internal/server/hosted_api.go @@ -88,6 +88,13 @@ func (h *HostedAPIHandler) HandlePush(w http.ResponseWriter, r *http.Request) { } ecosystem := path.ecosystem + // Refuse before reading the body, so a client sending + // "Expect: 100-continue" isn't made to upload a file that is refused. + if err := h.proxy.CheckPush(ecosystem, target); err != nil { + h.pushError(w, err) + return + } + body := bufio.NewReader(http.MaxBytesReader(w, r.Body, h.maxUploadBytes)) if _, err := body.Peek(1); errors.Is(err, io.EOF) { badRequest(w, "request body is empty") diff --git a/internal/server/hosted_api_test.go b/internal/server/hosted_api_test.go index 2185f3a8..eaa5a567 100644 --- a/internal/server/hosted_api_test.go +++ b/internal/server/hosted_api_test.go @@ -482,3 +482,27 @@ func TestHostedGenericFileIsServedByTheRunningServer(t *testing.T) { t.Errorf("download status = %d, body %q", resp.StatusCode, body) } } + +// unreadBody fails the test if the handler reads it. +type unreadBody struct{ t *testing.T } + +func (b unreadBody) Read([]byte) (int, error) { + b.t.Error("body of a refused push was read") + return 0, io.EOF +} + +func TestHostedRepushIsRefusedBeforeTheBodyIsRead(t *testing.T) { + s := newHostedTestServer(t) + const target = "/api/hosted/generic/tools/1.2.0/tool.tar.gz" + if rec := s.do(t, http.MethodPut, target, testHostedToken, "first"); rec.Code != http.StatusCreated { + t.Fatalf("first PUT status = %d, body %s", rec.Code, rec.Body) + } + + req := httptest.NewRequest(http.MethodPut, target, unreadBody{t}) + req.Header.Set("Authorization", "Bearer "+testHostedToken) + rec := httptest.NewRecorder() + s.router.ServeHTTP(rec, req) + if rec.Code != http.StatusConflict { + t.Fatalf("second PUT status = %d, want 409, body %s", rec.Code, rec.Body) + } +} From 5f431a3d892efa6db480820ba5818d68db71928e Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 20 Oct 2026 20:07:04 +0200 Subject: [PATCH 3/5] Test concurrent pushes of one file at the database level The handler-level test runs on SQLite only. This one also runs against Postgres when PROXY_DATABASE_URL is set. --- internal/database/local_artifacts_test.go | 35 +++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/internal/database/local_artifacts_test.go b/internal/database/local_artifacts_test.go index 08caceae..7930675b 100644 --- a/internal/database/local_artifacts_test.go +++ b/internal/database/local_artifacts_test.go @@ -3,7 +3,9 @@ package database import ( "database/sql" "errors" + "fmt" "path/filepath" + "sync" "testing" "time" ) @@ -384,3 +386,36 @@ func TestRecordLocalArtifactCanKeepTheVersionRecord(t *testing.T) { } }) } + +func TestRecordLocalArtifactConcurrentPushesHaveOneWinner(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + pkg := &Package{PURL: "pkg:generic/tools", Ecosystem: "generic", Name: "tools"} + const pushes = 8 + start := make(chan struct{}) + errs := make([]error, pushes) + var wg sync.WaitGroup + for i := range pushes { + wg.Go(func() { + <-start + v := &Version{PURL: "pkg:generic/tools@1.2.0", PackagePURL: pkg.PURL} + a := localPush("pkg:generic/tools@1.2.0", "tool.tar.gz", fmt.Sprintf("generic/tools/push-%d", i)) + errs[i] = db.RecordLocalArtifact(pkg, v, false, a) + }) + } + close(start) + wg.Wait() + + winners := 0 + for i, err := range errs { + switch { + case err == nil: + winners++ + case !errors.Is(err, ErrLocalArtifactExists): + t.Errorf("push %d error = %v, want ErrLocalArtifactExists", i, err) + } + } + if winners != 1 { + t.Errorf("%d pushes succeeded, want exactly 1", winners) + } + }) +} From 48916bcb09591eae000f57c6220dca6107ebe2e5 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Tue, 20 Oct 2026 20:21:01 +0200 Subject: [PATCH 4/5] Never let a cache fill replace a pushed artifact - Record a fetched artifact with its package and version in one transaction that refuses to replace a pushed artifact, so a fetch that finishes after a push keeps both the pushed file and its version's integrity. Pushes and cache fills share this write, including the compare-and-swap on the storage path that queues every replaced object. - Don't open the stored file for ecosystems without a manifest. - Derive multi-file versions from the file being named in the path. - Validate DELETE paths like PUT paths, answering 400 for malformed ones. - Parse ?sha256= with integrity.ParseHex. --- internal/database/local_artifacts.go | 95 +++++------------- internal/database/local_artifacts_test.go | 35 +++++++ internal/database/pending_deletes_test.go | 4 +- internal/database/queries.go | 113 ++++++++++++++++++---- internal/handler/handler.go | 22 ++--- internal/handler/hosted.go | 73 ++++++++------ internal/handler/hosted_test.go | 42 ++++++++ internal/server/hosted_api.go | 12 +-- internal/server/hosted_api_test.go | 15 +++ 9 files changed, 271 insertions(+), 140 deletions(-) diff --git a/internal/database/local_artifacts.go b/internal/database/local_artifacts.go index 1bd85552..26aea31d 100644 --- a/internal/database/local_artifacts.go +++ b/internal/database/local_artifacts.go @@ -2,7 +2,6 @@ package database import ( "database/sql" - "errors" "fmt" "time" ) @@ -75,10 +74,6 @@ func (db *DB) DeleteLocalArtifacts(versionPURL, filename string) (int, error) { return int(deleted), tx.Commit() } -// ErrLocalArtifactExists is returned by RecordLocalArtifact when the file -// was already pushed. -var ErrLocalArtifactExists = errors.New("local artifact already exists") - // RecordLocalArtifact records a pushed artifact in one transaction: the // package if it is missing, the version, and the artifact. An artifact cached // from an upstream under the same name is replaced and its stored object @@ -89,72 +84,28 @@ var ErrLocalArtifactExists = errors.New("local artifact already exists") // When setIntegrity is false, an existing version record is left as it is. // Otherwise its integrity is set to v.Integrity. func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, setIntegrity bool, a *Artifact) error { - tx, err := db.Beginx() - if err != nil { - return err - } - defer func() { _ = tx.Rollback() }() - now := time.Now() - - if _, err := tx.Exec(db.Rebind(` - INSERT INTO packages (purl, ecosystem, name, created_at, updated_at) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(purl) DO NOTHING - `), pkg.PURL, pkg.Ecosystem, pkg.Name, now, now); err != nil { - return fmt.Errorf("inserting package: %w", err) - } - - versionConflict := `DO NOTHING` - if setIntegrity { - versionConflict = `DO UPDATE SET integrity = excluded.integrity, updated_at = excluded.updated_at` - } - if _, err := tx.Exec(db.Rebind(` - INSERT INTO versions (purl, package_purl, integrity, created_at, updated_at) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(purl) `+versionConflict, - ), v.PURL, v.PackagePURL, v.Integrity, now, now); err != nil { - return fmt.Errorf("recording version: %w", err) - } - - var previous sql.NullString - if err := tx.Get(&previous, db.Rebind(`SELECT storage_path FROM artifacts WHERE version_purl = ? AND filename = ?`), - a.VersionPURL, a.Filename); err != nil && !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("reading artifact: %w", err) - } - - res, err := tx.Exec(db.Rebind(` - INSERT INTO artifacts (version_purl, filename, upstream_url, storage_path, content_hash, - size, content_type, fetched_at, hit_count, created_at, updated_at, - source, manifest) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?) - ON CONFLICT(version_purl, filename) DO UPDATE SET - upstream_url = excluded.upstream_url, - storage_path = excluded.storage_path, - content_hash = excluded.content_hash, - size = excluded.size, - content_type = excluded.content_type, - fetched_at = excluded.fetched_at, - updated_at = excluded.updated_at, - source = excluded.source, - manifest = excluded.manifest - WHERE artifacts.source IS NULL OR artifacts.source <> 'local' - `), a.VersionPURL, a.Filename, a.UpstreamURL, a.StoragePath, a.ContentHash, - a.Size, a.ContentType, inUTC(a.FetchedAt), now, now, SourceLocal, a.Manifest) - if err != nil { - return fmt.Errorf("recording artifact: %w", err) - } - n, err := res.RowsAffected() - if err != nil { - return fmt.Errorf("recording artifact: %w", err) - } - if n == 0 { - return ErrLocalArtifactExists - } - - if previous.Valid && previous.String != a.StoragePath.String { - if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), previous.String, now.UTC()); err != nil { - return fmt.Errorf("queueing replaced artifact: %w", err) + local := *a + local.Source = SourceLocal + return db.recordArtifact(&local, func(tx execer) error { + now := time.Now() + if _, err := tx.Exec(db.Rebind(` + INSERT INTO packages (purl, ecosystem, name, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(purl) DO NOTHING + `), pkg.PURL, pkg.Ecosystem, pkg.Name, now, now); err != nil { + return fmt.Errorf("inserting package: %w", err) } - } - return tx.Commit() + versionConflict := `DO NOTHING` + if setIntegrity { + versionConflict = `DO UPDATE SET integrity = excluded.integrity, updated_at = excluded.updated_at` + } + if _, err := tx.Exec(db.Rebind(` + INSERT INTO versions (purl, package_purl, integrity, created_at, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(purl) `+versionConflict, + ), v.PURL, v.PackagePURL, v.Integrity, now, now); err != nil { + return fmt.Errorf("recording version: %w", err) + } + return nil + }, ErrLocalArtifactExists) } diff --git a/internal/database/local_artifacts_test.go b/internal/database/local_artifacts_test.go index 7930675b..b476d8d6 100644 --- a/internal/database/local_artifacts_test.go +++ b/internal/database/local_artifacts_test.go @@ -419,3 +419,38 @@ func TestRecordLocalArtifactConcurrentPushesHaveOneWinner(t *testing.T) { } }) } + +func TestRecordCachedArtifactNeverReplacesAPush(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + pkg := &Package{PURL: "pkg:npm/request", Ecosystem: "npm", Name: "request"} + pushed := localPush("pkg:npm/request@1.0.0", "request-1.0.0.tgz", "npm/request/pushed") + if err := db.RecordLocalArtifact(pkg, &Version{PURL: "pkg:npm/request@1.0.0", PackagePURL: pkg.PURL, + Integrity: sql.NullString{String: "sha512-pushed", Valid: true}}, true, pushed); err != nil { + t.Fatalf("RecordLocalArtifact: %v", err) + } + + fetched := &Artifact{ + VersionPURL: "pkg:npm/request@1.0.0", + Filename: "request-1.0.0.tgz", + UpstreamURL: "https://registry.example/request-1.0.0.tgz", + StoragePath: sql.NullString{String: "npm/request/fetched", Valid: true}, + } + err := db.RecordCachedArtifact(pkg, &Version{PURL: "pkg:npm/request@1.0.0", PackagePURL: pkg.PURL, + Integrity: sql.NullString{String: "sha512-upstream", Valid: true}}, fetched) + if !errors.Is(err, ErrLocalArtifactExists) { + t.Fatalf("RecordCachedArtifact error = %v, want ErrLocalArtifactExists", err) + } + if err := db.UpsertArtifact(fetched); !errors.Is(err, ErrLocalArtifactExists) { + t.Errorf("UpsertArtifact error = %v, want ErrLocalArtifactExists", err) + } + + art, err := db.GetArtifact("pkg:npm/request@1.0.0", "request-1.0.0.tgz") + if err != nil || art == nil || art.Source != SourceLocal || art.StoragePath.String != "npm/request/pushed" { + t.Errorf("artifact = %+v, %v, want the push", art, err) + } + v, err := db.GetVersionByPURL("pkg:npm/request@1.0.0") + if err != nil || v == nil || v.Integrity.String != "sha512-pushed" { + t.Errorf("version = %+v, %v, want the push's integrity", v, err) + } + }) +} diff --git a/internal/database/pending_deletes_test.go b/internal/database/pending_deletes_test.go index 9af6c48d..82bdacaa 100644 --- a/internal/database/pending_deletes_test.go +++ b/internal/database/pending_deletes_test.go @@ -79,7 +79,7 @@ func TestUpsertArtifactSkipsRecordThatMoved(t *testing.T) { UpstreamURL: "https://example.com/" + pendingFilename, StoragePath: sql.NullString{String: "npm/pending/1.0.0/c/pending-1.0.0.tgz", Valid: true}, } - applied, err := db.upsertArtifactFrom(late, read) + applied, err := db.upsertArtifactFrom(db, late, read) if err != nil { t.Fatalf("upsertArtifactFrom: %v", err) } @@ -91,7 +91,7 @@ func TestUpsertArtifactSkipsRecordThatMoved(t *testing.T) { } missing := &Artifact{VersionPURL: "pkg:npm/pending@2.0.0", Filename: pendingFilename, UpstreamURL: "u"} - applied, err = db.upsertArtifactFrom(missing, sql.NullString{}) + applied, err = db.upsertArtifactFrom(db, missing, sql.NullString{}) if err != nil || !applied { t.Errorf("insert of a new record: applied=%v err=%v", applied, err) } diff --git a/internal/database/queries.go b/internal/database/queries.go index 33ea93db..9560d880 100644 --- a/internal/database/queries.go +++ b/internal/database/queries.go @@ -52,7 +52,17 @@ func (db *DB) GetPackageByEcosystemName(ecosystem, name string) (*Package, error return &pkg, nil } +// execer runs statements on the database or inside a transaction. +type execer interface { + Exec(query string, args ...any) (sql.Result, error) + Get(dest any, query string, args ...any) error +} + func (db *DB) UpsertPackage(pkg *Package) error { + return db.upsertPackage(db, pkg) +} + +func (db *DB) upsertPackage(exec execer, pkg *Package) error { now := time.Now() var query string @@ -90,7 +100,7 @@ func (db *DB) UpsertPackage(pkg *Package) error { ` } - _, err := db.Exec(query, + _, err := exec.Exec(query, pkg.PURL, pkg.Ecosystem, pkg.Name, pkg.LatestVersion, pkg.License, pkg.Description, pkg.Homepage, pkg.RepositoryURL, pkg.RegistryURL, pkg.EnrichedAt, now, now, @@ -136,6 +146,10 @@ func (db *DB) GetVersionsByPackagePURL(packagePURL string) ([]Version, error) { } func (db *DB) UpsertVersion(v *Version) error { + return db.upsertVersion(db, v) +} + +func (db *DB) upsertVersion(exec execer, v *Version) error { now := time.Now() var query string @@ -167,7 +181,7 @@ func (db *DB) UpsertVersion(v *Version) error { ` } - _, err := db.Exec(query, + _, err := exec.Exec(query, v.PURL, v.PackagePURL, v.License, v.Integrity, v.PublishedAt, v.Yanked, v.EnrichedAt, now, now, ) @@ -334,32 +348,87 @@ func inUTC(t sql.NullTime) sql.NullTime { } func (db *DB) UpsertArtifact(a *Artifact) error { - for range upsertArtifactAttempts { - var previous sql.NullString - query := db.Rebind(`SELECT storage_path FROM artifacts WHERE version_purl = ? AND filename = ?`) - if err := db.Get(&previous, query, a.VersionPURL, a.Filename); err != nil && !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("reading artifact: %w", err) + return db.recordArtifact(a, nil, ErrLocalArtifactExists) +} + +// ErrLocalArtifactExists is returned when an artifact would replace one that +// was pushed to the proxy. Pushed artifacts are only removed by deleting them. +var ErrLocalArtifactExists = errors.New("a pushed artifact already exists") + +// RecordCachedArtifact records an artifact fetched from an upstream together +// with its package and version, in one transaction. If the artifact was +// pushed in the meantime, nothing is written and ErrLocalArtifactExists is +// returned, so the push's version integrity is kept as well. +func (db *DB) RecordCachedArtifact(pkg *Package, v *Version, a *Artifact) error { + return db.recordArtifact(a, func(tx execer) error { + if err := db.upsertPackage(tx, pkg); err != nil { + return err } - applied, err := db.upsertArtifactFrom(a, previous) - if err != nil { + return db.upsertVersion(tx, v) + }, ErrLocalArtifactExists) +} + +// artifactSlot is what an artifact record holds before it is replaced. +type artifactSlot struct { + StoragePath sql.NullString `db:"storage_path"` + Source sql.NullString `db:"source"` +} + +// recordArtifact writes before's records and a in one transaction. A pushed +// artifact is never replaced: the transaction is rolled back and localErr +// returned. A storage path the record stops pointing at is queued for +// deletion in the same transaction, since each fetch or push stores its own +// object and nothing else would remove it. The write applies only if the +// record still holds the path read before it, so commits racing on one +// artifact each queue the path they replaced. +func (db *DB) recordArtifact(a *Artifact, before func(tx execer) error, localErr error) error { + for range upsertArtifactAttempts { + done, err := db.tryRecordArtifact(a, before, localErr) + if err != nil || done { return err } - if !applied { - continue + } + return errors.New("recording artifact: record kept changing") +} + +func (db *DB) tryRecordArtifact(a *Artifact, before func(tx execer) error, localErr error) (bool, error) { + tx, err := db.Beginx() + if err != nil { + return false, err + } + defer func() { _ = tx.Rollback() }() + + if before != nil { + if err := before(tx); err != nil { + return false, err } - if previous.Valid && previous.String != a.StoragePath.String { - if err := db.QueuePendingDelete(previous.String); err != nil { - return fmt.Errorf("queueing replaced artifact: %w", err) - } + } + + var current artifactSlot + query := db.Rebind(`SELECT storage_path, source FROM artifacts WHERE version_purl = ? AND filename = ?`) + if err := tx.Get(¤t, query, a.VersionPURL, a.Filename); err != nil && !errors.Is(err, sql.ErrNoRows) { + return false, fmt.Errorf("reading artifact: %w", err) + } + if current.Source.String == SourceLocal { + return false, localErr + } + + applied, err := db.upsertArtifactFrom(tx, a, current.StoragePath) + if err != nil || !applied { + return false, err + } + previous := current.StoragePath + if previous.Valid && previous.String != a.StoragePath.String { + if _, err := tx.Exec(db.Rebind(queuePendingDeleteQuery), previous.String, time.Now().UTC()); err != nil { + return false, fmt.Errorf("queueing replaced artifact: %w", err) } - return nil } - return errors.New("upserting artifact: record kept changing") + return true, tx.Commit() } -// upsertArtifactFrom writes a if the record is absent or still holds -// previous, and reports whether it did. -func (db *DB) upsertArtifactFrom(a *Artifact, previous sql.NullString) (bool, error) { +// upsertArtifactFrom writes a if the record is absent, or still holds +// previous and wasn't pushed, and reports whether it did. +func (db *DB) upsertArtifactFrom(exec execer, a *Artifact, previous sql.NullString) (bool, error) { now := time.Now() var query string @@ -380,6 +449,7 @@ func (db *DB) upsertArtifactFrom(a *Artifact, previous sql.NullString) (bool, er source = EXCLUDED.source, manifest = EXCLUDED.manifest WHERE artifacts.storage_path IS NOT DISTINCT FROM $15 + AND (artifacts.source IS NULL OR artifacts.source <> 'local') ` } else { query = ` @@ -398,10 +468,11 @@ func (db *DB) upsertArtifactFrom(a *Artifact, previous sql.NullString) (bool, er source = excluded.source, manifest = excluded.manifest WHERE artifacts.storage_path IS ? + AND (artifacts.source IS NULL OR artifacts.source <> 'local') ` } - res, err := db.Exec(query, + res, err := exec.Exec(query, a.VersionPURL, a.Filename, a.UpstreamURL, a.StoragePath, a.ContentHash, a.Size, a.ContentType, inUTC(a.FetchedAt), a.HitCount, inUTC(a.LastAccessedAt), now, now, a.sourceOrUpstream(), a.Manifest, previous, diff --git a/internal/handler/handler.go b/internal/handler/handler.go index d29508eb..9280a049 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -642,7 +642,12 @@ func (p *Proxy) storeArtifact(ctx context.Context, ecosystem, name, version, fil // Update database if err := p.updateCacheDB(ecosystem, name, pkgPURL, upstreamURL, storagePath, sharedArtifact, declared); err != nil { - p.Logger.Warn("failed to update cache database", "error", err) + if errors.Is(err, database.ErrLocalArtifactExists) { + p.Logger.Info("artifact was pushed while it was fetched; keeping the pushed copy", + "purl", versionPURL, "filename", filename) + } else { + p.Logger.Warn("failed to update cache database", "error", err) + } // Continue anyway - we have the file. Queue it for deletion in case // no record points at it; the queue skips it while one does. if qErr := p.DB.QueuePendingDelete(storagePath); qErr != nil { @@ -903,10 +908,6 @@ func (p *Proxy) updateCacheDB(ecosystem, name, pkgPURL, upstreamURL, storagePath RegistryURL: sql.NullString{String: upstreamURL, Valid: true}, EnrichedAt: sql.NullTime{Time: now, Valid: true}, } - if err := p.DB.UpsertPackage(pkg); err != nil { - return fmt.Errorf("upserting package: %w", err) - } - // Upsert version ver := &database.Version{ PURL: artifact.PURL, @@ -914,10 +915,6 @@ func (p *Proxy) updateCacheDB(ecosystem, name, pkgPURL, upstreamURL, storagePath Integrity: sql.NullString{String: declared, Valid: declared != ""}, EnrichedAt: sql.NullTime{Time: now, Valid: true}, } - if err := p.DB.UpsertVersion(ver); err != nil { - return fmt.Errorf("upserting version: %w", err) - } - // Upsert artifact art := &database.Artifact{ VersionPURL: artifact.PURL, @@ -929,10 +926,11 @@ func (p *Proxy) updateCacheDB(ecosystem, name, pkgPURL, upstreamURL, storagePath ContentType: sql.NullString{String: artifact.MediaType, Valid: true}, FetchedAt: sql.NullTime{Time: now, Valid: true}, } - if err := p.DB.UpsertArtifact(art); err != nil { - return fmt.Errorf("upserting artifact: %w", err) + // One transaction, so a push that landed while this was fetched keeps + // both its artifact and its version's integrity. + if err := p.DB.RecordCachedArtifact(pkg, ver, art); err != nil { + return fmt.Errorf("recording artifact: %w", err) } - return nil } diff --git a/internal/handler/hosted.go b/internal/handler/hosted.go index 72b39293..6342a543 100644 --- a/internal/handler/hosted.go +++ b/internal/handler/hosted.go @@ -71,14 +71,10 @@ type hostedEcosystem struct { // filename is the name clients request the archive by, derived from the // package name and version. When nil, the filename is the last segment of - // the push path, and the package name must be a single segment so the - // path is unambiguous. + // the push path, the package name must be a single segment so the path is + // unambiguous, and a version can hold several files (see multiFile). filename func(name, version string) string - // multiFile is set when a version can hold several files. Such versions - // get no version-level integrity, which could only describe one of them. - multiFile bool - // readManifest reads the package manifest out of a pushed archive and // checks that it describes target. When nil, pushes carry no manifest. readManifest func(archive io.Reader, target HostedTarget) (json.RawMessage, error) @@ -115,10 +111,16 @@ var hostedEcosystems = map[string]hostedEcosystem{ // come from an upstream. genericEcosystem: { validName: func(segments []string) bool { return len(segments) == 1 }, - multiFile: true, }, } +// multiFile reports whether a version can hold several files: those named in +// the push path. Such versions get no version-level integrity, which could +// only describe one of them. +func (e hostedEcosystem) multiFile() bool { + return e.filename == nil +} + // HostedSupported reports whether packages can be pushed for ecosystem. func HostedSupported(ecosystem string) bool { _, ok := hostedEcosystems[ecosystem] @@ -130,7 +132,7 @@ func HostedSupported(ecosystem string) bool { // version. func HostedFileInPath(ecosystem string) bool { eco, ok := hostedEcosystems[ecosystem] - return ok && eco.filename == nil + return ok && eco.multiFile() } func lastSegment(name string) string { @@ -141,29 +143,44 @@ func lastSegment(name string) string { // where it is stored. filename is required for ecosystems whose pushes name // the file (see HostedFileInPath) and must be empty for the others. func HostedTargetFor(ecosystem, name, version, filename string) (HostedTarget, error) { + if err := ValidateHostedPath(ecosystem, name, version, filename); err != nil { + return HostedTarget{}, err + } + eco := hostedEcosystems[ecosystem] + if eco.filename != nil { + filename = eco.filename(name, version) + } else if filename == "" { + return HostedTarget{}, ErrHostedInvalidName + } + return HostedTarget{Name: name, Version: version, Filename: filename}, nil +} + +// ValidateHostedPath checks the parts of a hosted API path: a supported +// ecosystem, one of its package names, and path segments that are safe to +// store. filename may be empty, as in a delete of a whole version, but is +// only allowed for ecosystems whose pushes name the file. +func ValidateHostedPath(ecosystem, name, version, filename string) error { eco, ok := hostedEcosystems[ecosystem] if !ok { - return HostedTarget{}, fmt.Errorf("%w: %s", ErrHostedUnsupported, ecosystem) + return fmt.Errorf("%w: %s", ErrHostedUnsupported, ecosystem) } segments := strings.Split(name, "/") if !eco.validName(segments) { - return HostedTarget{}, ErrHostedInvalidName + return ErrHostedInvalidName } checked := append(slices.Clone(segments), version) - if eco.filename == nil { + if filename != "" { + if !eco.multiFile() { + return ErrHostedInvalidName + } checked = append(checked, filename) - } else if filename != "" { - return HostedTarget{}, ErrHostedInvalidName } for _, segment := range checked { if !validHostedSegment(segment) { - return HostedTarget{}, ErrHostedInvalidName + return ErrHostedInvalidName } } - if eco.filename != nil { - filename = eco.filename(name, version) - } - return HostedTarget{Name: name, Version: version, Filename: filename}, nil + return nil } func validHostedSegment(s string) bool { @@ -269,14 +286,16 @@ type storedPush struct { // checkPush reads the manifest of a stored push and runs the scanners on it. func (p *Proxy) checkPush(ctx context.Context, ecosystem string, target HostedTarget, versionPURL string, push *storedPush) error { - stored, err := p.Storage.Open(ctx, push.storagePath) - if err != nil { - return fmt.Errorf("reading pushed archive: %w", err) - } - push.manifest, err = readHostedManifest(ecosystem, stored, target) - _ = stored.Close() - if err != nil { - return err + if hostedEcosystems[ecosystem].readManifest != nil { + stored, err := p.Storage.Open(ctx, push.storagePath) + if err != nil { + return fmt.Errorf("reading pushed archive: %w", err) + } + push.manifest, err = readHostedManifest(ecosystem, stored, target) + _ = stored.Close() + if err != nil { + return err + } } if p.Scanners != nil && p.Scanners.Enabled() { @@ -299,7 +318,7 @@ func (p *Proxy) discardStored(ctx context.Context, storagePath string) { // Multi-file versions can't have one integrity, so an existing version record // is left alone and each file is checked against its own sha256. func (p *Proxy) recordLocalArtifact(ecosystem string, target HostedTarget, pkgPURL, versionPURL string, push storedPush) error { - oneFilePerVersion := !hostedEcosystems[ecosystem].multiFile + oneFilePerVersion := !hostedEcosystems[ecosystem].multiFile() err := p.DB.RecordLocalArtifact( &database.Package{PURL: pkgPURL, Ecosystem: ecosystem, Name: target.Name}, &database.Version{ diff --git a/internal/handler/hosted_test.go b/internal/handler/hosted_test.go index 01993de8..fbe00a54 100644 --- a/internal/handler/hosted_test.go +++ b/internal/handler/hosted_test.go @@ -489,3 +489,45 @@ func TestConcurrentPushesOfTheSameFileHaveOneWinner(t *testing.T) { t.Errorf("served %q, want the winning push %q", served, fmt.Sprintf("build %d", winner)) } } + +// A fetch that started before a push and records after it must not replace +// the pushed artifact or its version's integrity. +func TestCacheFillAfterPushKeepsThePush(t *testing.T) { + proxy, _, store, _ := setupTestProxy(t) + proxy.HostedEnabled = true + pushRequest(t, proxy) + + const fetchedPath = "npm/request/2.88.3/fetched/request-2.88.3.tgz" + if _, _, err := store.Store(context.Background(), fetchedPath, strings.NewReader("upstream bytes")); err != nil { + t.Fatalf("storing fetched copy: %v", err) + } + fetched := testArtifact("upstream bytes", "pkg:npm/request@2.88.3", "request-2.88.3.tgz", "application/gzip") + err := proxy.updateCacheDB("npm", "request", "pkg:npm/request", "https://registry.npmjs.org/request/-/request-2.88.3.tgz", + fetchedPath, fetched, "sha512-upstream") + if !errors.Is(err, database.ErrLocalArtifactExists) { + t.Fatalf("updateCacheDB over a push error = %v, want ErrLocalArtifactExists", err) + } + + result, err := proxy.GetLocalArtifact(context.Background(), "npm", "request", "2.88.3", "request-2.88.3.tgz") + if err != nil || result == nil { + t.Fatalf("GetLocalArtifact = %v, %v, want the pushed copy", result, err) + } + served, err := io.ReadAll(result.Reader) + _ = result.Reader.Close() + if err != nil || string(served) != pushedTarball { + t.Errorf("served %d bytes, %v, want the pushed tarball, integrity intact", len(served), err) + } +} + +func TestGenericPushDoesNotReadTheStoredFile(t *testing.T) { + proxy, _, store, _ := setupTestProxy(t) + proxy.HostedEnabled = true + target, err := HostedTargetFor("generic", "tools", "1.2.0", "tool.tar.gz") + if err != nil { + t.Fatalf("HostedTargetFor: %v", err) + } + store.openErr = errors.New("storage opened") + if _, err := proxy.PushLocalArtifact(context.Background(), "generic", target, strings.NewReader("bytes"), "", ""); err != nil { + t.Fatalf("PushLocalArtifact: %v", err) + } +} diff --git a/internal/server/hosted_api.go b/internal/server/hosted_api.go index ff4efeb4..69401560 100644 --- a/internal/server/hosted_api.go +++ b/internal/server/hosted_api.go @@ -2,14 +2,13 @@ package server import ( "bufio" - "crypto/sha256" - "encoding/hex" "errors" "io" "log/slog" "net/http" "strings" + "github.com/git-pkgs/integrity" "github.com/git-pkgs/proxy/internal/handler" "github.com/go-chi/chi/v5" ) @@ -182,6 +181,10 @@ func (h *HostedAPIHandler) HandleDelete(w http.ResponseWriter, r *http.Request) if !ok { return } + if err := handler.ValidateHostedPath(path.ecosystem, path.name, path.version, path.filename); err != nil { + badRequest(w, err.Error()) + return + } var err error if path.filename != "" { err = h.proxy.DeleteLocalFile(path.ecosystem, path.name, path.version, path.filename) @@ -248,9 +251,6 @@ func hostedPath(w http.ResponseWriter, r *http.Request, requireFile bool) (hoste // validSHA256 reports whether s is a hex-encoded sha256 digest. func validSHA256(s string) bool { - if len(s) != sha256.Size*2 { - return false - } - _, err := hex.DecodeString(s) + _, err := integrity.ParseHex(integrity.SHA256, s) return err == nil } diff --git a/internal/server/hosted_api_test.go b/internal/server/hosted_api_test.go index eaa5a567..3a24aaba 100644 --- a/internal/server/hosted_api_test.go +++ b/internal/server/hosted_api_test.go @@ -506,3 +506,18 @@ func TestHostedRepushIsRefusedBeforeTheBodyIsRead(t *testing.T) { t.Fatalf("second PUT status = %d, want 409, body %s", rec.Code, rec.Body) } } + +func TestHostedDeleteRejectsMalformedPaths(t *testing.T) { + s := newHostedTestServer(t) + for _, target := range []string{ + "/api/hosted/generic/tools/1.2.0/%2e%2e", + "/api/hosted/generic/tools/1.2.0/tool%0a", + "/api/hosted/npm/request/2.88.3/request-2.88.3.tgz/extra", + "/api/hosted/composer/monolog/3.5.0", + "/api/hosted/cargo/serde/1.0.200", + } { + if rec := s.do(t, http.MethodDelete, target, testHostedToken, ""); rec.Code != http.StatusBadRequest { + t.Errorf("DELETE %s status = %d, want 400, body %s", target, rec.Code, rec.Body) + } + } +} From 0832c6142be4ddbb3936298eab1366b3383c4b80 Mon Sep 17 00:00:00 2001 From: Simon Chrzanowski Date: Wed, 21 Oct 2026 12:33:43 +0200 Subject: [PATCH 5/5] Refuse pushes of a version that holds a pushed file under another name Composer names are case-insensitive, so acme/private and acme/Private share the version record and its integrity but get different filenames. A second push under the other case was accepted and broke the integrity check of the first archive. For ecosystems with one file per version, any pushed file of the version now refuses the push, both before the upload and atomically when recording it. --- docs/configuration.md | 2 +- internal/database/local_artifacts.go | 23 ++++++++-- internal/database/local_artifacts_test.go | 51 +++++++++++++++++++++++ internal/handler/hosted.go | 26 ++++++++---- internal/server/hosted_api_test.go | 16 +++++++ 5 files changed, 106 insertions(+), 12 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index f805387d..57cc423b 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1123,7 +1123,7 @@ hosted: 4. Check what is pushed: `proxy hosted list npm request` or `GET /api/hosted/npm/request`. 5. Remove it again: `proxy hosted delete npm request 2.88.3` or `DELETE /api/hosted/npm/request/2.88.3`. If the upstream has the version, clients get the upstream archive again. -A pushed file is never replaced: pushing the same version again answers 409, before the body is uploaded when the client sends `Expect: 100-continue` (as `proxy hosted push` does). Clients that pinned the file by its hash would otherwise get different bytes under the same version. To push it again, delete it first. +A pushed file is never replaced: pushing the same version again answers 409, also under a differently cased Composer name, before the body is uploaded when the client sends `Expect: 100-continue` (as `proxy hosted push` does). Clients that pinned the file by its hash would otherwise get different bytes under the same version. To push it again, delete it first. ### Supported ecosystems diff --git a/internal/database/local_artifacts.go b/internal/database/local_artifacts.go index 26aea31d..162b4718 100644 --- a/internal/database/local_artifacts.go +++ b/internal/database/local_artifacts.go @@ -81,9 +81,13 @@ func (db *DB) DeleteLocalArtifacts(versionPURL, filename string) (int, error) { // is refused with ErrLocalArtifactExists, so of two concurrent pushes of the // same file exactly one wins. // -// When setIntegrity is false, an existing version record is left as it is. -// Otherwise its integrity is set to v.Integrity. -func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, setIntegrity bool, a *Artifact) error { +// When oneFilePerVersion is false, an existing version record is left as it +// is. Otherwise its integrity is set to v.Integrity, and the push is refused +// if the version already holds a pushed artifact under any filename, since +// aliases of one name (Composer's are case-insensitive) share the version +// record and its integrity. The version upsert locks that record first, so +// concurrent pushes of aliases see each other. +func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, oneFilePerVersion bool, a *Artifact) error { local := *a local.Source = SourceLocal return db.recordArtifact(&local, func(tx execer) error { @@ -96,7 +100,7 @@ func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, setIntegrity bool, a return fmt.Errorf("inserting package: %w", err) } versionConflict := `DO NOTHING` - if setIntegrity { + if oneFilePerVersion { versionConflict = `DO UPDATE SET integrity = excluded.integrity, updated_at = excluded.updated_at` } if _, err := tx.Exec(db.Rebind(` @@ -106,6 +110,17 @@ func (db *DB) RecordLocalArtifact(pkg *Package, v *Version, setIntegrity bool, a ), v.PURL, v.PackagePURL, v.Integrity, now, now); err != nil { return fmt.Errorf("recording version: %w", err) } + if oneFilePerVersion { + var pushed int + if err := tx.Get(&pushed, db.Rebind(` + SELECT COUNT(*) FROM artifacts WHERE version_purl = ? AND source = ? + `), v.PURL, SourceLocal); err != nil { + return fmt.Errorf("checking pushed artifacts: %w", err) + } + if pushed > 0 { + return ErrLocalArtifactExists + } + } return nil }, ErrLocalArtifactExists) } diff --git a/internal/database/local_artifacts_test.go b/internal/database/local_artifacts_test.go index b476d8d6..a6d0ed5d 100644 --- a/internal/database/local_artifacts_test.go +++ b/internal/database/local_artifacts_test.go @@ -420,6 +420,57 @@ func TestRecordLocalArtifactConcurrentPushesHaveOneWinner(t *testing.T) { }) } +// Aliases of one name, such as Composer names differing in case, share the +// version record and its integrity, so only one of them may be pushed. +func TestRecordLocalArtifactOneFilePerVersionRefusesAliases(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + pkg := &Package{PURL: "pkg:composer/acme/private", Ecosystem: "composer", Name: "acme/private"} + const pushes = 8 + start := make(chan struct{}) + errs := make([]error, pushes) + var wg sync.WaitGroup + for i := range pushes { + wg.Go(func() { + <-start + v := &Version{PURL: "pkg:composer/acme/private@1.0.0", PackagePURL: pkg.PURL, + Integrity: sql.NullString{String: fmt.Sprintf("sha512-push-%d", i), Valid: true}} + a := localPush("pkg:composer/acme/private@1.0.0", fmt.Sprintf("alias-%d-1.0.0.zip", i), + fmt.Sprintf("composer/acme/private/push-%d", i)) + errs[i] = db.RecordLocalArtifact(pkg, v, true, a) + }) + } + close(start) + wg.Wait() + + winner := -1 + for i, err := range errs { + switch { + case err == nil: + if winner >= 0 { + t.Errorf("pushes %d and %d both succeeded, want exactly 1", winner, i) + } + winner = i + case !errors.Is(err, ErrLocalArtifactExists): + t.Errorf("push %d error = %v, want ErrLocalArtifactExists", i, err) + } + } + if winner < 0 { + t.Fatal("no push succeeded") + } + artifacts, err := db.GetArtifactsByVersionPURL("pkg:composer/acme/private@1.0.0") + if err != nil { + t.Fatal(err) + } + if len(artifacts) != 1 || artifacts[0].Filename != fmt.Sprintf("alias-%d-1.0.0.zip", winner) { + t.Errorf("artifacts = %+v, want only push %d", artifacts, winner) + } + v, err := db.GetVersionByPURL("pkg:composer/acme/private@1.0.0") + if err != nil || v == nil || v.Integrity.String != fmt.Sprintf("sha512-push-%d", winner) { + t.Errorf("version = %+v, %v, want the integrity of push %d", v, err, winner) + } + }) +} + func TestRecordCachedArtifactNeverReplacesAPush(t *testing.T) { runWithBothDatabases(t, func(t *testing.T, db *DB) { pkg := &Package{PURL: "pkg:npm/request", Ecosystem: "npm", Name: "request"} diff --git a/internal/handler/hosted.go b/internal/handler/hosted.go index 6342a543..5ff43ac0 100644 --- a/internal/handler/hosted.go +++ b/internal/handler/hosted.go @@ -253,8 +253,10 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target // CheckPush refuses a push of a denied version or of a file that was already // pushed, before any bytes are read, so a client can be told without -// uploading the file. A cached upstream copy under the same name is fine to -// replace. Recording the push checks again atomically, for concurrent pushes. +// uploading the file. For ecosystems with one file per version, any pushed +// file of the version counts, so a differently cased alias of the name is +// refused too. A cached upstream copy is fine to replace. Recording the push +// checks again atomically, for concurrent pushes. func (p *Proxy) CheckPush(ecosystem string, target HostedTarget) error { _, versionPURL, err := packagePURLStrings(ecosystem, target.Name, target.Version) if err != nil { @@ -263,12 +265,22 @@ func (p *Proxy) CheckPush(ecosystem string, target HostedTarget) error { if p.versionDenied(ecosystem, target.Name, target.Version) { return fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) } - record, err := p.DB.GetArtifact(versionPURL, target.Filename) - if err != nil { - return fmt.Errorf("checking existing artifact: %w", err) + var existing []database.Artifact + if hostedEcosystems[ecosystem].multiFile() { + record, err := p.DB.GetArtifact(versionPURL, target.Filename) + if err != nil { + return fmt.Errorf("checking existing artifact: %w", err) + } + if record != nil { + existing = append(existing, *record) + } + } else if existing, err = p.DB.GetArtifactsByVersionPURL(versionPURL); err != nil { + return fmt.Errorf("checking existing artifacts: %w", err) } - if record != nil && record.Source == database.SourceLocal { - return ErrHostedExists + for _, a := range existing { + if a.Source == database.SourceLocal { + return ErrHostedExists + } } return nil } diff --git a/internal/server/hosted_api_test.go b/internal/server/hosted_api_test.go index 3a24aaba..38b34b3a 100644 --- a/internal/server/hosted_api_test.go +++ b/internal/server/hosted_api_test.go @@ -386,6 +386,22 @@ func TestHostedRepushIsRefused(t *testing.T) { } } +func TestHostedComposerRepushUnderAnotherCaseIsRefused(t *testing.T) { + s := newHostedTestServer(t) + original := composerZip(`{"name":"acme/private","description":"first"}`) + if rec := s.do(t, http.MethodPut, "/api/hosted/composer/acme/private/1.0.0", testHostedToken, original); rec.Code != http.StatusCreated { + t.Fatalf("first PUT status = %d, body %s", rec.Code, rec.Body) + } + if rec := s.do(t, http.MethodPut, "/api/hosted/composer/acme/Private/1.0.0", testHostedToken, + composerZip(`{"name":"acme/private","description":"second"}`)); rec.Code != http.StatusConflict { + t.Fatalf("PUT of acme/Private status = %d, want 409, body %s", rec.Code, rec.Body) + } + rec := s.do(t, http.MethodGet, "/composer/files/acme/private/1.0.0/private-1.0.0.zip", "", "") + if rec.Code != http.StatusOK || rec.Body.String() != original { + t.Fatalf("download status = %d, want 200 and the first archive", rec.Code) + } +} + func TestHostedPushWithMatchingSHA256(t *testing.T) { s := newHostedTestServer(t) digest := sha256.Sum256([]byte(requestTarball))