diff --git a/cmd/proxy/hosted.go b/cmd/proxy/hosted.go index bbc947df..36a798d4 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 ( @@ -19,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 - 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. @@ -69,15 +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. -func (c hostedCommand) parse(args []string, want int) ([]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) } - if c.flags.NArg() != want { + if n := c.flags.NArg(); n < minArgs || n > maxArgs { return nil, fmt.Errorf("usage: %s", c.usage) } return c.flags.Args(), nil @@ -105,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 } @@ -121,12 +128,32 @@ 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() + size, err := io.Copy(digest, file) + if 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 } + 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 { @@ -139,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 } @@ -161,16 +188,20 @@ 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, 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 } - 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 +210,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 +245,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..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) @@ -62,14 +65,34 @@ 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 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) } } +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 +164,36 @@ 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 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 507cad59..57cc423b 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, 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 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. 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 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..b06122f6 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -894,6 +894,11 @@ func (c *Config) validateUpstream() error { return nil } +// HostedGenericRepository is the /generic/{name}/ path that serves files +// 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 // so that it only matches a whole path, and checks it has the `version` and // `file` groups. @@ -1504,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 48ae1c79..224c6517 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1885,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 83197768..162b4718 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) } @@ -68,3 +73,54 @@ func (db *DB) DeleteLocalArtifacts(versionPURL string) (int, error) { } return int(deleted), tx.Commit() } + +// 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 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 { + 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 oneFilePerVersion { + 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) + } + 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 81d5ec8b..a6d0ed5d 100644 --- a/internal/database/local_artifacts_test.go +++ b/internal/database/local_artifacts_test.go @@ -2,7 +2,10 @@ package database import ( "database/sql" + "errors" + "fmt" "path/filepath" + "sync" "testing" "time" ) @@ -183,7 +186,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 +199,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) } @@ -295,3 +298,210 @@ 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) + } + }) +} + +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) + } + }) +} + +// 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"} + 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/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..700e53e2 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.proxy.HostedEnabled { + 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 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 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..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" @@ -312,6 +313,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 { @@ -628,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/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 41ccb930..5ff43ac0 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,127 @@ 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, 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 + + // 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 }, }, } +// 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 := 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.multiFile() +} + 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) { + 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 !naming.validName(segments) { - return HostedTarget{}, ErrHostedInvalidName + if !eco.validName(segments) { + return ErrHostedInvalidName + } + checked := append(slices.Clone(segments), version) + if filename != "" { + if !eco.multiFile() { + return ErrHostedInvalidName + } + checked = append(checked, filename) } - for _, segment := range append(segments, version) { + for _, segment := range checked { if !validHostedSegment(segment) { - return HostedTarget{}, ErrHostedInvalidName + return ErrHostedInvalidName } } - return HostedTarget{Name: name, Version: version, Filename: naming.filename(name, version)}, nil + return nil } func validHostedSegment(s string) bool { @@ -114,16 +198,18 @@ 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) { + if err := p.CheckPush(ecosystem, target); err != nil { + return HostedArtifact{}, err + } pkgPURL, versionPURL, err := packagePURLStrings(ecosystem, target.Name, target.Version) if err != nil { return HostedArtifact{}, err } - if p.versionDenied(ecosystem, target.Name, target.Version) { - return HostedArtifact{}, fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) - } push := storedPush{ storagePath: storage.FetchPath(ecosystem, target.Name, target.Version, storage.NewFetchID(), target.Filename), @@ -140,6 +226,11 @@ 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 @@ -160,6 +251,40 @@ func (p *Proxy) PushLocalArtifact(ctx context.Context, ecosystem string, target }, nil } +// 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. 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 { + return err + } + if p.versionDenied(ecosystem, target.Name, target.Version) { + return fmt.Errorf("%w: %s", ErrVersionDenied, versionPURL) + } + 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) + } + for _, a := range existing { + if a.Source == database.SourceLocal { + return ErrHostedExists + } + } + return nil +} + // storedPush is a pushed archive once it is in storage. type storedPush struct { storagePath string @@ -173,14 +298,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() { @@ -195,41 +322,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) - } - } - - if err := p.DB.UpsertVersion(&database.Version{ - PURL: versionPURL, - PackagePURL: pkgPURL, - Integrity: sql.NullString{String: push.integrity, Valid: true}, - }); err != nil { - return fmt.Errorf("upserting version: %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 } - - 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: true}, - }); err != nil { - return fmt.Errorf("upserting artifact: %w", err) + if err != nil { + return fmt.Errorf("recording pushed artifact: %w", err) } return nil } @@ -266,19 +390,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 +441,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..fbe00a54 100644 --- a/internal/handler/hosted_test.go +++ b/internal/handler/hosted_test.go @@ -7,14 +7,37 @@ import ( "compress/gzip" "context" "errors" + "fmt" "io" "strings" + "sync" "testing" "github.com/git-pkgs/proxy/internal/database" "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 +72,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 +137,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 +313,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 +424,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) } @@ -415,3 +438,96 @@ 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)) + } +} + +// 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/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..69401560 100644 --- a/internal/server/hosted_api.go +++ b/internal/server/hosted_api.go @@ -8,6 +8,7 @@ import ( "net/http" "strings" + "github.com/git-pkgs/integrity" "github.com/git-pkgs/proxy/internal/handler" "github.com/go-chi/chi/v5" ) @@ -50,32 +51,48 @@ 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 + + // 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) { @@ -83,7 +100,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 +119,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 +163,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 +177,29 @@ 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) + 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) + } 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 +207,50 @@ 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 { + _, 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 101ff318..38b34b3a 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,173 @@ 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 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)) + 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) + } +} + +// 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) + } +} + +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) + } + } +}