✨ feat(forwardproxy): support HTTPS CONNECT as raw TCP passthrough

authbridge/authlib/listener/forwardproxy/server.go

@@ -168,7 +168,7 @@ func (s *Server) Handler() http.Handler {
 
 func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
 	if r.Method == http.MethodConnect {
-		http.Error(w, `{"error":"HTTPS CONNECT not supported — only HTTP proxy"}`, http.StatusMethodNotAllowed)
+		s.handleConnect(w, r)
 		return
 	}
 
@@ -404,3 +404,148 @@ func (s *Server) recordOutboundReject(pctx *pipeline.Context, action pipeline.Ac
 	}
 	s.Sessions.Append(sid, ev)
 }
+
+// connectDialTimeout bounds the upstream TCP dial for a CONNECT tunnel.
+// Once the tunnel is open the timeout no longer applies — the agent's TLS
+// handshake and subsequent traffic flow at their own pace.
+const connectDialTimeout = 30 * time.Second
+
+// handleConnect tunnels HTTPS (and any other TLS-wrapped protocol) through
+// the forward proxy as raw TCP. Mirrors the TLS-passthrough behavior of
+// envoy-sidecar mode: bytes are opaque to the proxy, so token-exchange and
+// the protocol parsers (mcp-parser, inference-parser) are no-ops by
+// definition. Pipeline gates (ibac, jwt-validation bypass logic, etc.)
+// still run on the CONNECT request itself so they can reject based on
+// destination host before the tunnel opens.
+//
+// mTLS is intentionally NOT applied to the upstream dial — the bytes
+// flowing through this tunnel ARE the agent's own end-to-end TLS, and
+// terminating that with sidecar-to-sidecar mTLS would break the agent's
+// trust path. CONNECT targets are opaque externals (LiteMaaS, Bedrock,
+// GitHub API, etc.) where the agent's existing TLS is the right answer.
+func (s *Server) handleConnect(w http.ResponseWriter, r *http.Request) {
+	pctx := &pipeline.Context{
+		Direction: pipeline.Outbound,
+		Scheme:    "tcp", // marker: bytes are opaque, not HTTP
+		Host:      r.Host,
+		Path:      "",
+		Headers:   r.Header.Clone(),
+		StartedAt: time.Now(),
+	}
+	defer func() {
+		s.OutboundPipeline.RunFinish(r.Context(), pctx, pipeline.OutcomeFromContext(pctx))
+	}()
+
+	if s.Sessions != nil {
+		if aid := s.Sessions.ActiveSession(); aid != "" {
+			pctx.Session = s.Sessions.View(aid)
+		}
+	}
+
+	// Run the outbound pipeline. Plugins that policy on host/identity
+	// (ibac, content gates) still get to allow/deny; plugins that need
+	// HTTP body (parsers) see no body, which they handle gracefully.
+	action := s.OutboundPipeline.Run(r.Context(), pctx)
+	if action.Type == pipeline.Reject {
+		s.recordOutboundReject(pctx, action)
+		httpx.WriteRejection(w, action)
+		return
+	}
+
+	// Verify hijack capability BEFORE dialing upstream. If hijacking
+	// isn't supported the failure mode should be a 500 to the client,
+	// not a half-opened TCP connection to the upstream. The actual
+	// Hijack() call happens after dial succeeds — http.Error needs an
+	// un-hijacked ResponseWriter to deliver the dial-failure 502.
+	if _, ok := w.(http.Hijacker); !ok {
+		slog.Error("forward-proxy: ResponseWriter does not support hijacking", "host", r.Host)
+		http.Error(w, `{"error":"connect not supported by listener"}`, http.StatusInternalServerError)
+		return
+	}
+
+	// Plain TCP dial. See package-level comment on why mTLS doesn't
+	// apply here. r.Host on a CONNECT carries "host:port" already.
+	upstream, err := net.DialTimeout("tcp", r.Host, connectDialTimeout)
+	if err != nil {
+		slog.Warn("forward-proxy: CONNECT upstream dial failed", "host", r.Host, "error", err)
+		http.Error(w, `{"error":"bad gateway"}`, http.StatusBadGateway)
+		return
+	}
+
+	clientConn, _, err := w.(http.Hijacker).Hijack()
+	if err != nil {
+		_ = upstream.Close()
+		slog.Error("forward-proxy: CONNECT hijack failed", "host", r.Host, "error", err)
+		return
+	}
+
+	// TCP keepalive on both ends. Streaming LLM completions can hold
+	// the tunnel open for minutes; without keepalives, a vanished peer
+	// (network partition, NAT entry expiry, peer reboot) parks the
+	// io.Copy goroutines until the OS finally times the socket out.
+	// 30s is loose enough to not perturb idle traffic and tight enough
+	// that operators get prompt cleanup on dead connections.
+	enableKeepalive(upstream)
+	enableKeepalive(clientConn)
+
+	// Tell the agent the tunnel is up. Per RFC 7231 §4.3.6 a 200 to
+	// CONNECT signals "tunnel established"; the body is empty and any
+	// subsequent bytes from either side are application data.
+	if _, err := clientConn.Write([]byte("HTTP/1.1 200 Connection Established\r\n\r\n")); err != nil {
+		_ = clientConn.Close()
+		_ = upstream.Close()
+		slog.Debug("forward-proxy: CONNECT 200 write failed", "host", r.Host, "error", err)
+		return
+	}
+
+	// Record a SessionRequest event so /v1/sessions and abctl show that
+	// a tunnel was opened. Mirrors the HTTP path's post-Allow recording
+	// (see handleRequest above). The MCP / Inference snapshots are nil
+	// by definition (CONNECT bytes are opaque), but Invocations from
+	// gate plugins (ibac, token-exchange's skip/no_route, etc.) and
+	// any plugin-public Plugins entries are still meaningful.
+	if s.Sessions != nil {
+		sid := s.Sessions.ActiveSession()
+		if sid == "" {
+			sid = session.DefaultSessionID
+		}
+		plugins := pipeline.SnapshotPlugins(pctx.Extensions.Custom)
+		ev := pipeline.SessionEvent{
+			At:          time.Now(),
+			Direction:   pipeline.Outbound,
+			Phase:       pipeline.SessionRequest,
+			Invocations: pipeline.SnapshotInvocations(pctx.Extensions.Invocations, pipeline.InvocationPhaseRequest),
+			Plugins:     plugins,
+			Identity:    pipeline.SnapshotIdentity(pctx),
+			Host:        pctx.Host,
+		}
+		if ev.Invocations != nil || plugins != nil {
+			s.Sessions.Append(sid, ev)
+		}
+	}
+
+	// Bidirectional copy. When either side closes, propagate the close
+	// to the other so both io.Copy goroutines exit. Close-on-each-side
+	// is idempotent on net.Conn.
+	go func() {
+		_, _ = io.Copy(upstream, clientConn)
+		_ = upstream.Close()
+		_ = clientConn.Close()
+	}()
+	_, _ = io.Copy(clientConn, upstream)
+	_ = clientConn.Close()
+	_ = upstream.Close()
+}
+
+// enableKeepalive turns on TCP keepalive with a 30s probe interval on
+// the underlying *net.TCPConn, if conn unwraps to one. No-op on other
+// connection types (notably *tls.Conn, which doesn't apply on the
+// CONNECT path since the bytes through the tunnel are already TLS).
+func enableKeepalive(conn net.Conn) {
+	tcp, ok := conn.(*net.TCPConn)
+	if !ok {
+		return
+	}
+	_ = tcp.SetKeepAlive(true)
+	_ = tcp.SetKeepAlivePeriod(30 * time.Second)
+}

