sower/cmd/sowerd/main.go

package main

import (
	"bufio"
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"net/http/httputil"
	"net/url"
	"os"
	"os/signal"
	"path/filepath"
	"runtime/pprof"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/cristalhq/aconfig"
	"github.com/cristalhq/aconfig/aconfigtoml"
	"github.com/lmittmann/tint"
	"github.com/sower-proxy/conns/relay"
	"github.com/sower-proxy/conns/reread"
	"github.com/sower-proxy/deferlog/v2"
	"github.com/sower-proxy/sower/config"
	"github.com/sower-proxy/sower/internal/install"
	transportSower "github.com/sower-proxy/sower/transport/sower"
	"golang.org/x/crypto/acme/autocert"
)

const (
	httpShutdownTimeout           = 5 * time.Second
	reverseProxyReadHeaderTimeout = 10 * time.Second
	reverseProxyIdleTimeout       = 30 * time.Second
	upstreamDialTimeout           = 10 * time.Second
	upstreamResponseTimeout       = 30 * time.Second
	systemCacheDir                = "/var/cache/sower"

	// Timeouts for the plain-HTTP listener, which only redirects to HTTPS and
	// answers ACME HTTP-01 challenges. Without them a peer that opens a
	// connection and never finishes a request keeps a goroutine, its socket
	// and its buffers for as long as it stays connected, which is an unbounded
	// hold on an internet-facing port.
	plainHTTPReadHeaderTimeout = 10 * time.Second
	plainHTTPReadTimeout       = 10 * time.Second
	plainHTTPWriteTimeout      = 10 * time.Second
	plainHTTPIdleTimeout       = 30 * time.Second

	// maxInboundConns caps how many connections sowerd serves at once. Each
	// one costs a goroutine stack, TLS record buffers and, while it is
	// relayed, up to 64 KiB of copy buffers, so an uncapped accept loop turns
	// a burst of connections (a scan wave, a reconnect storm, peers that
	// stopped reading) into resident memory proportional to the arrival rate.
	// At the measured per-connection cost this cap bounds that memory to a few
	// hundred MiB, and refusing a connection is cheaper than answering one.
	maxInboundConns = 2048

	// writeStallTimeout bounds a single write that makes no progress. The
	// relay library only arms read deadlines, so a peer that stops reading
	// blocks its write forever: both copy directions then wait on each other,
	// wg.Wait never returns and the connection pair, its goroutines and its
	// file descriptors leak for the process's lifetime. A stalled relayed
	// chunk is at most the copy buffer (64 KiB), so five minutes is far beyond
	// any live link while still reclaiming a dead one.
	writeStallTimeout = 5 * time.Minute
)

var (
	version, date string
)

func init() {
	setLogger(slog.LevelInfo)
}

func main() {
	if hasInstallFlag(os.Args[1:]) {
		if err := install.InstallService(stdinConfirm); err != nil {
			slog.Error("install service", "error", err)
			os.Exit(1)
		}
		return
	}

	conf, err := loadConfig()
	if err != nil {
		slog.Error("load config", "error", err)
		os.Exit(1)
	}

	setLogger(conf.LogLevel)
	slog.Info("load config",
		"version", version,
		"date", date,
		"config", sanitizeConfig(conf))

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	if err := run(ctx, conf); err != nil {
		slog.Error("sowerd exited", "error", err)
		os.Exit(1)
	}
}

func loadConfig() (config.SowerdConfig, error) {
	var conf config.SowerdConfig
	if err := aconfig.LoaderFor(&conf, aconfig.Config{
		AllowUnknownFlags:  false,
		AllowUnknownFields: false,
		FileFlag:           "c",
		Files: []string{
			"sowerd.toml",
			"config/sowerd.toml",
			"/etc/sower/sowerd.toml",
		},
		FileDecoders: map[string]aconfig.FileDecoder{
			".toml": aconfigtoml.New(),
		},
	}).Load(); err != nil {
		return config.SowerdConfig{}, err
	}
	if err := conf.Validate(); err != nil {
		return config.SowerdConfig{}, err
	}
	return conf, nil
}

