tiny-proxy/internal/distro/handler.go

// Package distro implements read-only pull-through caching for Linux
// distribution package mirrors: APT (debian type) and pacman (pacman type).
// Paths are /<type>/<vrepo>/<upstream path>, relayed to the upstream as-is.
// Mutable metadata (Release, Packages.*, <repo>.db, ...) is cached for
// metadata_ttl; content-addressed files (pool/**, by-hash/**, *.pkg.tar.*)
// are cached forever. Multiple upstreams may be configured as ordered
// mirrors: a 404, 401, 403, 429, 502, 503, 504 or connection error falls
// through to the next one, any other status is returned as-is.
package distro

import (
	"context"
	"io"
	"log/slog"
	"net/http"
	"strconv"
	"strings"

	"tiny-proxy/internal/auth"
	"tiny-proxy/internal/config"
	"tiny-proxy/internal/metrics"
	"tiny-proxy/internal/proxycache"
	"tiny-proxy/internal/upstream"
)

// Spec describes one distro repo flavor: its vrepo type (also the URL and
// cache-key prefix) and which upstream paths may change after publication.
type Spec struct {
	Type      config.VRepoType
	IsMutable func(path string) bool
}

var (
	Debian = Spec{Type: config.VRepoDebian, IsMutable: isMutableDebian}
	Pacman = Spec{Type: config.VRepoPacman, IsMutable: isMutablePacman}
)

// upstreamEntry pairs one upstream's config with its HTTP client.
type upstreamEntry struct {
	conf   config.UpstreamConf
	client *http.Client
}

type repo struct {
	cfg       config.VRepo
	upstreams []upstreamEntry
}

type Handler struct {
	spec  Spec
	repos map[string]*repo
	cache *proxycache.Cache
}

// NewHandler builds a handler for the given spec. clients[vrepo] must have
// one client per entry of vrepo.AllUpstreams(), in the same order.
func NewHandler(spec Spec, repos []config.VRepo, clients map[string][]*http.Client, cache *proxycache.Cache) *Handler {
	h := &Handler{spec: spec, repos: map[string]*repo{}, cache: cache}
	for _, rc := range repos {
		if rc.Type != spec.Type {
			continue
		}
		rp := &repo{cfg: rc}
		for i, u := range rc.AllUpstreams() {
			rp.upstreams = append(rp.upstreams, upstreamEntry{conf: u, client: clients[rc.Name][i]})
		}
		h.repos[rc.Name] = rp
	}
	return h
}

func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet && r.Method != http.MethodHead {
		http.Error(w, "read-only repository", http.StatusMethodNotAllowed)
		return
	}
	prefix := string(h.spec.Type)
	rest := strings.TrimPrefix(r.URL.Path, "/"+prefix+"/")
	vrepo, path, ok := strings.Cut(rest, "/")
	if !ok || !validPath(path) {
		http.Error(w, "bad path", http.StatusBadRequest)
		return
	}
	rp, exists := h.repos[vrepo]
	if !exists {
		http.Error(w, "unknown repository", http.StatusNotFound)
		return
	}
	if !auth.Check(rp.cfg.Auth, r) {
		auth.Unauthorized(w)
		return
	}
	// Configured and authorized, so it is safe as a metrics label; the cache
	// reads it back from the context to attribute its hits and misses.
	metrics.MarkVRepo(r.Context(), vrepo)

	key := prefix + "/" + vrepo + "/" + path
	ttl := rp.cfg.MutableTTL()
	if !h.spec.IsMutable(path) {
		ttl = 0
	}

	h.cache.Serve(w, r, key, ttl, func(ctx context.Context) (proxycache.Result, error) {
		return rp.fetch(ctx, r, vrepo, path)
	})
}

