From 92b26005f7c25bf77a025abcc56c7bfc66ab4aae Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Tue, 22 Sep 2026 18:10:15 -0700 Subject: [PATCH] e2e: serve the harness over locally trusted HTTPS atproto OAuth will not run against the harness as it was: the Go OAuth client wants a portless https:// PDS issuer, the PDS's OAuth provider refuses client IDs on .test-style names and same-site navigations, and oatproxy and the browser client resolve did:plc against a hardcoded https://plc.directory. `streamplace e2e --https-pds-hostname --https-station-hostname` takes public DNS names for 127.0.0.1 and: - mints a throwaway CA and terminates TLS on 127.0.0.1:443, routing by SNI to the dev-env PDS (and handles under it), the node (running --behind-https-proxy) and, as plc.directory, the dev-env PLC; - runs a CONNECT proxy the node and browser use to reach that PLC; - trusts the CA per process (SSL_CERT_DIR, NODE_EXTRA_CA_CERTS, the leaf's SPKI for Chromium), installing nothing system-wide; - publishes this build's lexicons into a local lexicon authority account the PDS resolves permission sets from, so the OAuth scope include:place.stream.authFull is the branch's own. Without the flags the harness behaves as before. Co-Authored-By: Claude Opus 5.5 (1M context) --- js/dev-env/run.mjs | 44 +++++- pkg/cmd/e2e.go | 128 ++++++++++++++-- pkg/cmd/e2e_https.go | 339 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 497 insertions(+), 14 deletions(-) create mode 100644 pkg/cmd/e2e_https.go diff --git a/js/dev-env/run.mjs b/js/dev-env/run.mjs index d443d6b5b..369eef25e 100644 --- a/js/dev-env/run.mjs +++ b/js/dev-env/run.mjs @@ -1,8 +1,46 @@ import { TestNetwork } from "./dist/index.js"; +// With DEV_ENV_PDS_HOSTNAME the PDS presents itself as https:// +// (its OAuth issuer, did:web and account handles under .) while +// still listening on plain HTTP; whoever sets it terminates TLS in front of +// it. `streamplace e2e --https-pds-hostname` does, see pkg/cmd/e2e_https.go. +// +// In that mode there is also a lexicon authority account, the local stand-in +// for the did:plc account that _lexicon.stream.place points at in production: +// the PDS resolves every lexicon (OAuth permission sets) from it. Its +// credentials are printed for the caller to publish into. +const hostname = process.env.DEV_ENV_PDS_HOSTNAME; +const lexiconPassword = "lexicons"; + (async () => { - const network = await TestNetwork.create({}); - console.log( - JSON.stringify({ "pds-url": network.pds.url, "plc-url": network.plc.url }), + const network = await TestNetwork.create( + hostname + ? { pds: { hostname, serviceHandleDomains: [`.${hostname}`] } } + : {}, ); + const out = { "pds-url": network.pds.url, "plc-url": network.plc.url }; + if (hostname) { + const res = await fetch( + `${network.pds.url}/xrpc/com.atproto.server.createAccount`, + { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + handle: `lexicons.${hostname}`, + email: `lexicons@${hostname}`, + password: lexiconPassword, + }), + }, + ); + if (!res.ok) { + throw new Error(`create lexicon account: ${await res.text()}`); + } + const { did } = await res.json(); + // PDS_LEXICON_AUTHORITY_DID, set after the fact: the account can only be + // created once the PDS is up, and the PDS reads this per lookup. + network.pds.ctx.cfg.lexicon.didAuthority = did; + out["lexicon-did"] = did; + out["lexicon-password"] = lexiconPassword; + } + console.log(JSON.stringify(out)); })(); diff --git a/pkg/cmd/e2e.go b/pkg/cmd/e2e.go index cb3a0caa4..f763c1413 100644 --- a/pkg/cmd/e2e.go +++ b/pkg/cmd/e2e.go @@ -8,6 +8,7 @@ import ( "fmt" "net" "net/http" + "net/url" "os" "os/exec" "os/signal" @@ -23,6 +24,7 @@ import ( urfavecli "github.com/urfave/cli/v3" "golang.org/x/sync/errgroup" "stream.place/streamplace/pkg/aqhttp" + "stream.place/streamplace/pkg/atproto" spcomatproto "stream.place/streamplace/pkg/comatproto" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/crypto/spkey" @@ -57,9 +59,25 @@ func makeE2eCommand(build *config.BuildFlags) *urfavecli.Command { Value: "js/dev-env/run.mjs", Sources: urfavecli.EnvVars("SP_DEV_ENV_MJS"), }, + // Together these turn on locally trusted HTTPS, so atproto OAuth + // works end to end; see e2e_https.go. + &urfavecli.StringFlag{ + Name: "https-pds-hostname", + Usage: "serve the PDS at https://, with account handles under it; it and *. must resolve to 127.0.0.1 (binds 127.0.0.1:443)", + Sources: urfavecli.EnvVars("SP_E2E_HTTPS_PDS_HOSTNAME"), + }, + &urfavecli.StringFlag{ + Name: "https-station-hostname", + Usage: "the node's broadcaster host, served at https://, which must resolve to 127.0.0.1", + Sources: urfavecli.EnvVars("SP_E2E_HTTPS_STATION_HOSTNAME"), + }, }, Action: func(ctx context.Context, cmd *urfavecli.Command) error { - return runE2E(ctx, cmd.String("dev-env")) + pdsHost, stationHost := cmd.String("https-pds-hostname"), cmd.String("https-station-hostname") + if (pdsHost == "") != (stationHost == "") { + return errors.New("--https-pds-hostname and --https-station-hostname go together") + } + return runE2E(ctx, cmd.String("dev-env"), pdsHost, stationHost) }, } } @@ -72,6 +90,9 @@ const nodeReadyTimeout = 90 * time.Second type e2eDevEnv struct { PDSURL string `json:"pds-url"` PLCURL string `json:"plc-url"` + // Only in HTTPS mode: the account the PDS resolves lexicons from. + LexiconDID string `json:"lexicon-did"` + LexiconPassword string `json:"lexicon-password"` } func freePort() (int, error) { @@ -83,15 +104,32 @@ func freePort() (int, error) { return l.Addr().(*net.TCPAddr).Port, nil } -func runE2E(ctx context.Context, devEnvPath string) error { +func runE2E(ctx context.Context, devEnvPath, httpsPDSHost, httpsStationHost string) error { // Ctrl-C / SIGTERM must unwind through the normal path, or none of the // teardown below runs: Go's default handling exits immediately, which left // the dev-env node, the forked node and the temp data dir behind. ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop() + // Claim the HTTPS listeners first: failing to bind 443 should stop us + // before anything else has started. + var tlsEnv *e2eHTTPS + handleDomain := "test" + if httpsPDSHost != "" { + var err error + tlsEnv, err = newE2EHTTPS(httpsPDSHost, httpsStationHost) + if err != nil { + return err + } + defer tlsEnv.Close() + handleDomain = httpsPDSHost + } + // Start the Node.js PDS/PLC dev environment. devEnvCmd := exec.CommandContext(ctx, "node", devEnvPath) + if tlsEnv != nil { + devEnvCmd.Env = append(os.Environ(), tlsEnv.DevEnvEnv()...) + } devEnvCmd.Stderr = os.Stderr devEnvStdout, err := devEnvCmd.StdoutPipe() if err != nil { @@ -102,11 +140,16 @@ func runE2E(ctx context.Context, devEnvPath string) error { } defer devEnvCmd.Process.Kill() //nolint:errcheck + // The PDS logs JSON to stdout too when LOG_ENABLED is set, so skip lines + // until the one that carries the URLs. var env e2eDevEnv scanner := bufio.NewScanner(devEnvStdout) - scanner.Scan() - if err := json.Unmarshal(scanner.Bytes(), &env); err != nil { - return fmt.Errorf("parse dev-env output: %w", err) + for env.PDSURL == "" { + if !scanner.Scan() { + return fmt.Errorf("dev-env exited without printing its URLs: %w", scanner.Err()) + } + log.Log(ctx, "dev-env: "+scanner.Text()) + _ = json.Unmarshal(scanner.Bytes(), &env) //nolint:errcheck // not every line is ours } go func() { for scanner.Scan() { @@ -120,7 +163,7 @@ func runE2E(ctx context.Context, devEnvPath string) error { if err != nil { return err } - handle := fmt.Sprintf("sp-%s.test", uu.String()[:8]) + handle := fmt.Sprintf("sp-%s.%s", uu.String()[:8], handleDomain) email := fmt.Sprintf("%s@example.com", handle) password := "test" out, err := comatproto.ServerCreateAccount(ctx, xrpcc, &comatproto.ServerCreateAccount_Input{ @@ -149,6 +192,12 @@ func runE2E(ctx context.Context, devEnvPath string) error { }, } + if env.LexiconDID != "" { + if err := publishLexicons(ctx, env); err != nil { + return fmt.Errorf("publish lexicons: %w", err) + } + } + // Pick free ports for the node. httpPort, err := freePort() if err != nil { @@ -163,6 +212,19 @@ func runE2E(ctx context.Context, devEnvPath string) error { return err } httpAddr := fmt.Sprintf("127.0.0.1:%d", httpPort) + broadcasterHost := httpAddr + if tlsEnv != nil { + broadcasterHost = httpsStationHost + pdsURL, err := url.Parse(env.PDSURL) + if err != nil { + return fmt.Errorf("parse dev-env pds url: %w", err) + } + plcURL, err := url.Parse(env.PLCURL) + if err != nil { + return fmt.Errorf("parse dev-env plc url: %w", err) + } + tlsEnv.Serve(ctx, pdsURL.Host, plcURL.Host, httpAddr) + } // Fork ourselves as a Streamplace server node. self, err := os.Executable() @@ -191,11 +253,14 @@ func runE2E(ctx context.Context, devEnvPath string) error { fmt.Sprintf("SP_PLC_URL=%s", env.PLCURL), fmt.Sprintf("SP_DATA_DIR=%s", dataDir), fmt.Sprintf("SP_DEV_ACCOUNT_CREDS=%s=%s", out.Did, password), - fmt.Sprintf("SP_BROADCASTER_HOST=127.0.0.1:%d", httpPort), + fmt.Sprintf("SP_BROADCASTER_HOST=%s", broadcasterHost), fmt.Sprintf("SP_WEBSOCKET_URL=ws://%s", httpAddr), "SP_STREAM_SESSION_TIMEOUT=30s", "SP_TRUST_PRIVATE_NETWORK=true", ) + if tlsEnv != nil { + nodeCmd.Env = append(nodeCmd.Env, tlsEnv.NodeEnv()...) + } nodeCmd.Stdout = os.Stderr nodeCmd.Stderr = os.Stderr // Own process group, so cleanup can take the whole tree down at once. @@ -304,11 +369,52 @@ func runE2E(ctx context.Context, devEnvPath string) error { } }) - // Print the env vars for the workflow to consume. - fmt.Printf("SERVER_URL=http://%s\n", httpAddr) - fmt.Printf("ACCOUNT_HANDLE=%s\n", out.Handle) - fmt.Printf("ACCOUNT_DID=%s\n", out.Did) + // Print the env vars for the workflow to consume, in one write: callers + // poll for SERVER_URL and then read the whole file. + vars := fmt.Sprintf("SERVER_URL=http://%s\nACCOUNT_HANDLE=%s\nACCOUNT_DID=%s\nACCOUNT_PASSWORD=%s\n", + httpAddr, out.Handle, out.Did, password) + if tlsEnv != nil { + // The same node over HTTPS at its public name, plus what a browser + // needs to reach and trust it (see e2e_https.go). + vars += fmt.Sprintf("SERVER_HTTPS_URL=%s\nPDS_HTTPS_URL=%s\nE2E_PROXY_URL=%s\nE2E_TLS_SPKI=%s\n", + tlsEnv.StationURL(), tlsEnv.PDSURL(), tlsEnv.ProxyURL(), tlsEnv.spki) + } + fmt.Print(vars) <-ctx.Done() return g.Wait() } + +// publishLexicons writes this build's lexicons, OAuth permission sets +// included, into the dev-env lexicon authority account: the local version of +// what the account behind _lexicon.stream.place holds in production. The PDS +// resolves `include:place.stream.authFull` from it when the app logs in. +func publishLexicons(ctx context.Context, env e2eDevEnv) error { + xrpcc := &xrpc.Client{Host: env.PDSURL, Client: &aqhttp.TrustedClient} + session, err := comatproto.ServerCreateSession(ctx, xrpcc, &comatproto.ServerCreateSession_Input{ + Identifier: env.LexiconDID, + Password: env.LexiconPassword, + }) + if err != nil { + return fmt.Errorf("create session: %w", err) + } + xrpcc.Auth = &xrpc.AuthInfo{Did: session.Did, AccessJwt: session.AccessJwt, RefreshJwt: session.RefreshJwt} + lexs, err := atproto.PublishedLexicons(ctx) + if err != nil { + return err + } + for _, lex := range lexs { + rkey := lex.ID + inp := spcomatproto.RepoCreateRecord_Input{ + Collection: "com.atproto.lexicon.schema", + Repo: session.Did, + Rkey: &rkey, + Record: &glex.LexiconTypeDecoder{Val: &atproto.SchemaFileWrapper{SchemaFile: *lex}}, + } + out := spcomatproto.RepoCreateRecord_Output{} + if err := xrpcc.Do(ctx, xrpc.Procedure, "application/json", "com.atproto.repo.createRecord", map[string]any{}, inp, &out); err != nil { + return fmt.Errorf("%s: %w", lex.ID, err) + } + } + return nil +} diff --git a/pkg/cmd/e2e_https.go b/pkg/cmd/e2e_https.go new file mode 100644 index 000000000..4d2c1957f --- /dev/null +++ b/pkg/cmd/e2e_https.go @@ -0,0 +1,339 @@ +package cmd + +import ( + "bufio" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/base64" + "encoding/pem" + "errors" + "fmt" + "io" + "math/big" + "net" + "net/http" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "stream.place/streamplace/pkg/log" +) + +// e2eHTTPS gives the e2e harness real, locally trusted HTTPS, so the atproto +// OAuth flow can run against the local PDS and node unmodified. Neither end +// accepts less: the Go OAuth client wants a PDS issuer that is https:// with +// no port, the PDS's OAuth provider refuses client IDs and redirect URIs on +// .test/.local/.localhost names, and oatproxy and the browser OAuth client +// both resolve did:plc against a hardcoded https://plc.directory. +// +// So, given public DNS records pointing a PDS hostname, everything under it +// (account handles) and a station (broadcaster) hostname at 127.0.0.1, the +// harness does the following. The two must be on different registrable +// domains, as they are in production: the PDS refuses to show its sign-in +// page to a same-site navigation (Sec-Fetch-Site: same-site). +// +// - mints a throwaway CA and one leaf for all of those and plc.directory; +// - terminates TLS on 127.0.0.1:443 and routes by SNI: the PDS hostname and +// handles to the dev-env PDS, the station hostname to the node's +// plain-HTTP listener (it runs with --behind-https-proxy), plc.directory +// to the dev-env PLC; +// - has the PDS resolve lexicons, the OAuth permission sets among them, +// from a local lexicon authority account that the harness publishes this +// build's lexicons into (see publishLexicons in e2e.go); +// - runs a CONNECT proxy that sends the names above to that listener and +// everything else where it was going. The node (HTTPS_PROXY) and the +// browser use it, which is how their hardcoded plc.directory lookups land +// on the local PLC. +// +// Nothing is installed system-wide: each process is told to trust the CA on +// its own (SSL_CERT_DIR for Go, NODE_EXTRA_CA_CERTS for Node, the leaf's SPKI +// for Chromium), and the CA's key is deleted with the run. +type e2eHTTPS struct { + pdsHost string + stationHost string + dir string + caPath string + trustDir string + cert tls.Certificate + spki string + + frontLn net.Listener // 127.0.0.1:443 + proxyLn net.Listener // CONNECT proxy +} + +const e2ePLCHost = "plc.directory" + +// newE2EHTTPS mints the certificates and claims every listener up front, so a +// harness that cannot bind 443 fails before it has started anything else. +func newE2EHTTPS(pdsHost, stationHost string) (*e2eHTTPS, error) { + h := &e2eHTTPS{pdsHost: pdsHost, stationHost: stationHost} + var err error + h.dir, err = os.MkdirTemp("", "streamplace-e2e-tls-*") + if err != nil { + return nil, err + } + if err := h.mintCerts(); err != nil { + h.Close() + return nil, fmt.Errorf("mint e2e certificates: %w", err) + } + h.frontLn, err = net.Listen("tcp", "127.0.0.1:443") + if err != nil { + h.Close() + return nil, fmt.Errorf("listen on 127.0.0.1:443 (OAuth needs portless https hosts; run as root, e.g. in the build container, or leave the --https-* hostnames empty to skip HTTPS): %w", err) + } + if h.proxyLn, err = net.Listen("tcp", "127.0.0.1:0"); err != nil { + h.Close() + return nil, err + } + return h, nil +} + +func (h *e2eHTTPS) Close() { + for _, ln := range []net.Listener{h.frontLn, h.proxyLn} { + if ln != nil { + ln.Close() //nolint:errcheck + } + } + if h.dir != "" { + os.RemoveAll(h.dir) //nolint:errcheck + } +} + +// StationURL and PDSURL are the public https URLs users visit. +func (h *e2eHTTPS) StationURL() string { return "https://" + h.stationHost } +func (h *e2eHTTPS) PDSURL() string { return "https://" + h.pdsHost } + +func (h *e2eHTTPS) ProxyURL() string { return "http://" + h.proxyLn.Addr().String() } + +// DevEnvEnv is the extra environment for js/dev-env: the PDS's public +// hostname, and trust in our CA for what it fetches over https (the node's +// OAuth client metadata, lexicons from its own public URL). +func (h *e2eHTTPS) DevEnvEnv() []string { + return []string{ + "DEV_ENV_PDS_HOSTNAME=" + h.pdsHost, + "NODE_EXTRA_CA_CERTS=" + h.caPath, + } +} + +// NodeEnv is the extra environment for the forked node, whose broadcaster +// host is the station hostname. Go reads SSL_CERT_DIR on top of the system +// bundle file, so public roots still work. +func (h *e2eHTTPS) NodeEnv() []string { + return []string{ + "SP_BEHIND_HTTPS_PROXY=true", + "SSL_CERT_DIR=" + h.trustDir, + "HTTPS_PROXY=" + h.ProxyURL(), + } +} + +// Serve starts the TLS listeners and the proxy. The backends are the dev-env +// PDS and PLC and the node's plain-HTTP listener, as host:port. +func (h *e2eHTTPS) Serve(ctx context.Context, pdsAddr, plcAddr, nodeAddr string) { + route := func(name string) string { + name = strings.ToLower(strings.TrimSuffix(name, ".")) + switch { + case name == e2ePLCHost: + return plcAddr + case name == h.stationHost: + return nodeAddr + case name == h.pdsHost || strings.HasSuffix(name, "."+h.pdsHost): + return pdsAddr + } + return "" + } + go serveTLSPipe(ctx, h.frontLn, h.cert, route) + go serveConnectProxy(ctx, h.proxyLn, func(hostport string) string { + host, port, err := net.SplitHostPort(hostport) + if err == nil && port == "443" && route(host) != "" { + return h.frontLn.Addr().String() + } + return hostport + }) +} + +func (h *e2eHTTPS) mintCerts() error { + now := time.Now() + caKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return err + } + caTmpl := &x509.Certificate{ + SerialNumber: randomSerial(), + Subject: pkix.Name{CommonName: "Streamplace e2e throwaway CA"}, + NotBefore: now.Add(-time.Hour), + NotAfter: now.Add(7 * 24 * time.Hour), + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageCRLSign, + BasicConstraintsValid: true, + IsCA: true, + MaxPathLenZero: true, + } + caDER, err := x509.CreateCertificate(rand.Reader, caTmpl, caTmpl, &caKey.PublicKey, caKey) + if err != nil { + return err + } + ca, err := x509.ParseCertificate(caDER) + if err != nil { + return err + } + + leafKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return err + } + leafTmpl := &x509.Certificate{ + SerialNumber: randomSerial(), + Subject: pkix.Name{CommonName: h.pdsHost}, + DNSNames: []string{h.pdsHost, "*." + h.pdsHost, h.stationHost, e2ePLCHost}, + NotBefore: now.Add(-time.Hour), + NotAfter: now.Add(7 * 24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + leafDER, err := x509.CreateCertificate(rand.Reader, leafTmpl, ca, &leafKey.PublicKey, caKey) + if err != nil { + return err + } + leaf, err := x509.ParseCertificate(leafDER) + if err != nil { + return err + } + h.cert = tls.Certificate{Certificate: [][]byte{leafDER}, PrivateKey: leafKey, Leaf: leaf} + sum := sha256.Sum256(leaf.RawSubjectPublicKeyInfo) + h.spki = base64.StdEncoding.EncodeToString(sum[:]) + + // The trust dir holds the CA and nothing else: Go parses every file in an + // SSL_CERT_DIR entry. + h.trustDir = filepath.Join(h.dir, "trust") + if err := os.Mkdir(h.trustDir, 0o700); err != nil { + return err + } + h.caPath = filepath.Join(h.trustDir, "ca.pem") + return os.WriteFile(h.caPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: caDER}), 0o600) +} + +func randomSerial() *big.Int { + n, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 127)) + if err != nil { + panic(err) + } + return n +} + +// serveTLSPipe terminates TLS on ln and pipes the plaintext, byte for byte, to +// the backend route picks from the SNI name. Piping rather than proxying HTTP +// keeps websockets working and leaves requests exactly as the client sent +// them, Host header included (the PDS answers handle lookups by Host). +func serveTLSPipe(ctx context.Context, ln net.Listener, cert tls.Certificate, route func(sni string) string) { + tlsLn := tls.NewListener(ln, &tls.Config{ + Certificates: []tls.Certificate{cert}, + MinVersion: tls.VersionTLS12, + }) + for { + conn, err := tlsLn.Accept() + if err != nil { + if ctx.Err() == nil && !errors.Is(err, net.ErrClosed) { + log.Log(ctx, "e2e tls: accept failed", "addr", ln.Addr(), "err", err) + } + return + } + go func() { + tc := conn.(*tls.Conn) + hsCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + if err := tc.HandshakeContext(hsCtx); err != nil { + conn.Close() //nolint:errcheck + return + } + sni := tc.ConnectionState().ServerName + backend := route(sni) + if backend == "" { + log.Log(ctx, "e2e tls: no backend for server name", "sni", sni, "addr", ln.Addr()) + conn.Close() //nolint:errcheck + return + } + upstream, err := net.Dial("tcp", backend) + if err != nil { + log.Log(ctx, "e2e tls: backend dial failed", "sni", sni, "backend", backend, "err", err) + conn.Close() //nolint:errcheck + return + } + pipeConns(conn, upstream) + }() + } +} + +// serveConnectProxy is an HTTP CONNECT proxy whose only trick is resolve, +// which may send a host:port somewhere other than DNS would. +func serveConnectProxy(ctx context.Context, ln net.Listener, resolve func(hostport string) string) { + srv := &http.Server{ + ReadHeaderTimeout: 10 * time.Second, + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodConnect { + http.Error(w, "this proxy only speaks CONNECT", http.StatusMethodNotAllowed) + return + } + upstream, err := net.DialTimeout("tcp", resolve(r.Host), 10*time.Second) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + hj, ok := w.(http.Hijacker) + if !ok { + upstream.Close() //nolint:errcheck + http.Error(w, "cannot hijack", http.StatusInternalServerError) + return + } + conn, buf, err := hj.Hijack() + if err != nil { + upstream.Close() //nolint:errcheck + return + } + if _, err := conn.Write([]byte("HTTP/1.1 200 Connection established\r\n\r\n")); err != nil { + conn.Close() //nolint:errcheck + upstream.Close() //nolint:errcheck + return + } + pipeConns(&bufferedConn{Conn: conn, r: buf.Reader}, upstream) + }), + } + go func() { + <-ctx.Done() + srv.Close() //nolint:errcheck + }() + err := srv.Serve(ln) + if err != nil && !errors.Is(err, http.ErrServerClosed) && !errors.Is(err, net.ErrClosed) && ctx.Err() == nil { + log.Log(ctx, "e2e proxy: serve failed", "err", err) + } +} + +// bufferedConn reads through the bufio.Reader a hijacked connection came +// with, in case the client pipelined bytes behind its CONNECT. +type bufferedConn struct { + net.Conn + r *bufio.Reader +} + +func (c *bufferedConn) Read(p []byte) (int, error) { return c.r.Read(p) } + +// pipeConns copies both ways until either side is done, then closes both. +func pipeConns(a, b net.Conn) { + var once sync.Once + closeBoth := func() { + a.Close() //nolint:errcheck + b.Close() //nolint:errcheck + } + go func() { + io.Copy(a, b) //nolint:errcheck + once.Do(closeBoth) + }() + io.Copy(b, a) //nolint:errcheck + once.Do(closeBoth) +} -- knotmirror 0.0.1