func run(ctx context.Context, conf config.SowerdConfig) error {
	cacheDir, err := cacheDir()
	if err != nil {
		return err
	}

	certManager, tlsConf, err := buildTLSConfig(cacheDir, conf)
	if err != nil {
		return err
	}

	go dumpHeapOnSignal(ctx, cacheDir)

	fakeSite, dirServer, err := prepareFakeSite(conf.FakeSite, conf.ServeIP)
	if err != nil {
		return err
	}

	siteRouter := newSiteRouter(conf.SiteRoutes)

	httpAddr := net.JoinHostPort(conf.ServeIP, "80")
	httpServer := newRedirectServer(httpAddr, certManager.HTTPHandler(fakeSiteHandler(dirServer)))

	httpErrCh := make(chan error, 1)
	go func() {
		err := httpServer.ListenAndServe()
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			httpErrCh <- fmt.Errorf("listen http %s: %w", httpAddr, err)
		}
		close(httpErrCh)
	}()

	httpsAddr := net.JoinHostPort(conf.ServeIP, "443")
	rawLn, err := net.Listen("tcp", httpsAddr)
	if err != nil {
		_ = shutdownHTTP(context.Background(), httpServer)
		return fmt.Errorf("listen https %s: %w", httpsAddr, err)
	}
	// The stall wrapper sits below the TLS listener so that every write on the
	// connection is bounded: handshake, proxied response, relayed or hijacked
	// tunnel. Idle tunnels issue no writes and stay untouched.
	ln := tls.NewListener(newWriteStallListener(rawLn, writeStallTimeout), tlsConf)
	defer ln.Close()

	protocolHandlers := []proxyProtocolHandler{
		newSowerProtocolHandler(transportSower.New(conf.Password)),
	}

	httpsErrCh := make(chan error, 1)
	go func() {
		httpsErrCh <- serve443(ctx, ln, fakeSite, siteRouter, protocolHandlers, newConnLimiter(maxInboundConns))
		close(httpsErrCh)
	}()

	select {
	case err := <-httpErrCh:
		if err != nil {
			_ = ln.Close()
			return err
		}
	case err := <-httpsErrCh:
		if err != nil {
			_ = shutdownHTTP(context.Background(), httpServer)
			return err
		}
	case <-ctx.Done():
	}

	shutdownCtx, cancel := context.WithTimeout(context.Background(), httpShutdownTimeout)
	defer cancel()

	_ = ln.Close()
	if err := shutdownHTTP(shutdownCtx, httpServer); err != nil {
		return err
	}
	return nil
}

func cacheDir() (string, error) {
	dir, fallbackErr := resolveCacheDir(os.UserCacheDir, systemCacheDir)
	if fallbackErr != nil {
		slog.Warn("user cache dir unavailable, fallback to system cache dir",
			"error", fallbackErr,
			"dir", dir)
	}
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return "", fmt.Errorf("create cache dir %s: %w", dir, err)
	}
	return dir, nil
}

func resolveCacheDir(userCacheDir func() (string, error), fallbackDir string) (string, error) {
	base, err := userCacheDir()
	if err != nil {
		return fallbackDir, err
	}
	return filepath.Join(base, "sower"), nil
}

func buildTLSConfig(cacheDir string, cfg config.SowerdConfig) (*autocert.Manager, *tls.Config, error) {
	certManager := &autocert.Manager{
		Prompt: autocert.AcceptTOS,
		Email:  cfg.Cert.Email,
		Cache:  autocert.DirCache(cacheDir),
		// Only issue for names the operator owns: every site-routed domain
		// plus the explicit cert.domains whitelist. A nil HostPolicy lets
		// anyone trigger issuance for arbitrary SNI names, exhausting the
		// account's order/rate limits and polluting the cache.
		HostPolicy: autocert.HostWhitelist(certIssueDomains(cfg)...),
	}

	tlsConf := &tls.Config{
		GetCertificate: certManager.GetCertificate,
		MinVersion:     tls.VersionTLS12,
		NextProtos:     []string{"http/1.1"},
	}

	if cfg.Cert.Cert == "" {
		tlsConf.GetCertificate = getCertificateWithFallback(certManager, cfg)
		return certManager, tlsConf, nil
	}

	cert, err := tls.LoadX509KeyPair(cfg.Cert.Cert, cfg.Cert.Key)
	if err != nil {
		return nil, nil, fmt.Errorf("load cert pair: %w", err)
	}

	tlsConf.GetCertificate = nil
	tlsConf.Certificates = []tls.Certificate{cert}
	return certManager, tlsConf, nil
}

