diff --git a/README.md b/README.md index 3c75169f..8ffdc4f6 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,7 @@ The proxy never uploads artifact bytes to a scanner. Each scanner is notified wi | Arch | Arch Linux | | ✗ | | Chef | Chef | | ✗ | | Generic | Any | | ✓ | +| URL (open, immutable) | Any | | ✓ | | Helm | Kubernetes | | ✓ | | Vagrant | Vagrant | | ✗ | @@ -912,6 +913,7 @@ Recently cached: | `GET /v2/homebrew/core/*` | Homebrew core bottle manifests and blobs from GHCR | | `GET /apk/{repository}/*` | Alpine APK repository protocol | | `GET /generic/{name}/*` | Generic HTTP download proxy (GitHub release assets, mise/aqua) | +| `GET /url/[sha256/{hex}/]{host}/*` | Open cache of immutable https downloads, opt-in via `url_proxy.enabled` | | `GET /debian/*` | Debian/APT repository protocol (main archive) | | `GET /debian/{repository}/*` | Debian/APT repository protocol (named archive, e.g. security) | | `GET /rpm/*` | RPM/Yum repository protocol | diff --git a/config.example.yaml b/config.example.yaml index 2058b660..ae7ab573 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -75,6 +75,22 @@ storage: # internal Host header or the SigV4 signature will not validate. # direct_serve_base_url: "https://minio.example.com" + # Base URL where the bucket serves objects anonymously, including any key + # prefix from the storage URL. When set, redirects point here instead of at + # presigned URLs, so they never expire. The bucket must allow anonymous + # reads under it (e.g. a bucket policy on the url/ prefix). + # direct_serve_public_url: "http://rgw.example.com:7480/bucket/prefix" + +# Open URL cache at /url/{host}/{path} (see docs/configuration.md). Caches any +# public https URL as an immutable artifact. Off by default: it is an open +# proxy for anyone who can reach it. +# url_proxy: +# enabled: true +# # Redirect GETs to the stored object instead of streaming it. +# direct_serve: true +# # Bound on one whole upstream download. Default: "9m". +# fetch_timeout: "9m" + # Database configuration database: # Database driver: "sqlite" (default) or "postgres" diff --git a/docs/configuration.md b/docs/configuration.md index 531952a8..72bb3e1e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -44,6 +44,7 @@ storage: | `storage.path` | `PROXY_STORAGE_PATH` | `-storage-path` | Local path (deprecated, use url) | | `storage.max_size` | `PROXY_STORAGE_MAX_SIZE` | - | Max cache size (e.g., "10GB") | | `storage.cache_artifacts` | `PROXY_STORAGE_CACHE_ARTIFACTS` | - | Store fetched artifacts (default: true); `false` streams them from upstream | +| `storage.direct_serve_public_url` | `PROXY_STORAGE_DIRECT_SERVE_PUBLIC_URL` | - | Anonymous bucket URL for redirects instead of presigning (see [Open URL cache](#open-url-cache)) | `storage.max_size` counts cached artifacts only. An artifact replaced by a refetch stays in storage for at least an hour, or `storage.direct_serve_ttl` if longer, so requests already reading it can finish, and storage use can exceed the limit by what was replaced in that time. @@ -58,7 +59,7 @@ storage: Every download is a fresh upstream fetch, and concurrent requests for the same artifact are not combined. Artifacts with a digest known up front (OCI blobs, Swift archives, Helm charts) are verified while streaming: the response is sent chunked, and on a mismatch the connection is aborted before the response completes so the client never receives a tampered artifact as a good one. The same happens when the upstream connection fails mid-download. -`cache_artifacts: false` cannot be combined with `scanning.enabled`, `storage.direct_serve` or `mirror_api`, which all need stored artifacts, and the `mirror` command refuses to run with it. +`cache_artifacts: false` cannot be combined with `scanning.enabled`, `storage.direct_serve`, `mirror_api` or `url_proxy.enabled`, which all need stored artifacts, and the `mirror` command refuses to run with it. ### Amazon S3 @@ -280,6 +281,54 @@ large mutable downloads (`releases/latest/download/...`) off this route. This is the cache behind [mise](https://mise.jdx.dev)'s aqua backend; see the mise section in the README for the client-side `url_replacements`. +### Open URL cache + +`/generic/` only reaches configured upstreams. For build scripts that download +pinned source tarballs from many, changing hosts, the `/url/` route caches any +public https URL instead: + +```yaml +url_proxy: + enabled: true # PROXY_URL_PROXY_ENABLED + direct_serve: true # PROXY_URL_PROXY_DIRECT_SERVE + fetch_timeout: "9m" # PROXY_URL_PROXY_FETCH_TIMEOUT +storage: + # Optional: where the bucket serves objects anonymously. + direct_serve_public_url: "http://rgw.example.com:7480/bucket/prefix" +``` + +`GET /url/{host}/{path}?{query}` fetches `https://{host}/{path}?{query}` and +`GET /url/sha256/{hex}/{host}/{path}` also checks the download against that +SHA-256. Every file is treated as immutable. It is fetched once, streamed into +the artifact cache with no size limit (unlike the metadata cache), and served +from there without revalidation, including while its host is down. + +- **Digest.** Pass the expected `sha256` whenever the client knows it. A download + that does not match is not cached and returns 502, and a cached copy with a + different digest is refetched. Without a digest, a URL whose content changes + keeps serving the first copy until it is evicted. +- **Redirects.** With `url_proxy.direct_serve`, GET requests get a 302 to the stored + object, on a cache hit and right after a miss is stored, so the bytes never + pass through the proxy again. The target is + `storage.direct_serve_public_url/{storage path}` when that is set, otherwise + a presigned URL valid for `storage.direct_serve_ttl`. Backends that support + neither are streamed. HEAD is answered from the cache record and never + redirected. This setting is independent of `storage.direct_serve`. +- **Expired objects.** Before redirecting, the proxy checks that the object still + exists. A record whose object was removed behind its back, for example by a + bucket lifecycle rule, is refetched once and counted in + `proxy_cache_missing_objects_total`. +- **Reachability.** Only https on port 443 is fetched. The upstream client refuses + loopback, private and link-local addresses on every redirect hop (subject + to `upstream.allow_private_hosts` and `upstream.allow_loopback`), ignores + `HTTPS_PROXY`, and never sends `upstream.auth` credentials. + +The route is still an open proxy for its clients. Anyone who can reach it can +make the proxy download and store any public file, so expose it only to +trusted networks and set `storage.max_size`. A cold miss sends no bytes +until the whole file is stored, so keep `fetch_timeout` below your ingress or +load balancer's read timeout. + `upstream.oci_default` sets the registry used by unprefixed `/v2` requests, while `upstream.oci` selects named registries through the `upstream/{name}/` repository prefix. For example, `oci://proxy.example.com/upstream/ghcr/owner/chart` diff --git a/internal/config/config.go b/internal/config/config.go index 97caa4cc..0499228a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -150,6 +150,10 @@ type Config struct { // Disabled by default to prevent unauthenticated users from triggering downloads. MirrorAPI bool `json:"mirror_api" yaml:"mirror_api"` + // URLProxy configures the open /url/ route, which caches any public + // https URL as an immutable artifact. Disabled by default. + URLProxy URLProxyConfig `json:"url_proxy" yaml:"url_proxy"` + // Gradle configures Gradle HttpBuildCache behavior. Gradle GradleConfig `json:"gradle" yaml:"gradle"` @@ -378,6 +382,34 @@ type StorageConfig struct { // of the proxy. False is incompatible with scanning, direct_serve and // mirror_api, which all depend on stored artifacts. Default: true. CacheArtifacts bool `json:"cache_artifacts" yaml:"cache_artifacts"` + // DirectServePublicURL is the base URL under which the bucket serves + // stored objects anonymously, including any key prefix of the storage + // URL (e.g. "http://rgw:7480/bucket/prefix"). When set, redirects point + // at DirectServePublicURL/{storage path} instead of a presigned URL, so + // they never expire. The bucket must allow anonymous reads there. + DirectServePublicURL string `json:"direct_serve_public_url" yaml:"direct_serve_public_url"` +} + +// URLProxyConfig configures the /url/ route. +// +// The route proxies any public https URL, so unlike upstream.generic it is an +// open HTTP proxy for its clients: keep it on a network only trusted clients +// reach. Fetched files are assumed immutable and cached in the artifact +// cache, so a URL whose content changes keeps serving the first copy unless +// the client passes the expected sha256. +type URLProxyConfig struct { + // Enabled mounts the /url/ route. + Enabled bool `json:"enabled" yaml:"enabled"` + + // DirectServe redirects GET requests to the stored object, using + // storage.direct_serve_public_url or, failing that, a presigned URL + // valid for storage.direct_serve_ttl. It applies to this route only, + // independently of storage.direct_serve. + DirectServe bool `json:"direct_serve" yaml:"direct_serve"` + + // FetchTimeout bounds one whole upstream download, body included. + // Uses Go duration syntax. Default: "9m". + FetchTimeout string `json:"fetch_timeout" yaml:"fetch_timeout"` } // GradleConfig configures Gradle-specific features. @@ -909,6 +941,7 @@ func (c *Config) LoadFromEnv() { setEnvString(&c.Storage.DirectServeTTL, "PROXY_STORAGE_DIRECT_SERVE_TTL") setEnvString(&c.Storage.DirectServeBaseURL, "PROXY_STORAGE_DIRECT_SERVE_BASE_URL") setEnvBool(&c.Storage.CacheArtifacts, "PROXY_STORAGE_CACHE_ARTIFACTS") + setEnvString(&c.Storage.DirectServePublicURL, "PROXY_STORAGE_DIRECT_SERVE_PUBLIC_URL") setEnvString(&c.Database.Driver, "PROXY_DATABASE_DRIVER") setEnvString(&c.Database.Path, "PROXY_DATABASE_PATH") setEnvString(&c.Database.URL, "PROXY_DATABASE_URL") @@ -953,6 +986,9 @@ func (c *Config) LoadFromEnv() { setEnvString(&c.Scanning.FetchBaseURL, "PROXY_SCANNING_FETCH_BASE_URL") setEnvBool(&c.CacheMetadata, "PROXY_CACHE_METADATA") setEnvBool(&c.MirrorAPI, "PROXY_MIRROR_API") + setEnvBool(&c.URLProxy.Enabled, "PROXY_URL_PROXY_ENABLED") + setEnvBool(&c.URLProxy.DirectServe, "PROXY_URL_PROXY_DIRECT_SERVE") + setEnvString(&c.URLProxy.FetchTimeout, "PROXY_URL_PROXY_FETCH_TIMEOUT") setEnvString(&c.MetadataTTL, "PROXY_METADATA_TTL") setEnvString(&c.MetadataMaxSize, "PROXY_METADATA_MAX_SIZE") setEnvString(&c.HTTPTimeout, "PROXY_HTTP_TIMEOUT") @@ -974,6 +1010,33 @@ func validateAbsoluteURL(fieldName, value string) error { return nil } +// validateHTTPURL is validateAbsoluteURL restricted to http and https, with no +// query or fragment, for URLs that object paths are appended to. +func validateHTTPURL(fieldName, value string) error { + u, err := url.Parse(value) + if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" || + u.RawQuery != "" || u.Fragment != "" { + return fmt.Errorf("invalid %s %q: must be an http or https URL without query or fragment", fieldName, value) + } + return nil +} + +// validateURLProxy checks the /url/ route settings and the public object URL +// its redirects use. +func (c *Config) validateURLProxy() error { + if c.Storage.DirectServePublicURL != "" { + if err := validateHTTPURL("storage.direct_serve_public_url", c.Storage.DirectServePublicURL); err != nil { + return err + } + } + if c.URLProxy.FetchTimeout != "" { + if d, err := time.ParseDuration(c.URLProxy.FetchTimeout); err != nil || d <= 0 { + return fmt.Errorf("invalid url_proxy.fetch_timeout %q: must be a positive duration", c.URLProxy.FetchTimeout) + } + } + return nil +} + // Validate checks the configuration for errors. func (c *Config) Validate() error { if c.Listen == "" { @@ -1077,6 +1140,8 @@ func (c *Config) validateCacheArtifacts() error { return fmt.Errorf("storage.cache_artifacts: false cannot be combined with storage.direct_serve: no artifacts are stored to redirect to") case c.MirrorAPI: return fmt.Errorf("storage.cache_artifacts: false cannot be combined with mirror_api: mirrored artifacts would never be served") + case c.URLProxy.Enabled: + return fmt.Errorf("storage.cache_artifacts: false cannot be combined with url_proxy.enabled: /url/ serves only stored artifacts") } return nil } @@ -1086,6 +1151,10 @@ func (c *Config) validateComponents() error { return err } + if err := c.validateURLProxy(); err != nil { + return err + } + if err := c.Upstream.Validate(); err != nil { return err } @@ -1337,6 +1406,23 @@ func (c *Config) ParseGradleBuildCacheSweepInterval() time.Duration { return d } +// defaultURLProxyFetchTimeout stays under the 10 minute read timeout common +// on ingress controllers, so a slow fetch fails here with a clear error. +const defaultURLProxyFetchTimeout = 9 * time.Minute + +// ParseURLProxyFetchTimeout returns the /url/ upstream fetch timeout. +// Returns 9 minutes if unset or invalid. +func (c *Config) ParseURLProxyFetchTimeout() time.Duration { + if c.URLProxy.FetchTimeout == "" { + return defaultURLProxyFetchTimeout + } + d, err := time.ParseDuration(c.URLProxy.FetchTimeout) + if err != nil || d <= 0 { + return defaultURLProxyFetchTimeout + } + return d +} + // ParseDirectServeTTL returns the presigned URL expiry duration. // Returns 15 minutes if unset. func (c *Config) ParseDirectServeTTL() time.Duration { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 39b03364..7fedcc33 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1369,6 +1369,7 @@ func TestValidateCacheArtifactsDisabled(t *testing.T) { {"with scanning", func(c *Config) { c.Scanning.Enabled = true }, "scanning.enabled"}, {"with direct_serve", func(c *Config) { c.Storage.DirectServe = true }, "storage.direct_serve"}, {"with mirror_api", func(c *Config) { c.MirrorAPI = true }, "mirror_api"}, + {"with url_proxy", func(c *Config) { c.URLProxy.Enabled = true }, "url_proxy.enabled"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -1416,3 +1417,51 @@ func TestLoadCacheArtifactsFromEnv(t *testing.T) { t.Error("Storage.CacheArtifacts should be false") } } + +func TestValidateDirectServePublicURL(t *testing.T) { + for _, bad := range []string{"bucket/prefix", "s3://bucket", "http://rgw/bucket?x=1", "http://rgw/bucket#f"} { + cfg := Default() + cfg.Storage.DirectServePublicURL = bad + if err := cfg.Validate(); err == nil { + t.Errorf("direct_serve_public_url %q: expected validation error", bad) + } + } + cfg := Default() + cfg.Storage.DirectServePublicURL = "http://bucket.internal:7480/goproxy/pkgproxy" + if err := cfg.Validate(); err != nil { + t.Errorf("valid direct_serve_public_url: %v", err) + } +} + +func TestURLProxyFetchTimeout(t *testing.T) { + cfg := Default() + if got := cfg.ParseURLProxyFetchTimeout(); got != defaultURLProxyFetchTimeout { + t.Errorf("default = %v", got) + } + cfg.URLProxy.FetchTimeout = "20m" + if err := cfg.Validate(); err != nil { + t.Fatalf("valid fetch_timeout: %v", err) + } + if got := cfg.ParseURLProxyFetchTimeout(); got != 20*time.Minute { + t.Errorf("parsed = %v", got) + } + for _, bad := range []string{"soon", "0", "-1m"} { + cfg.URLProxy.FetchTimeout = bad + if err := cfg.Validate(); err == nil { + t.Errorf("fetch_timeout %q: expected validation error", bad) + } + } +} + +func TestLoadFromEnvURLProxy(t *testing.T) { + t.Setenv("PROXY_URL_PROXY_ENABLED", "true") + t.Setenv("PROXY_URL_PROXY_DIRECT_SERVE", "true") + t.Setenv("PROXY_URL_PROXY_FETCH_TIMEOUT", "3m") + t.Setenv("PROXY_STORAGE_DIRECT_SERVE_PUBLIC_URL", "http://rgw:7480/b/p") + cfg := Default() + cfg.LoadFromEnv() + if !cfg.URLProxy.Enabled || !cfg.URLProxy.DirectServe || cfg.URLProxy.FetchTimeout != "3m" || + cfg.Storage.DirectServePublicURL != "http://rgw:7480/b/p" { + t.Errorf("got url_proxy %+v, public url %q", cfg.URLProxy, cfg.Storage.DirectServePublicURL) + } +} diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 37e353ca..d142f78d 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -95,6 +95,8 @@ func packagePURLStrings(ecosystem, name, version string) (string, string, error) const contentTypeJSON = "application/json" +const contentTypeOctetStream = "application/octet-stream" + const ( headerAccept = "Accept" headerAcceptEncoding = "Accept-Encoding" @@ -172,8 +174,11 @@ type Proxy struct { // URLs so clients receive a public address even when the proxy reaches // storage at an internal one. DirectServeBaseURL string - HTTPClient *http.Client - AuthForURL func(string) (headerName, headerValue string) + // DirectServePublicURL, if set, is where the bucket serves objects + // anonymously; redirects point under it instead of at presigned URLs. + DirectServePublicURL string + HTTPClient *http.Client + AuthForURL func(string) (headerName, headerValue string) // StreamArtifacts streams artifacts from upstream without storing them. // Each request fetches its own copy: there is no cache to check and @@ -334,9 +339,9 @@ func (p *Proxy) checkCache(ctx context.Context, pkgPURL, versionPURL, filename s } if p.DirectServe { - signed, err := p.Storage.SignedURL(ctx, artifact.StoragePath, p.DirectServeTTL) + redirect, err := p.directServeURL(ctx, artifact.StoragePath) if err == nil { - result.RedirectURL = rewriteSignedURLHost(signed, p.DirectServeBaseURL) + result.RedirectURL = redirect p.recordCacheHit(artifact.Ecosystem, versionPURL, filename) return result, nil } @@ -375,6 +380,21 @@ func (p *Proxy) checkCache(ctx context.Context, pkgPURL, versionPURL, filename s return result, nil } +// directServeURL returns an address clients can download storagePath from +// directly: under DirectServePublicURL when the bucket serves it anonymously, +// otherwise a presigned URL. It returns storage.ErrSignedURLUnsupported when +// the backend can do neither. +func (p *Proxy) directServeURL(ctx context.Context, storagePath string) (string, error) { + if p.DirectServePublicURL != "" { + return storage.PublicObjectURL(p.DirectServePublicURL, storagePath), nil + } + signed, err := p.Storage.SignedURL(ctx, storagePath, p.DirectServeTTL) + if err != nil { + return "", err + } + return rewriteSignedURLHost(signed, p.DirectServeBaseURL), nil +} + // rewriteSignedURLHost replaces the scheme and host of a signed URL with those // from baseURL, preserving the path and query (which carry the signature). // Returns signed unchanged if baseURL is empty or either URL fails to parse. @@ -534,9 +554,10 @@ func (p *Proxy) openStoredArtifact(ctx context.Context, artifact artifacts.Artif } return &CacheResult{ - Reader: reader, - Artifact: artifact, - Cached: false, + Reader: reader, + Artifact: artifact, + Cached: false, + storagePath: storagePath, }, nil } diff --git a/internal/handler/urlproxy.go b/internal/handler/urlproxy.go new file mode 100644 index 00000000..12fedc27 --- /dev/null +++ b/internal/handler/urlproxy.go @@ -0,0 +1,308 @@ +package handler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "net" + "net/http" + "path" + "regexp" + "strings" + "time" + + "github.com/git-pkgs/proxy/internal/metrics" +) + +const ( + urlEcosystem = "url" + + // urlVersionHexLen is how much of the request path's sha256 identifies a + // URL within its host: 128 bits, far beyond any collision risk. + urlVersionHexLen = 32 + + // urlDigestSegment introduces the optional expected digest, as in + // /url/sha256/{hex}/{host}/{path}. + urlDigestSegment = "sha256" + + // urlDefaultFilename names an artifact whose URL path has no usable + // last segment. + urlDefaultFilename = "download" + + // urlWriteDeadlineSlack is added to the fetch timeout when extending the + // response write deadline, to leave time for storing and responding. + urlWriteDeadlineSlack = time.Minute +) + +var ( + // urlHostPattern accepts lowercase DNS names. Ports, userinfo and IPv6 + // literals are rejected: the route only fetches https on port 443. + urlHostPattern = regexp.MustCompile(`^[a-z0-9]([a-z0-9-]*[a-z0-9])?(\.[a-z0-9]([a-z0-9-]*[a-z0-9])?)*$`) + + urlDigestPattern = regexp.MustCompile(`^[0-9a-f]{64}$`) + + // urlFilenameUnsafe matches what is replaced in a cached filename, which + // becomes a storage key segment and part of a public object URL. + urlFilenameUnsafe = regexp.MustCompile(`[^A-Za-z0-9._+-]`) +) + +// URLHandler caches any public https URL as an immutable artifact. A request +// for /url/{host}/{path}?{query} fetches https://{host}/{path}?{query} once, +// stores it in the artifact cache, and serves every later request from there +// without revalidating. /url/sha256/{hex}/{host}/{path} also checks the bytes +// against that digest: a mismatching download is rejected rather than cached, +// and a cached copy with another digest is refetched. +// +// Unlike GenericHandler this is an open proxy for its clients; it should only +// be reachable by trusted networks. Its fetcher must refuse internal targets +// and must not carry configured upstream credentials. +// +// With DirectServe, GET requests are redirected to the stored object (see +// Proxy.directServeURL), so cache hits never stream through the proxy. +type URLHandler struct { + proxy *Proxy + directServe bool + fetchTimeout time.Duration + + // upstreamBase maps a host to the URL its paths are appended to. + // Tests point it at an httptest server. + upstreamBase func(host string) string +} + +// NewURLHandler creates a /url/ handler. proxy.Fetcher is used for upstream +// downloads; fetchTimeout extends the response write deadline to match it. +func NewURLHandler(proxy *Proxy, directServe bool, fetchTimeout time.Duration) *URLHandler { + return &URLHandler{ + proxy: proxy, + directServe: directServe, + fetchTimeout: fetchTimeout, + upstreamBase: func(host string) string { return "https://" + host }, + } +} + +// urlRequest is one parsed /url/ request. +type urlRequest struct { + host string + escapedPath string // with a leading slash, as received + query string + digest string // lowercase hex sha256, or empty + version string + filename string +} + +// parseURLRequest splits a /url/ request path (prefix already stripped) into +// its target and cache identity. +func parseURLRequest(r *http.Request) (urlRequest, bool) { + escaped := strings.TrimPrefix(r.URL.EscapedPath(), "/") + if containsPathTraversal(r.URL.Path) || strings.Contains(r.URL.Path, "\\") { + return urlRequest{}, false + } + + var req urlRequest + first, rest, _ := strings.Cut(escaped, "/") + if first == urlDigestSegment { + digest, after, _ := strings.Cut(rest, "/") + digest = strings.ToLower(digest) + if !urlDigestPattern.MatchString(digest) { + return urlRequest{}, false + } + req.digest = digest + escaped = after + } + + host, tail, ok := strings.Cut(escaped, "/") + host = strings.ToLower(host) + if !ok || tail == "" || !validURLHost(host) { + return urlRequest{}, false + } + req.host = host + req.escapedPath = "/" + tail + req.query = r.URL.RawQuery + + sum := sha256.Sum256([]byte(req.escapedPath + "?" + req.query)) + req.version = hex.EncodeToString(sum[:])[:urlVersionHexLen] + req.filename = urlFilename(r.URL.Path) + return req, true +} + +func validURLHost(host string) bool { + if !urlHostPattern.MatchString(host) { + return false + } + // A bare dotted quad also matches the DNS pattern; accept it only as a + // well-formed IPv4 address. safehttp still refuses internal ones. + if ip := net.ParseIP(host); ip != nil { + return ip.To4() != nil + } + return strings.Contains(host, ".") +} + +// urlFilename derives a storage-safe filename from the last path segment. +func urlFilename(decodedPath string) string { + if strings.HasSuffix(decodedPath, "/") { + return urlDefaultFilename + } + name := urlFilenameUnsafe.ReplaceAllString(path.Base(decodedPath), "_") + if name == "" || name == "." || name == ".." || strings.Trim(name, "_") == "" { + return urlDefaultFilename + } + return name +} + +func (req urlRequest) upstreamURL(base string) string { + u := base + req.escapedPath + if req.query != "" { + u += "?" + req.query + } + return u +} + +// Routes returns the HTTP handler for /url/ requests. Mount it at /url. +func (h *URLHandler) Routes() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet && r.Method != http.MethodHead { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + req, ok := parseURLRequest(r) + if !ok { + http.Error(w, "expected /url/[sha256/{hex}/]{host}/{path}", http.StatusBadRequest) + return + } + + // A cold fetch of a large file can outlast the server's write + // timeout; this route is bounded by its fetch timeout instead. + _ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(h.fetchTimeout + urlWriteDeadlineSlack)) + + downloadURL := req.upstreamURL(h.upstreamBase(req.host)) + h.proxy.Logger.Info("url download", + "host", req.host, "url", downloadURL, "sha256", req.digest) + + if r.Method == http.MethodGet && !h.directServe { + h.serveStreamed(w, r, req, downloadURL) + return + } + h.serveStored(w, r, req, downloadURL) + }) +} + +// serveStreamed streams the artifact through the proxy. The shared pipeline +// already refetches a record whose object is gone. +func (h *URLHandler) serveStreamed(w http.ResponseWriter, r *http.Request, req urlRequest, downloadURL string) { + result, err := h.proxy.getOrFetchArtifactFromURL( + r.Context(), urlEcosystem, req.host, req.version, req.filename, downloadURL, nil, req.digest) + if err != nil { + h.proxy.serveArtifactError(w, err, "failed to fetch url") + return + } + h.serve(w, r.Method, result) +} + +// serveStored answers HEAD, and GET when redirecting, from the cache record +// without reading the object: a redirect to it, or for HEAD just its headers. +// HEAD is never redirected because a presigned GET URL rejects HEAD. +func (h *URLHandler) serveStored(w http.ResponseWriter, r *http.Request, req urlRequest, downloadURL string) { + ctx := r.Context() + p := h.proxy + result, err := p.urlStoredArtifact(ctx, req, downloadURL) + if err != nil { + p.serveArtifactError(w, err, "failed to fetch url") + return + } + if result.Reader != nil { + _ = result.Reader.Close() + result.Reader = nil + } + + if r.Method == http.MethodGet { + redirect, err := p.directServeURL(ctx, result.storagePath) + if err != nil { + p.Logger.Warn("cannot redirect to stored url artifact, streaming it", + "path", result.storagePath, "error", err) + streamed, err := p.openStoredArtifact(ctx, result.Artifact, result.storagePath) + if err != nil { + p.serveArtifactError(w, err, "failed to read cached url") + return + } + result = streamed + } else { + result.RedirectURL = redirect + } + } + h.serve(w, r.Method, result) +} + +func (h *URLHandler) serve(w http.ResponseWriter, method string, result *CacheResult) { + if result.Artifact.MediaType == "" { + result.Artifact.MediaType = contentTypeOctetStream + } + serveArtifact(w, method, result) +} + +// urlStoredArtifact returns a cache record for req whose object is known to +// exist, fetching the artifact first on a miss. The result's Reader, if any, +// is for the caller to close. +// +// A record whose object is gone (expired by a bucket lifecycle rule, say) is +// a miss. The shared fetch then finds it cannot open the stored object and +// refetches, replacing the record; if that fetch fails too the error is +// returned, so there is never more than one refetch per request. +func (p *Proxy) urlStoredArtifact(ctx context.Context, req urlRequest, downloadURL string) (*CacheResult, error) { + if p.versionDenied(urlEcosystem, req.host, req.version) { + return nil, fmt.Errorf("%w: %s", ErrVersionDenied, canonicalVersionPURL(urlEcosystem, req.host, req.version)) + } + pkgPURL, versionPURL, err := packagePURLStrings(urlEcosystem, req.host, req.version) + if err != nil { + return nil, err + } + + if cached := p.urlCachedRecord(ctx, pkgPURL, versionPURL, req); cached != nil { + return cached, nil + } + metrics.RecordCacheMiss(urlEcosystem) + return p.coalescedFetchFromURL(ctx, urlEcosystem, req.host, req.version, req.filename, + pkgPURL, versionPURL, downloadURL, nil, req.digest) +} + +// urlCachedRecord is checkCache for callers that will not read the object: +// it confirms the object exists instead of opening it. +func (p *Proxy) urlCachedRecord(ctx context.Context, pkgPURL, versionPURL string, req urlRequest) *CacheResult { + record, err := p.DB.GetCachedArtifact(pkgPURL, versionPURL, req.filename) + if err != nil { + p.Logger.Warn("failed to read url cache record", "purl", versionPURL, "error", err) + return nil + } + if record == nil { + return nil + } + if _, err := newIntegrityChecks(record.Artifact.Digest.Encoded(), record.Integrity.String); err != nil { + p.rejectUnusableCacheRecord(record, versionPURL, req.filename, err) + return nil + } + if !artifactHashMatches(record.Artifact.Digest.Encoded(), req.digest) { + // Left in place; the fetch replacing it discards it under the key. + return nil + } + + exists, err := p.Storage.Exists(ctx, record.StoragePath) + if err != nil { + metrics.RecordStorageError("exists") + p.Logger.Warn("failed to check stored url artifact, will refetch", + "path", record.StoragePath, "error", err) + return nil + } + if !exists { + metrics.RecordMissingObject(urlEcosystem) + p.Logger.Warn("cached url artifact missing from storage, will refetch", + "purl", versionPURL, "path", record.StoragePath) + return nil + } + + p.recordCacheHit(urlEcosystem, versionPURL, req.filename) + return &CacheResult{ + Artifact: record.Artifact, + Cached: true, + storagePath: record.StoragePath, + } +} diff --git a/internal/handler/urlproxy_test.go b/internal/handler/urlproxy_test.go new file mode 100644 index 00000000..4fddd741 --- /dev/null +++ b/internal/handler/urlproxy_test.go @@ -0,0 +1,401 @@ +package handler + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "io" + "net/http" + "net/http/httptest" + "path/filepath" + "strconv" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/storage" + "github.com/git-pkgs/registries/fetch" +) + +const ( + testURLHost = "downloads.example.org" + testURLPath = "/releases/tool-1.0.0.tar.gz" + testURLRoute = "/" + testURLHost + testURLPath + testPublic = "http://rgw.example:7480/bucket/prefix" +) + +// urlUpstream is a fake https origin serving body at testURLPath. +type urlUpstream struct { + *httptest.Server + body atomic.Value // []byte + available atomic.Bool + requests atomic.Int32 +} + +func newURLUpstream(t *testing.T, body []byte) *urlUpstream { + t.Helper() + u := &urlUpstream{} + u.body.Store(body) + u.available.Store(true) + u.Server = httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !u.available.Load() { + http.Error(w, "unavailable", http.StatusServiceUnavailable) + return + } + if r.URL.Path != testURLPath { + http.NotFound(w, r) + return + } + u.requests.Add(1) + _, _ = w.Write(u.body.Load().([]byte)) + })) + t.Cleanup(u.Close) + return u +} + +func newTestURLHandler(t *testing.T, upstream *urlUpstream, directServe bool) (*URLHandler, *Proxy, *mockStorage) { + t.Helper() + proxy, _, store, _ := setupTestProxy(t) + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(upstream.Client()), fetch.WithMaxRetries(0)) + t.Cleanup(func() { _ = fetcher.Close() }) + proxy.Fetcher = fetcher + h := NewURLHandler(proxy, directServe, time.Minute) + h.upstreamBase = func(string) string { return upstream.URL } + return h, proxy, store +} + +func serveURLRequest(h *URLHandler, method, target string) *httptest.ResponseRecorder { + w := httptest.NewRecorder() + h.Routes().ServeHTTP(w, httptest.NewRequest(method, target, nil)) + return w +} + +func sha256Of(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +func TestParseURLRequest(t *testing.T) { + digest := strings.Repeat("ab", 32) + tests := []struct { + target string + ok bool + host string + path string + query string + digest string + filename string + }{ + {"/www.boost.org/releases/boost-1.83.0.tar.gz", true, "www.boost.org", "/releases/boost-1.83.0.tar.gz", "", "", "boost-1.83.0.tar.gz"}, + {"/WWW.Boost.ORG/x.tgz", true, "www.boost.org", "/x.tgz", "", "", "x.tgz"}, + {"/api.github.com/repos/o/r/tarball/v1?a=1&b=2", true, "api.github.com", "/repos/o/r/tarball/v1", "a=1&b=2", "", "v1"}, + {"/sha256/" + strings.ToUpper(digest) + "/ftp.gnu.org/gnu/gawk/gawk-5.3.1.tar.xz", true, "ftp.gnu.org", "/gnu/gawk/gawk-5.3.1.tar.xz", "", digest, "gawk-5.3.1.tar.xz"}, + {"/example.org/dir/", true, "example.org", "/dir/", "", "", "download"}, + {"/example.org/a%20b%2Bc.tar.gz", true, "example.org", "/a%20b%2Bc.tar.gz", "", "", "a_b+c.tar.gz"}, + {"/1.2.3.4/file", true, "1.2.3.4", "/file", "", "", "file"}, + {"/sha256/abc/example.org/file", false, "", "", "", "", ""}, + {"/example.org", false, "", "", "", "", ""}, + {"/example.org/", false, "", "", "", "", ""}, + {"/localhost/file", false, "", "", "", "", ""}, + {"/example.org:8443/file", false, "", "", "", "", ""}, + {"/user@example.org/file", false, "", "", "", "", ""}, + {"/example.org/a/../../etc/passwd", false, "", "", "", "", ""}, + {"/-bad.example.org/file", false, "", "", "", "", ""}, + } + for _, tt := range tests { + t.Run(tt.target, func(t *testing.T) { + req, ok := parseURLRequest(httptest.NewRequest(http.MethodGet, tt.target, nil)) + if ok != tt.ok { + t.Fatalf("ok = %v, want %v (%+v)", ok, tt.ok, req) + } + if !ok { + return + } + if req.host != tt.host || req.escapedPath != tt.path || req.query != tt.query || + req.digest != tt.digest || req.filename != tt.filename { + t.Errorf("got %+v", req) + } + if len(req.version) != urlVersionHexLen { + t.Errorf("version = %q", req.version) + } + }) + } +} + +func TestParseURLRequest_VersionDependsOnPathAndQuery(t *testing.T) { + parse := func(target string) string { + req, ok := parseURLRequest(httptest.NewRequest(http.MethodGet, target, nil)) + if !ok { + t.Fatalf("parse %s failed", target) + } + return req.version + } + a := parse("/example.org/f.tgz") + if a != parse("/sha256/"+strings.Repeat("0", 64)+"/example.org/f.tgz") { + t.Error("digest changed the version") + } + if a == parse("/example.org/f.tgz?x=1") || a == parse("/example.org/g/f.tgz") { + t.Error("distinct URLs share a version") + } +} + +func TestURLHandler_RejectsMethodsAndBadPaths(t *testing.T) { + h := NewURLHandler(testProxy(), false, time.Minute) + if w := serveURLRequest(h, http.MethodPost, testURLRoute); w.Code != http.StatusMethodNotAllowed { + t.Errorf("POST: status = %d", w.Code) + } + if w := serveURLRequest(h, http.MethodGet, "/localhost/x"); w.Code != http.StatusBadRequest { + t.Errorf("bad host: status = %d", w.Code) + } +} + +func TestURLHandler_StreamsAndCaches(t *testing.T) { + body := []byte("tarball bytes") + upstream := newURLUpstream(t, body) + h, _, _ := newTestURLHandler(t, upstream, false) + + for i := range 2 { + w := serveURLRequest(h, http.MethodGet, testURLRoute) + if w.Code != http.StatusOK || w.Body.String() != string(body) { + t.Fatalf("request %d: status %d body %q", i, w.Code, w.Body.String()) + } + if got := w.Header().Get("Location"); got != "" { + t.Errorf("request %d: unexpected redirect %s", i, got) + } + } + if got := upstream.requests.Load(); got != 1 { + t.Errorf("upstream requests = %d, want 1", got) + } + + // Immutable: served from cache while the origin is down. + upstream.available.Store(false) + if w := serveURLRequest(h, http.MethodGet, testURLRoute); w.Code != http.StatusOK { + t.Errorf("origin down: status = %d", w.Code) + } +} + +func TestURLHandler_HeadAnswersFromRecord(t *testing.T) { + body := []byte("head me") + upstream := newURLUpstream(t, body) + h, proxy, _ := newTestURLHandler(t, upstream, true) + proxy.DirectServePublicURL = testPublic + + w := serveURLRequest(h, http.MethodHead, testURLRoute) + if w.Code != http.StatusOK { + t.Fatalf("status = %d", w.Code) + } + if w.Body.Len() != 0 || w.Header().Get("Location") != "" { + t.Errorf("HEAD got body %d bytes, Location %q", w.Body.Len(), w.Header().Get("Location")) + } + if got := w.Header().Get("Content-Length"); got != strconv.Itoa(len(body)) { + t.Errorf("Content-Length = %q", got) + } + if got := w.Header().Get("ETag"); got != `"`+sha256Of(body)+`"` { + t.Errorf("ETag = %q", got) + } +} + +func TestURLHandler_Digest(t *testing.T) { + body := []byte("pinned bytes") + upstream := newURLUpstream(t, body) + h, proxy, _ := newTestURLHandler(t, upstream, false) + + bad := "/sha256/" + strings.Repeat("0", 64) + testURLRoute + if w := serveURLRequest(h, http.MethodGet, bad); w.Code != http.StatusBadGateway { + t.Fatalf("mismatch: status = %d, want 502", w.Code) + } + if cached, _ := proxy.GetCachedArtifact(context.Background(), urlEcosystem, testURLHost, + mustParse(t, testURLRoute).version, "tool-1.0.0.tar.gz"); cached != nil { + t.Fatal("mismatching download was cached") + } + + good := "/sha256/" + sha256Of(body) + testURLRoute + if w := serveURLRequest(h, http.MethodGet, good); w.Code != http.StatusOK || w.Body.String() != string(body) { + t.Fatalf("match: status %d body %q", w.Code, w.Body.String()) + } + + // The origin republished; a client pinning the new digest refetches. + newBody := []byte("republished bytes") + upstream.body.Store(newBody) + before := upstream.requests.Load() + w := serveURLRequest(h, http.MethodGet, "/sha256/"+sha256Of(newBody)+testURLRoute) + if w.Code != http.StatusOK || w.Body.String() != string(newBody) { + t.Fatalf("new digest: status %d body %q", w.Code, w.Body.String()) + } + if upstream.requests.Load() != before+1 { + t.Error("new digest did not refetch") + } +} + +func mustParse(t *testing.T, target string) urlRequest { + t.Helper() + req, ok := parseURLRequest(httptest.NewRequest(http.MethodGet, target, nil)) + if !ok { + t.Fatalf("parse %s failed", target) + } + return req +} + +func TestURLHandler_RedirectsToPublicURL(t *testing.T) { + body := []byte("redirect me") + upstream := newURLUpstream(t, body) + h, proxy, store := newTestURLHandler(t, upstream, true) + proxy.DirectServePublicURL = testPublic + "/" + + req := mustParse(t, testURLRoute) + wantPrefix := testPublic + "/url/" + testURLHost + "/" + req.version + "/" + + var first string + for i := range 2 { + w := serveURLRequest(h, http.MethodGet, testURLRoute) + if w.Code != http.StatusFound { + t.Fatalf("request %d: status = %d: %s", i, w.Code, w.Body.String()) + } + loc := w.Header().Get("Location") + if !strings.HasPrefix(loc, wantPrefix) || !strings.HasSuffix(loc, "/tool-1.0.0.tar.gz") { + t.Fatalf("request %d: Location = %s", i, loc) + } + key := strings.TrimPrefix(loc, testPublic+"/") + if _, ok := store.files[key]; !ok { + t.Fatalf("request %d: no stored object at %s", i, key) + } + if i == 0 { + first = loc + } else if loc != first { + t.Errorf("hit redirected to %s, miss to %s", loc, first) + } + } + if got := upstream.requests.Load(); got != 1 { + t.Errorf("upstream requests = %d, want 1", got) + } +} + +func TestURLHandler_RedirectsToSignedURL(t *testing.T) { + upstream := newURLUpstream(t, []byte("signed")) + h, _, store := newTestURLHandler(t, upstream, true) + store.signedURL = "https://bucket.s3.amazonaws.com/obj?X-Amz-Signature=abc" + + w := serveURLRequest(h, http.MethodGet, testURLRoute) + if w.Code != http.StatusFound || w.Header().Get("Location") != store.signedURL { + t.Fatalf("status %d Location %q", w.Code, w.Header().Get("Location")) + } +} + +func TestURLHandler_StreamsWhenRedirectUnsupported(t *testing.T) { + body := []byte("no signing") + upstream := newURLUpstream(t, body) + h, _, _ := newTestURLHandler(t, upstream, true) + + for i := range 2 { + w := serveURLRequest(h, http.MethodGet, testURLRoute) + if w.Code != http.StatusOK || w.Body.String() != string(body) { + t.Fatalf("request %d: status %d body %q", i, w.Code, w.Body.String()) + } + } +} + +func TestURLHandler_RefetchesMissingObject(t *testing.T) { + for _, directServe := range []bool{true, false} { + t.Run("direct_serve="+strconv.FormatBool(directServe), func(t *testing.T) { + body := []byte("lifecycle expired me") + upstream := newURLUpstream(t, body) + h, proxy, store := newTestURLHandler(t, upstream, directServe) + proxy.DirectServePublicURL = testPublic + + if w := serveURLRequest(h, http.MethodGet, testURLRoute); w.Code >= 400 { + t.Fatalf("first: status = %d", w.Code) + } + clear(store.files) // as a bucket lifecycle rule would + + w := serveURLRequest(h, http.MethodGet, testURLRoute) + switch { + case directServe && w.Code != http.StatusFound: + t.Fatalf("after expiry: status = %d, want 302", w.Code) + case !directServe && (w.Code != http.StatusOK || w.Body.String() != string(body)): + t.Fatalf("after expiry: status %d body %q", w.Code, w.Body.String()) + } + if got := upstream.requests.Load(); got != 2 { + t.Errorf("upstream requests = %d, want 2", got) + } + if len(store.files) != 1 { + t.Errorf("stored objects = %d, want 1", len(store.files)) + } + if directServe { + key := strings.TrimPrefix(w.Header().Get("Location"), testPublic+"/") + if _, ok := store.files[key]; !ok { + t.Errorf("redirected to %s, which is not stored", key) + } + } + + // Gone again with the origin down: one failed refetch, no loop. + clear(store.files) + upstream.available.Store(false) + if w := serveURLRequest(h, http.MethodGet, testURLRoute); w.Code != http.StatusBadGateway { + t.Errorf("expired and origin down: status = %d, want 502", w.Code) + } + if got := upstream.requests.Load(); got != 2 { + t.Errorf("upstream requests = %d, want 2 (unavailable requests are not counted)", got) + } + }) + } +} + +// zeroReader yields n zero bytes without holding them in memory. +type zeroReader struct{ n int64 } + +func (z *zeroReader) Read(p []byte) (int, error) { + if z.n <= 0 { + return 0, io.EOF + } + if int64(len(p)) > z.n { + p = p[:z.n] + } + clear(p) + z.n -= int64(len(p)) + return len(p), nil +} + +func TestURLHandler_LargeFileIsNotCapped(t *testing.T) { + if testing.Short() { + t.Skip("streams 120MB") + } + const size = 120 << 20 // over the 100MB metadata buffer cap + upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Length", strconv.Itoa(size)) + _, _ = io.Copy(w, &zeroReader{n: size}) + })) + t.Cleanup(upstream.Close) + + ctx := context.Background() + store, err := storage.OpenBucket(ctx, "file://"+filepath.Join(t.TempDir(), "cache")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + proxy, _, _, _ := setupTestProxy(t) + proxy.Storage = store + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(upstream.Client()), fetch.WithMaxRetries(0)) + t.Cleanup(func() { _ = fetcher.Close() }) + proxy.Fetcher = fetcher + h := NewURLHandler(proxy, false, time.Minute) + h.upstreamBase = func(string) string { return upstream.URL } + + srv := httptest.NewServer(h.Routes()) + t.Cleanup(srv.Close) + resp, err := http.Get(srv.URL + "/big.example.org/ghc.tar.xz") + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + hash := sha256.New() + n, err := io.Copy(hash, resp.Body) + if err != nil || resp.StatusCode != http.StatusOK || n != size { + t.Fatalf("status %d, read %d bytes, err %v", resp.StatusCode, n, err) + } + want := sha256.New() + _, _ = io.Copy(want, &zeroReader{n: size}) + if hex.EncodeToString(hash.Sum(nil)) != hex.EncodeToString(want.Sum(nil)) { + t.Error("body hash mismatch") + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 4021099d..0647da99 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -131,6 +131,14 @@ var ( []string{"ecosystem"}, ) + MissingObjects = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "proxy_cache_missing_objects_total", + Help: "Cache records whose stored object was gone, so the artifact was refetched", + }, + []string{"ecosystem"}, + ) + HealthProbeFailures = prometheus.NewCounterVec( prometheus.CounterOpts{ Name: "proxy_health_probe_failures_total", @@ -242,6 +250,7 @@ func init() { StorageErrors, ActiveRequests, IntegrityFailures, + MissingObjects, HealthProbeFailures, EcosystemDownloadedBytes, EcosystemDownloads, @@ -311,6 +320,11 @@ func RecordIntegrityFailure(ecosystem string) { IntegrityFailures.WithLabelValues(ecosystem).Inc() } +// RecordMissingObject counts a cache record found without its stored object. +func RecordMissingObject(ecosystem string) { + MissingObjects.WithLabelValues(ecosystem).Inc() +} + // RecordHealthProbeFailure increments the health probe failure counter. // step is one of: "write", "size", "read", "verify", "delete". func RecordHealthProbeFailure(step string) { diff --git a/internal/server/dashboard.go b/internal/server/dashboard.go index a3ac9ebd..b7acb17f 100644 --- a/internal/server/dashboard.go +++ b/internal/server/dashboard.go @@ -143,6 +143,7 @@ func supportedEcosystems() []string { "pypi", "rpm", "swift", + "url", } } diff --git a/internal/server/middleware.go b/internal/server/middleware.go index 58eeeb76..f7b1ab3d 100644 --- a/internal/server/middleware.go +++ b/internal/server/middleware.go @@ -96,7 +96,7 @@ func requestEcosystem(path string) string { switch segment { case "npm", "cargo", "hex", "pub", "pypi", "maven", "gradle", "nuget", "conan", "conda", "cran", "julia", "debian", "rpm", - "helm", "homebrew", "generic", "swift": + "helm", "homebrew", "generic", "swift", "url": return segment case "apk": return "alpine" diff --git a/internal/server/server.go b/internal/server/server.go index 1066d6d6..4d248d8b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -261,6 +261,7 @@ func (s *Server) serve(listener net.Listener) error { proxy.DirectServeTTL = s.cfg.ParseDirectServeTTL() proxy.DirectServeBaseURL = s.cfg.Storage.DirectServeBaseURL proxy.StreamArtifacts = !s.cfg.Storage.CacheArtifacts + proxy.DirectServePublicURL = s.cfg.Storage.DirectServePublicURL // Create router with Chi r := chi.NewRouter() @@ -463,6 +464,38 @@ func (s *Server) mountProtocolHandlers(r chi.Router, proxy *handler.Proxy) { r.Mount("/debian", http.StripPrefix("/debian", debianHandler.Routes())) r.Mount("/rpm", http.StripPrefix("/rpm", rpmHandler.Routes())) r.Mount("/generic", http.StripPrefix("/generic", genericHandler.Routes())) + + if s.cfg.URLProxy.Enabled { + urlHandler := handler.NewURLHandler(s.newURLProxy(proxy), s.cfg.URLProxy.DirectServe, s.cfg.ParseURLProxyFetchTimeout()) + r.Mount("/url", http.StripPrefix("/url", urlHandler.Routes())) + } +} + +// newURLProxy returns a Proxy sharing proxy's cache and policies, but +// fetching through a client fit for arbitrary hosts: the same safehttp dial +// gate (no loopback, private or link-local targets, checked on every redirect +// hop), no configured upstream credentials or OCI token exchange, no +// environment HTTP proxy that would bypass the gate, and the /url/ fetch +// timeout. +func (s *Server) newURLProxy(proxy *handler.Proxy) *handler.Proxy { + client := newUpstreamClient(s.cfg.Upstream) + if transport, ok := client.Transport.(*http.Transport); ok { + transport.Proxy = nil + } + if s.accessLog != nil { + client.Transport = upstreamhttp.NewAccessLogTransport(client.Transport, s.accessLog, s.logger) + } + client.Timeout = s.cfg.ParseURLProxyFetchTimeout() + + urlProxy := handler.NewProxy(s.db, s.storage, fetch.NewFetcher(fetch.WithHTTPClient(client)), proxy.Resolver, s.logger) + urlProxy.Denylist = proxy.Denylist + urlProxy.Scanners = proxy.Scanners + urlProxy.ScanSigningKey = proxy.ScanSigningKey + urlProxy.ScanFetchBaseURL = proxy.ScanFetchBaseURL + urlProxy.DirectServeTTL = proxy.DirectServeTTL + urlProxy.DirectServeBaseURL = proxy.DirectServeBaseURL + urlProxy.DirectServePublicURL = proxy.DirectServePublicURL + return urlProxy } // configureScanning builds the scanner group from cfg and wires it into diff --git a/internal/server/urlproxy_test.go b/internal/server/urlproxy_test.go new file mode 100644 index 00000000..21217acf --- /dev/null +++ b/internal/server/urlproxy_test.go @@ -0,0 +1,67 @@ +package server + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + + "github.com/git-pkgs/proxy/internal/config" + "github.com/git-pkgs/proxy/internal/handler" +) + +func newURLProxyForTest(t *testing.T, upstream config.UpstreamConfig) *handler.Proxy { + t.Helper() + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + s := &Server{cfg: &config.Config{Upstream: upstream}, logger: logger} + p := s.newURLProxy(handler.NewProxy(nil, nil, nil, nil, logger)) + t.Cleanup(func() { + if c, ok := p.Fetcher.(io.Closer); ok { + _ = c.Close() + } + }) + return p +} + +func TestURLProxyRefusesLoopbackTargets(t *testing.T) { + var hits int + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits++ + _, _ = io.WriteString(w, "internal") + })) + defer srv.Close() + + p := newURLProxyForTest(t, config.UpstreamConfig{}) + _, err := p.Fetcher.Fetch(context.Background(), srv.URL+"/secret") + if err == nil { + t.Fatal("fetching a loopback target succeeded") + } + t.Logf("refused: %v", err) + if hits != 0 { + t.Errorf("loopback target received %d requests", hits) + } +} + +func TestURLProxySendsNoConfiguredCredentials(t *testing.T) { + var gotAuth string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + _, _ = io.WriteString(w, "ok") + })) + defer srv.Close() + + p := newURLProxyForTest(t, config.UpstreamConfig{ + AllowLoopback: true, + Auth: map[string]config.AuthConfig{srv.URL: {Type: "bearer", Token: "s3cret"}}, + }) + artifact, err := p.Fetcher.Fetch(context.Background(), srv.URL+"/file") + if err != nil { + t.Fatalf("fetch: %v", err) + } + _ = artifact.Body.Close() + if gotAuth != "" { + t.Errorf("upstream received Authorization %q", gotAuth) + } +} diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 0f64ed77..b386ed6b 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -18,6 +18,8 @@ import ( "encoding/hex" "errors" "io" + "net/url" + "strings" "time" ) @@ -118,3 +120,16 @@ func isFetchDir(name string) bool { } return true } + +// PublicObjectURL returns the address of storagePath under baseURL, a bucket +// location that serves objects anonymously, such as +// "http://rgw:7480/bucket/prefix". baseURL must include any key prefix the +// storage URL adds, since storage paths never carry it. Each path segment is +// escaped on its own so slashes keep separating keys. +func PublicObjectURL(baseURL, storagePath string) string { + segments := strings.Split(storagePath, "/") + for i, segment := range segments { + segments[i] = url.PathEscape(segment) + } + return strings.TrimSuffix(baseURL, "/") + "/" + strings.Join(segments, "/") +} diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go index 67745d0c..ba85e44b 100644 --- a/internal/storage/storage_test.go +++ b/internal/storage/storage_test.go @@ -76,3 +76,17 @@ func assertLargeFileRoundTrip(t *testing.T, s Storage) { t.Error("large file content mismatch") } } + +func TestPublicObjectURL(t *testing.T) { + tests := []struct{ base, path, want string }{ + {"http://rgw:7480/bucket/prefix", "url/example.org/abc/0123/f.tar.gz", "http://rgw:7480/bucket/prefix/url/example.org/abc/0123/f.tar.gz"}, + {"http://rgw:7480/bucket/prefix/", "url/h/v/id/a+b.tgz", "http://rgw:7480/bucket/prefix/url/h/v/id/a+b.tgz"}, + {"https://cdn.example", "npm/@scope/pkg/1.0.0/id/pkg 1.tgz", "https://cdn.example/npm/@scope/pkg/1.0.0/id/pkg%201.tgz"}, + {"https://cdn.example", "x/a?b#c", "https://cdn.example/x/a%3Fb%23c"}, + } + for _, tt := range tests { + if got := PublicObjectURL(tt.base, tt.path); got != tt.want { + t.Errorf("PublicObjectURL(%q, %q) = %q, want %q", tt.base, tt.path, got, tt.want) + } + } +}