authbridge/authlib/listener/forwardproxy/server_test.go

@@ -1,9 +1,11 @@
 package forwardproxy
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -96,7 +98,42 @@ func TestForwardProxy_Exchange(t *testing.T) {
 	}
 }
 
-func TestForwardProxy_CONNECT_Rejected(t *testing.T) {
+// TestForwardProxy_CONNECT_TunnelsBytes asserts that CONNECT opens a raw
+// TCP tunnel to the upstream and shuttles bytes in both directions. This
+// is the basis for HTTPS passthrough — the agent's TLS client and the
+// upstream TLS server complete their handshake through the proxy with
+// the proxy never inspecting (or being able to inspect) the encrypted
+// bytes. Mirrors envoy-sidecar's TLS-passthrough filter chain.
+func TestForwardProxy_CONNECT_TunnelsBytes(t *testing.T) {
+	// Bare TCP echo — stand-in for any TLS server. We only need to
+	// prove that bytes the agent writes reach the upstream and bytes
+	// the upstream writes reach the agent. Loop until the client
+	// closes so we can exercise multiple round-trips, which is closer
+	// to what a real TLS handshake (multi-roundtrip, interleaved
+	// reads and writes) does over the tunnel.
+	upstream, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen upstream: %v", err)
+	}
+	defer upstream.Close()
+	go func() {
+		conn, err := upstream.Accept()
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		buf := make([]byte, 1024)
+		for {
+			n, err := conn.Read(buf)
+			if err != nil {
+				return
+			}
+			if _, err := conn.Write(append([]byte("echo:"), buf[:n]...)); err != nil {
+				return
+			}
+		}
+	}()
+
 	a := auth.New(auth.Config{})
 	srv, err := NewServer(outboundPipelineFromAuth(t, a), nil, nil)
 	if err != nil {
@@ -105,13 +142,132 @@ func TestForwardProxy_CONNECT_Rejected(t *testing.T) {
 	proxy := httptest.NewServer(srv.Handler())
 	defer proxy.Close()
 
-	req, _ := http.NewRequest("CONNECT", proxy.URL, nil)
-	resp, err := http.DefaultClient.Do(req)
+	// Open raw TCP to the proxy and speak HTTP CONNECT directly so we
+	// can drive the bytes manually (net/http's client wraps everything
+	// up too tightly for this kind of test).
+	proxyAddr := strings.TrimPrefix(proxy.URL, "http://")
+	tunnel, err := net.DialTimeout("tcp", proxyAddr, 5*time.Second)
 	if err != nil {
-		t.Fatalf("request failed: %v", err)
+		t.Fatalf("dial proxy: %v", err)
+	}
+	defer tunnel.Close()
+
+	target := upstream.Addr().String()
+	if _, err := tunnel.Write([]byte("CONNECT " + target + " HTTP/1.1\r\nHost: " + target + "\r\n\r\n")); err != nil {
+		t.Fatalf("write CONNECT: %v", err)
+	}
+
+	// Read the proxy's "200 Connection Established" status line + empty headers.
+	br := bufio.NewReader(tunnel)
+	line, err := br.ReadString('\n')
+	if err != nil {
+		t.Fatalf("read CONNECT response: %v", err)
+	}
+	if !strings.Contains(line, "200") {
+		t.Fatalf("CONNECT response = %q, want 200", line)
+	}
+	// Drain the remaining response headers up to the empty line.
+	for {
+		hdr, err := br.ReadString('\n')
+		if err != nil {
+			t.Fatalf("read headers: %v", err)
+		}
+		if hdr == "\r\n" || hdr == "\n" {
+			break
+		}
+	}
+
+	// Tunnel is up. Drive multiple round-trips to model the multi-RTT
+	// nature of a real TLS handshake. A single write/read would catch
+	// basic plumbing but miss half-duplex regressions in the
+	// bidirectional copy.
+	for _, payload := range []string{"hello", "world", "third"} {
+		if _, err := tunnel.Write([]byte(payload)); err != nil {
+			t.Fatalf("write %q through tunnel: %v", payload, err)
+		}
+		got := make([]byte, 32)
+		_ = tunnel.SetReadDeadline(time.Now().Add(2 * time.Second))
+		n, err := br.Read(got)
+		if err != nil && err != io.EOF {
+			t.Fatalf("read %q response: %v", payload, err)
+		}
+		want := "echo:" + payload
+		if string(got[:n]) != want {
+			t.Errorf("tunnel response for %q = %q, want %q", payload, got[:n], want)
+		}
+	}
+}
+
+// TestForwardProxy_CONNECT_PipelineDeny asserts that a pipeline reject
+// fires BEFORE the upstream is dialed and produces an HTTP error to the
+// CONNECT-issuing client. Plugins like ibac depend on this — they must
+// be able to deny based on destination host before bytes start flowing.
+func TestForwardProxy_CONNECT_PipelineDeny(t *testing.T) {
+	router, _ := routing.NewRouter("exchange", []routing.Route{})
+	a := auth.New(auth.Config{
+		Router:        router,
+		NoTokenPolicy: auth.NoTokenPolicyDeny, // any outbound w/o token denies
+	})
+	srv, err := NewServer(outboundPipelineFromAuth(t, a), nil, nil)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	proxy := httptest.NewServer(srv.Handler())
+	defer proxy.Close()
+
+	// CONNECT request without an Authorization header — token-exchange's
+	// no-token-policy denies it. Use raw TCP since net/http's client may
+	// retry or buffer in ways that obscure the response.
+	proxyAddr := strings.TrimPrefix(proxy.URL, "http://")
+	conn, err := net.DialTimeout("tcp", proxyAddr, 5*time.Second)
+	if err != nil {
+		t.Fatalf("dial proxy: %v", err)
+	}
+	defer conn.Close()
+	if _, err := conn.Write([]byte("CONNECT example.com:443 HTTP/1.1\r\nHost: example.com:443\r\n\r\n")); err != nil {
+		t.Fatalf("write CONNECT: %v", err)
+	}
+	br := bufio.NewReader(conn)
+	line, err := br.ReadString('\n')
+	if err != nil {
+		t.Fatalf("read response: %v", err)
+	}
+	if strings.Contains(line, "200") {
+		t.Errorf("CONNECT got 200 status line %q, expected pipeline-driven rejection", line)
+	}
+}
+
+// TestForwardProxy_CONNECT_BadGatewayOnDialFailure asserts that a CONNECT
+// to an unreachable upstream produces 502, not a half-opened tunnel.
+// The handler must respond before hijacking — once hijacked, http.Error
+// is no-op-ish and the agent gets a confusing connection reset.
+func TestForwardProxy_CONNECT_BadGatewayOnDialFailure(t *testing.T) {
+	a := auth.New(auth.Config{})
+	srv, err := NewServer(outboundPipelineFromAuth(t, a), nil, nil)
+	if err != nil {
+		t.Fatalf("NewServer: %v", err)
+	}
+	proxy := httptest.NewServer(srv.Handler())
+	defer proxy.Close()
+
+	// 127.0.0.1:1 is virtually guaranteed to refuse on a normal host.
+	proxyAddr := strings.TrimPrefix(proxy.URL, "http://")
+	conn, err := net.DialTimeout("tcp", proxyAddr, 5*time.Second)
+	if err != nil {
+		t.Fatalf("dial proxy: %v", err)
+	}
+	defer conn.Close()
+	if _, err := conn.Write([]byte("CONNECT 127.0.0.1:1 HTTP/1.1\r\nHost: 127.0.0.1:1\r\n\r\n")); err != nil {
+		t.Fatalf("write CONNECT: %v", err)
+	}
+	br := bufio.NewReader(conn)
+	_ = conn.SetReadDeadline(time.Now().Add(connectDialTimeout + 5*time.Second))
+	line, err := br.ReadString('\n')
+	if err != nil {
+		t.Fatalf("read response: %v", err)
 	}
-	if resp.StatusCode != http.StatusMethodNotAllowed {
-		t.Errorf("status = %d, want 405", resp.StatusCode)
+	if !strings.Contains(line, "502") {
+		t.Errorf("CONNECT response = %q, want 502 on upstream dial failure", line)
 	}
 }