// getCertificateWithFallback answers the TLS handshake for names outside the
// issuance whitelist instead of failing it. Real sites present their default
// vhost certificate for arbitrary SNI, so a handshake failure on unknown
// names is itself an active-probing signal. Whitelisted names keep the plain
// autocert path: their failures (ACME issuance, cache) surface as-is and are
// never masked by a wrong-name certificate.
func getCertificateWithFallback(m *autocert.Manager, cfg config.SowerdConfig) func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
	domains := certIssueDomains(cfg)
	whitelist := make(map[string]struct{}, len(domains))
	for _, d := range domains {
		whitelist[strings.ToLower(d)] = struct{}{}
	}
	return func(hello *tls.ClientHelloInfo) (*tls.Certificate, error) {
		if _, ok := whitelist[strings.ToLower(hello.ServerName)]; ok {
			return m.GetCertificate(hello)
		}
		if len(domains) == 0 {
			return m.GetCertificate(hello)
		}
		// Carry the client's capabilities over so autocert derives the same
		// cache key (ECDSA vs RSA) it would for a direct visit; a bare hello
		// would count as non-ECDSA and force a needless RSA issuance.
		fallbackHello := *hello
		fallbackHello.ServerName = domains[0]
		return m.GetCertificate(&fallbackHello)
	}
}

// certIssueDomains collects the deduplicated domain whitelist for autocert
// issuance: every site_routes domain plus cert.domains.
func certIssueDomains(cfg config.SowerdConfig) []string {
	seen := make(map[string]struct{}, len(cfg.Cert.Domains))
	domains := make([]string, 0, len(cfg.Cert.Domains))
	add := func(d string) {
		d = strings.ToLower(d)
		if _, ok := seen[d]; ok {
			return
		}
		seen[d] = struct{}{}
		domains = append(domains, d)
	}
	for _, d := range cfg.Cert.Domains {
		add(d)
	}
	for _, route := range cfg.SiteRoutes {
		for _, d := range route.Domains {
			add(d)
		}
	}
	return domains
}

func prepareFakeSite(fakeSite string, serveIP string) (string, http.Handler, error) {
	si, err := os.Stat(fakeSite)
	if err == nil && si.IsDir() {
		slog.Info("serve fake site on http", "dir", fakeSite)
		// The directory fake site is served by the loopback HTTP server.
		// Prefer a concrete serve_ip as the relay target (its :80 listener
		// exists); otherwise fall back to the loopback address.
		target := "127.0.0.1:80"
		if ip := net.ParseIP(serveIP); ip != nil && !ip.IsUnspecified() {
			target = net.JoinHostPort(serveIP, "80")
		}
		return target, http.FileServer(http.Dir(fakeSite)), nil
	}
	if err != nil && !os.IsNotExist(err) {
		return "", nil, fmt.Errorf("stat fake site: %w", err)
	}
	return fakeSite, nil, nil
}

func fakeSiteHandler(dirServer http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if dirServer != nil && isLocalRemoteAddr(r.RemoteAddr) {
			dirServer.ServeHTTP(w, r)
			return
		}

		target := "https://" + r.Host + r.URL.RequestURI()
		http.Redirect(w, r, target, http.StatusPermanentRedirect)
	})
}

// newRedirectServer builds the plain-HTTP listener. Every timeout is set:
// this listener answers the whole internet and only ever serves a redirect or
// an ACME challenge, so a connection that is slow, idle or half-sent has
// nothing to wait for and must not be able to pin its goroutine and socket.
func newRedirectServer(addr string, handler http.Handler) *http.Server {
	return &http.Server{
		Addr:              addr,
		Handler:           handler,
		ReadHeaderTimeout: plainHTTPReadHeaderTimeout,
		ReadTimeout:       plainHTTPReadTimeout,
		WriteTimeout:      plainHTTPWriteTimeout,
		IdleTimeout:       plainHTTPIdleTimeout,
	}
}

// connLimiter is a counting semaphore over served connections. A connection
// that finds no slot is closed instead of queued: queueing would keep the
// memory the cap exists to bound.
type connLimiter struct {
	slots chan struct{}
}

func newConnLimiter(n int) *connLimiter {
	return &connLimiter{slots: make(chan struct{}, n)}
}

func (l *connLimiter) acquire() bool {
	select {
	case l.slots <- struct{}{}:
		return true
	default:
		return false
	}
}

func (l *connLimiter) release() { <-l.slots }

