package handler import ( "bytes" "context" "fmt" "io" "net/http" "strconv" "strings" "time" "github.com/git-pkgs/proxy/internal/metrics" ) // manifestRequest is one client manifest request as the upstream sees it. type manifestRequest struct { registryURL string name string reference string accept string cacheAccept string method string } func (req manifestRequest) immutable() bool { return manifestDigestReferencePattern.MatchString(req.reference) } // coalesceKey identifies requests that can share one upstream exchange. The // normalized Accept is part of it because registries answer a tag with an // index or a single manifest depending on it, and the method because a HEAD // never fetches a manifest body the client did not ask for. func (req manifestRequest) coalesceKey() string { return strings.Join([]string{containerManifestCacheEcosystem, req.method, req.registryURL, req.name, req.reference, req.cacheAccept}, "\x00") } func (req manifestRequest) url(reference string) string { return fmt.Sprintf("%s/v2/%s/manifests/%s", req.registryURL, req.name, reference) } // manifestOutcome is what callers sharing a manifest request write. Exactly // one of manifest, relay and digestInvalid is set; otherwise err is the // upstream failure, answered from the caller's stale copy if it has one. type manifestOutcome struct { manifest *cachedContainerManifest relay *capturedResponse digestInvalid bool err error } // capturedResponse is an upstream response other than a manifest, such as a // 404 or 401, read in full so every caller sharing the request can relay it. // A body that could not be read to the end is relayed up to that point and // then fails the same way, so clients still see the truncation. type capturedResponse struct { status int header http.Header body []byte readErr error trailer http.Header } func (h *ContainerHandler) captureResponse(resp *http.Response) *capturedResponse { captured := &capturedResponse{status: resp.StatusCode, header: resp.Header.Clone()} if resp.Request != nil && resp.Request.Method == http.MethodHead { return captured } limit := h.proxy.MetadataMaxSize if limit <= 0 { limit = defaultMetadataMaxSize } captured.body, captured.readErr = io.ReadAll(io.LimitReader(resp.Body, limit+1)) if captured.readErr == nil && int64(len(captured.body)) > limit { captured.body, captured.readErr = captured.body[:limit], ErrMetadataTooLarge } captured.trailer = resp.Trailer.Clone() return captured } type failingReader struct{ err error } func (f failingReader) Read([]byte) (int, error) { return 0, f.err } func (c *capturedResponse) write(p *Proxy, w http.ResponseWriter, r *http.Request) { var body io.Reader = bytes.NewReader(c.body) if c.readErr != nil { body = io.MultiReader(body, failingReader{c.readErr}) } resp := &http.Response{ StatusCode: c.status, Header: c.header.Clone(), Trailer: c.trailer.Clone(), Body: io.NopCloser(body), } p.relayResponse(w, r, resp, copyContainerManifestHeaders) } // coalescedManifest runs the upstream part of a manifest request once for // concurrent callers with the same key. Like coalesceMetadata, it runs on a // context detached from the first caller, and each waiter still leaves when // its own context ends. The result is a whole manifest with its headers, // which a metadataResult does not carry, hence the separate group. func (h *ContainerHandler) coalescedManifest(ctx context.Context, req manifestRequest) manifestOutcome { results := h.manifestFlight.DoChan(req.coalesceKey(), func() (any, error) { return h.fetchManifest(context.WithoutCancel(ctx), req), nil }) select { case <-ctx.Done(): return manifestOutcome{err: ctx.Err()} case res := <-results: outcome, ok := res.Val.(manifestOutcome) if !ok { return manifestOutcome{err: errSharedFetchAbandoned} } return outcome } } // fetchManifest answers a manifest request that the cache could not. Other // instances sharing the database take turns through a fetch lease, so a tag // that expired everywhere at once is revalidated once. func (h *ContainerHandler) fetchManifest(ctx context.Context, req manifestRequest) (outcome manifestOutcome) { usable := func(m *cachedContainerManifest) bool { return m != nil && (req.immutable() || h.containerManifestFresh(m)) } // A shared request that finished since the caller looked has refreshed // the entry already. cached, fromFallback := h.loadContainerManifestForAccept(ctx, req.registryURL, req.name, req.reference, req.accept, req.cacheAccept) if usable(cached) { return manifestOutcome{manifest: cached} } // Without a TTL nothing another instance stores is fresh, so there is // nothing to wait for. if req.immutable() || h.proxy.MetadataTTL > 0 { var stored *cachedContainerManifest release, lease, err := h.proxy.awaitFetchLease(ctx, metrics.CacheKindMetadata, req.coalesceKey(), func() bool { stored, _ = h.loadContainerManifestForAccept(ctx, req.registryURL, req.name, req.reference, req.accept, req.cacheAccept) return usable(stored) }) if err != nil { return manifestOutcome{err: err} } switch lease { case leaseStored: return manifestOutcome{manifest: stored} case leaseFailed: if cached != nil { return manifestOutcome{err: errFetchFailedElsewhere} } case leaseFetch: } defer func() { release(outcome.err != nil) }() } switch { case req.method == http.MethodHead: return h.headManifest(ctx, req, cached, fromFallback) case req.immutable(): return h.getManifest(ctx, req, req.reference, cached, fromFallback) default: return h.revalidateTag(ctx, req, cached, fromFallback) } } // revalidateTag answers a GET for a tag that is stale or not cached. Docker // Hub counts manifest GETs against its pull limit but not HEADs, so a stale // tag is resolved with a HEAD first and the manifest fetched by digest only // when it is not cached yet. A registry whose HEAD gives no digest gets the // GET for the tag, as before. A tag never seen is fetched with one GET: a // HEAD first would only add a round trip. func (h *ContainerHandler) revalidateTag(ctx context.Context, req manifestRequest, cached *cachedContainerManifest, fromFallback bool) manifestOutcome { if cached == nil { return h.getManifest(ctx, req, req.reference, nil, false) } resp, err := h.requestManifest(ctx, http.MethodHead, req, req.reference, "") if err != nil { return manifestOutcome{err: err} } _ = resp.Body.Close() digest := resp.Header.Get("Docker-Content-Digest") switch { case resp.StatusCode == http.StatusOK && manifestDigestReferencePattern.MatchString(digest): if outcome, ok := h.tagFromCachedDigest(ctx, req, cached, fromFallback, resp, digest); ok { return outcome } if outcome := h.getManifest(ctx, req, digest, nil, false); outcome.manifest != nil { return outcome } // The tag may have moved again, or the registry may not serve // manifests by digest to this client; ask for the tag itself. case shouldServeStaleManifest(resp.StatusCode): return manifestOutcome{err: fmt.Errorf("upstream status %d", resp.StatusCode)} } return h.getManifest(ctx, req, req.reference, cached, fromFallback) } // headManifest answers a client HEAD. When the digest upstream names is // cached, the tag entry is pointed at it, so the next GET is a cache hit. // Otherwise the upstream headers are passed on and nothing is stored. func (h *ContainerHandler) headManifest(ctx context.Context, req manifestRequest, cached *cachedContainerManifest, fromFallback bool) manifestOutcome { resp, err := h.requestManifest(ctx, http.MethodHead, req, req.reference, "") if err != nil { return manifestOutcome{err: err} } _ = resp.Body.Close() digest := resp.Header.Get("Docker-Content-Digest") switch { case resp.StatusCode == http.StatusOK && manifestDigestReferencePattern.MatchString(digest): if req.immutable() && digestMismatch(req.reference, digest) { h.proxy.Logger.Error("upstream manifest HEAD names another digest", "name", req.name, "reference", req.reference, "upstream", digest) return manifestOutcome{digestInvalid: true} } if !req.immutable() { if outcome, ok := h.tagFromCachedDigest(ctx, req, cached, fromFallback, resp, digest); ok { return outcome } } size := int64(-1) if value := resp.Header.Get(headerContentLength); value != "" { if parsed, err := strconv.ParseInt(value, 10, 64); err == nil { size = parsed } } return manifestOutcome{manifest: &cachedContainerManifest{ contentType: resp.Header.Get(headerContentType), contentDigest: digest, etag: resp.Header.Get(headerETag), size: size, lastModified: parseHTTPTime(resp.Header.Get(headerLastModified)), }} case shouldServeStaleManifest(resp.StatusCode) && (cached != nil || referrersTagPattern.MatchString(req.reference)): return manifestOutcome{err: fmt.Errorf("upstream status %d", resp.StatusCode)} } return manifestOutcome{relay: h.captureResponse(resp)} } // digestMismatch reports whether two digests of the same algorithm differ. // Digests of different algorithms can't be compared without the body. func digestMismatch(requested, upstream string) bool { reqAlg, reqHex, _ := strings.Cut(requested, ":") upAlg, upHex, _ := strings.Cut(upstream, ":") return reqAlg == upAlg && !strings.EqualFold(reqHex, upHex) } // tagFromCachedDigest answers a tag from the cache once a HEAD named the // digest it points to: the cached tag entry if the tag did not move, or the // manifest stored under the new digest. It reports false when neither is // cached. func (h *ContainerHandler) tagFromCachedDigest(ctx context.Context, req manifestRequest, cached *cachedContainerManifest, fromFallback bool, resp *http.Response, digest string) (manifestOutcome, bool) { if cached != nil && cached.contentDigest == digest { cached.fetchedAt = time.Now() h.refreshManifest(ctx, req, cached, fromFallback) return manifestOutcome{manifest: cached}, true } stored, err := h.loadContainerManifest(ctx, h.containerManifestSharedCacheKey(req.registryURL, req.name, digest, req.cacheAccept)) if err != nil || stored == nil || stored.contentDigest != digest || !containerManifestCacheCompatible(req.accept, stored) { return manifestOutcome{}, false } moved := *stored moved.fetchedAt = time.Now() if etag := resp.Header.Get(headerETag); etag != "" { moved.etag = etag } if lastModified := parseHTTPTime(resp.Header.Get(headerLastModified)); !lastModified.IsZero() { moved.lastModified = lastModified } h.storeContainerManifestForAccept(ctx, req.registryURL, req.name, req.reference, req.accept, req.cacheAccept, &moved) return manifestOutcome{manifest: &moved}, true } // refreshManifest stores cached again with its new fetch time. func (h *ContainerHandler) refreshManifest(ctx context.Context, req manifestRequest, cached *cachedContainerManifest, fromFallback bool) { if fromFallback { // The shared entry holds another representation for other // clients; only this client's own entries are refreshed. h.storeContainerManifestLegacyEntries(ctx, req.registryURL, req.name, req.reference, req.accept, req.cacheAccept, cached) return } h.storeContainerManifestForAccept(ctx, req.registryURL, req.name, req.reference, req.accept, req.cacheAccept, cached) } // getManifest GETs fetchRef, the requested reference or the digest a tag // resolved to, and stores the manifest under the requested reference and // its digest. cached, if set, is revalidated with its ETag. func (h *ContainerHandler) getManifest(ctx context.Context, req manifestRequest, fetchRef string, cached *cachedContainerManifest, fromFallback bool) manifestOutcome { etag := "" if cached != nil { etag = cached.etag } resp, err := h.requestManifest(ctx, http.MethodGet, req, fetchRef, etag) if err != nil { return manifestOutcome{err: err} } defer func() { _ = resp.Body.Close() }() if resp.StatusCode == http.StatusNotModified && cached != nil { cached.fetchedAt = time.Now() h.refreshManifest(ctx, req, cached, fromFallback) return manifestOutcome{manifest: cached} } if resp.StatusCode != http.StatusOK { if shouldServeStaleManifest(resp.StatusCode) && (cached != nil || referrersTagPattern.MatchString(req.reference)) { return manifestOutcome{err: fmt.Errorf("upstream status %d", resp.StatusCode)} } return manifestOutcome{relay: h.captureResponse(resp)} } body, err := h.proxy.ReadMetadata(resp.Body) if err != nil { return manifestOutcome{err: fmt.Errorf("reading manifest: %w", err)} } contentDigest := resp.Header.Get("Docker-Content-Digest") // The references fetched and stored under, and the digest the upstream // says the manifest has, are all cache keys, so each digest among them // must be the manifest's own. One the proxy can't compute is refused // rather than cached unchecked. for _, expected := range []string{req.reference, fetchRef, contentDigest} { if !strings.Contains(expected, ":") { continue } if err := verifyDigest(expected, body); err != nil { h.proxy.Logger.Error("upstream manifest failed digest verification", "name", req.name, "reference", req.reference, "expected", expected, "error", err) return manifestOutcome{digestInvalid: true} } } if contentDigest == "" { contentDigest = sha256Digest(body) } manifest := &cachedContainerManifest{ body: body, contentType: resp.Header.Get(headerContentType), contentDigest: contentDigest, etag: resp.Header.Get(headerETag), size: int64(len(body)), lastModified: parseHTTPTime(resp.Header.Get(headerLastModified)), fetchedAt: time.Now(), } h.storeContainerManifestForAccept(ctx, req.registryURL, req.name, req.reference, req.accept, req.cacheAccept, manifest) if manifest.contentDigest != req.reference && manifestDigestReferencePattern.MatchString(manifest.contentDigest) { h.storeContainerManifestForAccept(ctx, req.registryURL, req.name, manifest.contentDigest, req.accept, req.cacheAccept, manifest) } return manifestOutcome{manifest: manifest} } // requestManifest sends one manifest request upstream with the client's // Accept header. The registry client adds credentials and follows bearer // challenges for HEAD as for GET. func (h *ContainerHandler) requestManifest(ctx context.Context, method string, req manifestRequest, reference, etag string) (*http.Response, error) { upstream, err := http.NewRequestWithContext(ctx, method, req.url(reference), nil) if err != nil { return nil, fmt.Errorf("creating manifest request: %w", err) } upstream.Header.Set("Accept", req.accept) if etag != "" { upstream.Header.Set("If-None-Match", etag) } return h.proxy.HTTPClient.Do(upstream) }