diff --git a/authbridge/authlib/listener/forwardproxy/server.go b/authbridge/authlib/listener/forwardproxy/server.go index 439259a2d..1b6e1836e 100644 --- a/authbridge/authlib/listener/forwardproxy/server.go +++ b/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) +} diff --git a/authbridge/authlib/listener/forwardproxy/server_test.go b/authbridge/authlib/listener/forwardproxy/server_test.go index 2f7c1053b..3c2f8bf0a 100644 --- a/authbridge/authlib/listener/forwardproxy/server_test.go +++ b/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) } }