// writeStallListener arms a write deadline on every accepted connection, so a
// write that a peer never accepts fails instead of blocking the writer
// forever. It wraps the raw listener so the TLS layer - and therefore every
// path above it - is covered.
type writeStallListener struct {
	net.Listener
	stall time.Duration
}

func newWriteStallListener(ln net.Listener, stall time.Duration) *writeStallListener {
	return &writeStallListener{Listener: ln, stall: stall}
}

func (l *writeStallListener) Accept() (net.Conn, error) {
	conn, err := l.Listener.Accept()
	if err != nil {
		return nil, err
	}
	return &writeStallConn{Conn: conn, stall: l.stall}, nil
}

type writeStallConn struct {
	net.Conn
	stall time.Duration
}

func (c *writeStallConn) Write(b []byte) (int, error) {
	if err := c.Conn.SetWriteDeadline(time.Now().Add(c.stall)); err != nil {
		return 0, err
	}
	return c.Conn.Write(b)
}

func dumpHeapProfile(dir string) (string, error) {
	path := filepath.Join(dir, fmt.Sprintf("sowerd-heap-%d.pprof", time.Now().Unix()))
	f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600)
	if err != nil {
		return "", fmt.Errorf("create heap profile %s: %w", path, err)
	}
	defer f.Close()

	if err := pprof.WriteHeapProfile(f); err != nil {
		return "", fmt.Errorf("write heap profile %s: %w", path, err)
	}
	return path, nil
}

func serve443(ctx context.Context, ln net.Listener, fakeSite string, router siteRouter, handlers []proxyProtocolHandler, limiter *connLimiter) error {
	for {
		conn, err := ln.Accept()
		if err != nil {
			if ctx.Err() != nil || errors.Is(err, net.ErrClosed) {
				return nil
			}

			var netErr net.Error
			if errors.As(err, &netErr) && netErr.Temporary() {
				slog.Warn("temporary accept error", "error", err)
				time.Sleep(100 * time.Millisecond)
				continue
			}
			return fmt.Errorf("accept tls connection: %w", err)
		}

		if !limiter.acquire() {
			slog.Debug("connection limit reached, dropping", "remote", conn.RemoteAddr(), "limit", maxInboundConns)
			_ = conn.Close()
			continue
		}
		go func() {
			defer limiter.release()
			handleConn(conn, fakeSite, router, handlers)
		}()
	}
}

func handleConn(conn net.Conn, fakeSite string, router siteRouter, handlers []proxyProtocolHandler) {
	rereadConn := reread.New(conn)
	var hijacked atomic.Bool
	defer func() {
		if !hijacked.Load() {
			rereadConn.Close()
		}
	}()

	// Complete the TLS handshake explicitly before probing: on first
	// contact the handshake can take tens of seconds while GetCertificate
	// drives ACME issuance, and the short probe deadline must not race it
	// (the first connection to a new domain would otherwise always fail).
	if tlsConn, ok := conn.(*tls.Conn); ok {
		handshakeCtx, cancel := context.WithTimeout(context.Background(), tlsHandshakeTimeout)
		if err := tlsConn.HandshakeContext(handshakeCtx); err != nil {
			cancel()
			return
		}
		cancel()
	}

	_ = rereadConn.SetReadDeadline(time.Now().Add(protocolProbeTimeout))

	var (
		addr net.Addr
		dur  time.Duration
		err  error
	)
	defer func() {
		deferlog.DebugWarn(err, "relay conn", "took", dur, "addr", addr)
	}()

	rereadConn.Reread()
	probeBuf, err := readProtocolProbe(rereadConn)
	if err != nil {
		return
	}

	for _, handler := range handlers {
		switch handler.Probe(probeBuf) {
		case probeNoMatch, probeNeedMore:
			continue
		case probeMatch:
			rereadConn.Reread()
			// Bound the header read with its own (longer than the probe)
			// deadline: the header follows the TLS handshake immediately,
			// but a slow client needs more than the 1s probe window, and an
			// attacker that sends only the 0x80 probe byte must not hold
			// the connection (and its goroutine) forever.
			_ = rereadConn.SetReadDeadline(time.Now().Add(protocolHeaderTimeout))
			if addr, err = handler.Unwrap(rereadConn); err == nil {
				_ = rereadConn.SetReadDeadline(time.Time{})
				rereadConn.Stop()
				dur, err = relay.RelayTo(rereadConn, addr.String())
				return
			}

			slog.Debug("protocol auth or decode failed, fallback", "protocol", handler.Name(), "error", err)
			_ = rereadConn.SetReadDeadline(time.Time{})
			rereadConn.Stop().Reread()
			dur, err = fallbackConn(rereadConn, conn, fakeSite, router, &hijacked, probeBuf)
			return
		}
	}

	probeBuf = extendHTTPRequestProbe(rereadConn, probeBuf)
	rereadConn.Stop().Reread()
	_ = rereadConn.SetReadDeadline(time.Time{})
	dur, err = fallbackConn(rereadConn, conn, fakeSite, router, &hijacked, probeBuf)
}