// fetch walks the upstream list in configured order: 404, 401, 403, 429,
// 502, 503, 504 and transport errors fall through to the next upstream
// (mirrors of the same archive may lag, reject anonymous reads, throttle a
// busy client, or sit down behind a gateway); any other status, 500
// included, is returned as-is, and so is whatever the last mirror answers.
// When no mirror served the path but one
// throttled us, the answer is a 429 rather than a 404, so the client retries
// instead of treating the file as gone. Deliberately mirrors maven's fetch
// loop minus its filters — the maven loop is woven through
// eligible()/moreEligible() and extracting it would couple two handlers for
// ~15 saved lines. The result carries the answering upstream's cache policy:
// an upstream marked cache: false is proxied through without being stored
// here.
func (rp *repo) fetch(ctx context.Context, r *http.Request, vrepo, path string) (proxycache.Result, error) {
	limited, retryAfter := false, "" // whether a mirror throttled us, and its Retry-After
	for i, up := range rp.upstreams {
		upstreamURL := strings.TrimSuffix(up.conf.URL, "/") + "/" + path
		req, err := http.NewRequestWithContext(ctx, r.Method, upstreamURL, nil)
		if err != nil {
			return proxycache.Result{}, err
		}
		if rng := r.Header.Get("Range"); rng != "" {
			req.Header.Set("Range", rng)
		}
		resp, err := up.client.Do(req)
		if err != nil {
			slog.WarnContext(ctx, "distro upstream error, trying next", "vrepo", vrepo, "path", path, "url", up.conf.URL, "err", err)
			continue
		}
		if resp.StatusCode == http.StatusTooManyRequests && !limited {
			slog.WarnContext(ctx, "distro upstream rate limited", "vrepo", vrepo, "path", path, "url", up.conf.URL)
			limited, retryAfter = true, resp.Header.Get("Retry-After")
		}
		if shouldFallthrough(resp.StatusCode) && i+1 < len(rp.upstreams) {
			resp.Body.Close()
			slog.DebugContext(ctx, "distro upstream miss, trying next", "vrepo", vrepo, "path", path, "url", up.conf.URL, "status", resp.StatusCode)
			continue
		}
		if limited && resp.StatusCode == http.StatusNotFound {
			// A mirror was throttled, so "not found" is not
			// trustworthy: report the throttling instead.
			resp.Body.Close()
			return proxycache.Result{Resp: rateLimitedResponse(r, path, retryAfter)}, nil
		}
		return proxycache.Result{
			Resp:    resp,
			NoStore: !up.conf.CacheEnabled(),
			// Mirrors serve large packages; a dropped connection halfway
			// through one should resume from this same mirror, not restart
			// or fall through to another that may carry a different build.
			Resume:         upstream.Resumer(up.client, r.Method, upstreamURL),
			ResumeAttempts: up.conf.RetryAttempts(),
			ResumeMaxWait:  up.conf.RetryMaxWait(),
		}, nil
	}
	if limited {
		// Every mirror failed and at least one throttled us.
		return proxycache.Result{Resp: rateLimitedResponse(r, path, retryAfter)}, nil
	}
	return proxycache.Result{Resp: notFoundResponse(r, path)}, nil
}

// shouldFallthrough reports whether an upstream status means "try the next
// upstream": not found, anonymous access rejected (some mirrors answer
// 401/403 for anything when anonymous reads are disabled), rate limited
// (the mirror's client already waited out retry.attempts before giving up),
// or a gateway error — what an ingress answers for a chained tiny-proxy
// that is down, and what tiny-proxy itself answers when its own upstream
// fails. A plain 500 is the mirror's own answer and is returned as-is.
func shouldFallthrough(code int) bool {
	switch code {
	case http.StatusNotFound, http.StatusUnauthorized, http.StatusForbidden,
		http.StatusTooManyRequests, http.StatusBadGateway,
		http.StatusServiceUnavailable, http.StatusGatewayTimeout:
		return true
	}
	return false
}

// notFoundResponse synthesizes a 404 so proxycache relays it uncached when no
// upstream could serve the path.
func notFoundResponse(r *http.Request, path string) *http.Response {
	return statusResponse(r, http.StatusNotFound, "not found: "+path+"\n", nil)
}

// rateLimitedResponse synthesizes the 429 returned when throttling, not
// absence, is why the path could not be served, carrying over the throttling
// mirror's Retry-After if it sent one.
func rateLimitedResponse(r *http.Request, path, retryAfter string) *http.Response {
	header := http.Header{}
	if retryAfter != "" {
		header.Set("Retry-After", retryAfter)
	}
	return statusResponse(r, http.StatusTooManyRequests,
		"rate limited by upstream: "+path+"\n", header)
}

// statusResponse builds a synthetic response for proxycache to relay; only
// 200s are cached, so these are always passed through uncached.
func statusResponse(r *http.Request, code int, body string, header http.Header) *http.Response {
	if header == nil {
		header = http.Header{}
	}
	header.Set("Content-Type", "text/plain; charset=utf-8")
	return &http.Response{
		Status:        strconv.Itoa(code) + " " + http.StatusText(code),
		StatusCode:    code,
		Proto:         "HTTP/1.1",
		ProtoMajor:    1,
		ProtoMinor:    1,
		Header:        header,
		Body:          io.NopCloser(strings.NewReader(body)),
		ContentLength: int64(len(body)),
		Request:       r,
	}
}

// validPath rejects traversal and malformed paths. The mux already cleans
// dot segments before dispatch, but cache keys become localfs paths, so the
// guard is defense-in-depth for direct ServeHTTP calls (tests, embedding).
func validPath(p string) bool {
	if p == "" || strings.HasPrefix(p, "/") || strings.Contains(p, "\\") {
		return false
	}
	for _, seg := range strings.Split(p, "/") {
		if seg == "" || seg == "." || seg == ".." {
			return false
		}
	}
	return true
}