func fallbackConn(conn net.Conn, tlsConn net.Conn, fakeSite string, router siteRouter, hijacked *atomic.Bool, probe []byte) (time.Duration, error) {
	start := time.Now()
	if entry := router.lookup(sniFromConn(tlsConn)); entry != nil {
		return time.Since(start), reverseProxyConn(conn, entry, hijacked)
	}
	if httpRequestLineProbe(probe) {
		return time.Since(start), reverseProxyConn(conn, newFakeSiteEntry(fakeSite), hijacked)
	}
	return relay.RelayTo(conn, fakeSite)
}

func newFakeSiteEntry(fakeSite string) *siteEntry {
	return &siteEntry{
		upstream:     &url.URL{Scheme: "http", Host: fakeSite},
		preserveHost: true,
	}
}

func sanitizeConfig(cfg config.SowerdConfig) map[string]any {
	return map[string]any{
		"log_level":   cfg.LogLevel.String(),
		"serve_ip":    cfg.ServeIP,
		"fake_site":   cfg.FakeSite,
		"site_routes": len(cfg.SiteRoutes),
		"cert": map[string]any{
			"email":       cfg.Cert.Email,
			"cert_config": cfg.Cert.Cert != "",
			"key_config":  cfg.Cert.Key != "",
		},
	}
}

func hasInstallFlag(args []string) bool {
	for _, arg := range args {
		switch arg {
		case "-i", "--install":
			return true
		}
	}
	return false
}

func stdinConfirm(label string) bool {
	fmt.Printf("%s [y/N]: ", label)
	scanner := bufio.NewScanner(os.Stdin)
	if !scanner.Scan() {
		return false
	}
	answer := strings.TrimSpace(strings.ToLower(scanner.Text()))
	return answer == "y" || answer == "yes"
}

// isLocalRemoteAddr reports whether the connection originated on this host:
// loopback or an address bound to a local interface. Directory fake sites
// relay fallback traffic from this process; the relay's source address
// follows the target (loopback for the default 0.0.0.0 serve_ip, the
// serve_ip itself for a concrete bind), so a loopback-only check would
// wrongly redirect concrete serve_ip deployments.
func isLocalRemoteAddr(remoteAddr string) bool {
	host, _, err := net.SplitHostPort(remoteAddr)
	if err != nil {
		host = remoteAddr
	}
	host = strings.Trim(host, "[]")

	ip := net.ParseIP(host)
	if ip == nil {
		return false
	}
	if ip.IsLoopback() {
		return true
	}

	addrs, err := net.InterfaceAddrs()
	if err != nil {
		return false
	}
	for _, a := range addrs {
		if ipn, ok := a.(*net.IPNet); ok && ipn.IP.Equal(ip) {
			return true
		}
	}
	return false
}

func setLogger(level slog.Level) {
	fi, err := os.Stdout.Stat()
	noColor := err != nil || (fi.Mode()&os.ModeCharDevice) == 0
	deferlog.SetDefault(slog.New(tint.NewHandler(os.Stdout, &tint.Options{
		AddSource: true,
		NoColor:   noColor,
		Level:     level,
	})))
}

func shutdownHTTP(ctx context.Context, server *http.Server) error {
	if err := server.Shutdown(ctx); err != nil && !errors.Is(err, http.ErrServerClosed) {
		return fmt.Errorf("shutdown http server: %w", err)
	}
	return nil
}

// pathRoute maps a URL path prefix to an upstream URL, overriding the site
// entry default. Paths are normalized without a trailing slash and matched
// longest-prefix first.
type pathRoute struct {
	path     string
	upstream *url.URL
}

// siteEntry is the routing target for one site route: a default upstream plus
// optional path-prefix overrides and an optional visitor gate.
type siteEntry struct {
	upstream     *url.URL
	paths        []pathRoute
	basicAuth    *siteBasicAuth // nil when the route is not gated
	preserveHost bool           // forward the client's Host: site routes opt in, fake-site fallback always does
}

// resolve returns the upstream URL for the given request path. A configured
// root path "/" matches every request, acting as a catch-all override.
func (e *siteEntry) resolve(path string) *url.URL {
	for _, pr := range e.paths {
		if pr.path == "/" || path == pr.path || strings.HasPrefix(path, pr.path+"/") {
			return pr.upstream
		}
	}
	return e.upstream
}

// siteRouter maps TLS SNI to the site entry for fallback traffic.
type siteRouter struct {
	routes map[string]*siteEntry
}

func newSiteRouter(routes []config.SiteRoute) siteRouter {
	m := make(map[string]*siteEntry)
	// One throttle for every gated route: a visitor guessing at one route must
	// not get a fresh budget by moving to another.
	throttle := newAuthThrottle()
	for _, r := range routes {
		u, _ := url.Parse(r.Upstream) // validated in config.Validate
		entry := &siteEntry{upstream: u, basicAuth: newSiteBasicAuth(r.Auth, throttle), preserveHost: r.PreserveHost()}
		for path, upstream := range r.Routes {
			pu, _ := url.Parse(upstream) // validated in config.Validate
			entry.paths = append(entry.paths, pathRoute{
				path:     normalizeRoutePath(path),
				upstream: pu,
			})
		}
		sort.Slice(entry.paths, func(i, j int) bool {
			return len(entry.paths[i].path) > len(entry.paths[j].path)
		})
		for _, d := range r.Domains {
			m[strings.ToLower(d)] = entry
		}
	}
	return siteRouter{routes: m}
}

// normalizeRoutePath trims a single trailing slash so that "/a" and "/a/"
// route identically; "/" is kept as-is.
func normalizeRoutePath(path string) string {
	if path != "/" {
		path = strings.TrimSuffix(path, "/")
	}
	return path
}

// lookup returns the site entry for the given SNI, or nil if no route matches.
func (r siteRouter) lookup(sni string) *siteEntry {
	return r.routes[strings.ToLower(sni)]
}

// sniFromConn extracts the TLS SNI from a *tls.Conn.
func sniFromConn(conn net.Conn) string {
	tlsConn, ok := conn.(*tls.Conn)
	if !ok {
		return ""
	}
	return tlsConn.ConnectionState().ServerName
}

func stripResidualForwardingHeaders(h http.Header) {
	h.Del("X-Real-IP")
	for k := range h {
		if strings.HasPrefix(k, "X-Forwarded-") {
			h.Del(k)
		}
	}
}

// setEdgeProxyHeaders overwrites forwarding headers from the TLS-edge peer.
// ReverseProxy.Rewrite already deleted Forwarded / X-Forwarded-For /
// X-Forwarded-Host / X-Forwarded-Proto and does not auto-populate them.
// X-Forwarded-Proto is https because this path is the 443 listener after TLS
// termination (req.TLS is nil on the decrypted conn).
func setEdgeProxyHeaders(req *http.Request, forwardedHost, remoteAddr string) {
	if forwardedHost != "" {
		req.Header.Set("X-Forwarded-Host", forwardedHost)
	}
	req.Header.Set("X-Forwarded-Proto", "https")
	if clientIP, _, err := net.SplitHostPort(remoteAddr); err == nil {
		req.Header.Set("X-Forwarded-For", clientIP)
		req.Header.Set("X-Real-IP", clientIP)
	}
}

// proxyErrorHandler replaces the default ReverseProxy error handler, which
// only logs a bare "http: proxy error: <err>". With site routes and path
// overrides the routing context is what identifies which upstream failed
// (the SNI-matched route host, the request path and the upstream URL).
func proxyErrorHandler(upstream *url.URL) func(http.ResponseWriter, *http.Request, error) {
	return func(w http.ResponseWriter, req *http.Request, err error) {
		slog.Warn("proxy upstream error",
			"error", err,
			// Redacted: the upstream URL can carry Basic credentials in its
			// userinfo, which must not reach the log.
			"upstream", upstream.Redacted(),
			"method", req.Method,
			"host", req.Host,
			"path", req.URL.Path,
			"remote", req.RemoteAddr)

		w.WriteHeader(http.StatusBadGateway)
		_, _ = w.Write([]byte("bad gateway\n"))
	}
}

// reverseProxyConn serves the decrypted HTTP connection through a reverse
// proxy to the upstream selected per request by the site entry's path routes.
// Reverse proxies and their transports are created lazily per upstream and
// closed together when the connection ends.
func reverseProxyConn(conn net.Conn, entry *siteEntry, hijacked *atomic.Bool) error {
	proxies := make(map[string]*httputil.ReverseProxy, 1+len(entry.paths))
	transports := make([]*http.Transport, 0, 1+len(entry.paths))

	proxyFor := func(upstream *url.URL) *httputil.ReverseProxy {
		key := upstream.String()
		if proxy, ok := proxies[key]; ok {
			return proxy
		}
		proxy := &httputil.ReverseProxy{
			Rewrite: func(pr *httputil.ProxyRequest) {
				pr.SetURL(upstream)
				if entry.preserveHost {
					pr.Out.Host = pr.In.Host
				} else {
					pr.Out.Host = upstream.Host
				}
				stripResidualForwardingHeaders(pr.Out.Header)
				setEdgeProxyHeaders(pr.Out, pr.In.Host, pr.In.RemoteAddr)
				if upstream.User != nil {
					// net/http applies URL userinfo as Basic credentials only in
					// http.Client; ReverseProxy round-trips the request through a
					// Transport, which drops them. Configured upstream credentials
					// are explicit, so they override any inbound Authorization.
					password, _ := upstream.User.Password()
					pr.Out.Header.Set("Authorization", basicAuthHeader(upstream.User.Username(), password))
				}
			},
			ErrorHandler: proxyErrorHandler(upstream),
		}
		transport := &http.Transport{
			Proxy: http.ProxyFromEnvironment,
			DialContext: (&net.Dialer{
				Timeout:   upstreamDialTimeout,
				KeepAlive: upstreamResponseTimeout,
			}).DialContext,
			TLSHandshakeTimeout:   upstreamDialTimeout,
			ResponseHeaderTimeout: upstreamResponseTimeout,
			IdleConnTimeout:       reverseProxyIdleTimeout,
			ExpectContinueTimeout: time.Second,
		}
		transports = append(transports, transport)
		proxy.Transport = transport
		proxy.ErrorLog = slog.NewLogLogger(slog.Default().Handler(), slog.LevelWarn)
		proxies[key] = proxy
		return proxy
	}

	defer func() {
		for _, t := range transports {
			t.CloseIdleConnections()
		}
	}()

	handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
		// The gate covers the whole route, including its path overrides and
		// WebSocket upgrades: it runs before request dispatch to any upstream.
		if entry.basicAuth != nil && !entry.basicAuth.guard(w, req) {
			return
		}
		proxyFor(entry.resolve(req.URL.Path)).ServeHTTP(w, req)
	})

	ln := newSingleConnListener(conn)
	srv := &http.Server{
		Handler:           handler,
		ReadHeaderTimeout: reverseProxyReadHeaderTimeout,
		IdleTimeout:       reverseProxyIdleTimeout,
		ConnState: func(_ net.Conn, state http.ConnState) {
			switch state {
			case http.StateHijacked:
				hijacked.Store(true)
				_ = ln.Close()
			case http.StateClosed:
				_ = ln.Close()
			}
		},
	}

	if err := srv.Serve(ln); err != nil && !errors.Is(err, net.ErrClosed) && !errors.Is(err, http.ErrServerClosed) {
		return fmt.Errorf("serve reverse proxy connection: %w", err)
	}
	return nil
}

// singleConnListener wraps a single net.Conn as a net.Listener.
// Accept returns the conn once, then blocks until Close is called.
type singleConnListener struct {
	conn      net.Conn
	once      bool
	closed    chan struct{}
	closeOnce sync.Once
}

func newSingleConnListener(conn net.Conn) *singleConnListener {
	return &singleConnListener{
		conn:   conn,
		closed: make(chan struct{}),
	}
}

func (l *singleConnListener) Accept() (net.Conn, error) {
	if !l.once {
		l.once = true
		return l.conn, nil
	}
	<-l.closed
	return nil, net.ErrClosed
}

func (l *singleConnListener) Close() error {
	l.closeOnce.Do(func() {
		close(l.closed)
	})
	return nil
}

func (l *singleConnListener) Addr() net.Addr { return l.conn.LocalAddr() }