Feat: Add the praxis hop and the egress listener to the forward proxy

.github/workflows/praxis.yaml

@@ -14,6 +14,8 @@ on:
     paths:
       - "praxis/**"
       - ".github/workflows/praxis.yaml"
+      - "core/attribution/wire.go"
+      - "core/attribution/testdata/**"
       - "core/parse/**"
       - "core/internal/parsercommon/**"
       - "core/model/extensions.go"
@@ -37,6 +39,8 @@ on:
     paths:
       - "praxis/**"
       - ".github/workflows/praxis.yaml"
+      - "core/attribution/wire.go"
+      - "core/attribution/testdata/**"
       - "core/parse/**"
       - "core/internal/parsercommon/**"
       - "core/model/extensions.go"

core/attribution/testdata/wire.golden

@@ -0,0 +1 @@
+eyJhZ2VudCI6ImNsYXVkZS1jb2RlIiwicHJvYyI6InAxIiwiY2xpZW50UHJvYyI6InAyIiwic2Vzc2lvbiI6InMxIiwic291cmNlIjoiaGVhZGVyIiwic2Vzc2lvbkhlYWRlciI6IlgtQ2xhdWRlLUNvZGUtU2Vzc2lvbi1JZCIsImNoYWluU2Vzc2lvbiI6InMwIiwicmVjb3JkU2Vzc2lvbiI6InMxIn0

core/attribution/wire.go

@@ -1,6 +1,10 @@
 package attribution
 
-import "net/http"
+import (
+	"encoding/base64"
+	"encoding/json"
+	"net/http"
+)
 
 // The headers an Attribution will travel in: what a hop that resolved a request — a local relay
 // that ran Resolve at accept time — stamps on it for the recorder behind it, which then trusts
@@ -9,7 +13,8 @@ import "net/http"
 //
 // STRIP, THEN INJECT. A client can send these headers too, and one that did would choose its
 // own session and process. So the forward proxy removes them from what the client sent (Strip)
-// today, before any hop is trusted to set them. The encoder and decoder land with that hop.
+// today, before any hop is trusted to set them. The praxis hop sends one HeaderAttribution
+// instead: Wire.Encode is its encoder, and the gateway (praxis/) has the decoder.
 const (
 	HeaderAgent = "X-Cortex-Agent"
 	HeaderProc  = "X-Cortex-Proc"
@@ -31,3 +36,25 @@ func Strip(h http.Header) {
 		h.Del(k)
 	}
 }
+
+// HeaderAttribution carries a Wire from the praxis hop to cortex-gateway.
+const HeaderAttribution = "X-Cortex-Attribution"
+
+// Wire is the attribution result that the hop sends in X-Cortex-Attribution.
+// The JSON keys must stay equal to the Rust Attribution struct in praxis/filters.
+type Wire struct {
+	Agent         string `json:"agent,omitempty"`
+	Proc          string `json:"proc,omitempty"`
+	ClientProc    string `json:"clientProc,omitempty"`
+	Session       string `json:"session,omitempty"`
+	Source        string `json:"source,omitempty"`
+	SessionHeader string `json:"sessionHeader,omitempty"`
+	ChainSession  string `json:"chainSession,omitempty"`
+	RecordSession string `json:"recordSession,omitempty"`
+}
+
+// Encode gives the header value: the JSON of w in base64url with no padding.
+func (w Wire) Encode() string {
+	b, _ := json.Marshal(w) // a struct of strings always marshals
+	return base64.RawURLEncoding.EncodeToString(b)
+}

core/attribution/wire_test.go

@@ -1,7 +1,14 @@
 package attribution
 
 import (
+	"encoding/json"
+	"flag"
 	"net/http"
+	"os"
+	"path/filepath"
+	"slices"
+	"sort"
+	"strings"
 	"testing"
 )
 
@@ -17,3 +24,58 @@ func TestWire_StripRemovesOnlyAttribution(t *testing.T) {
 		t.Errorf("after Strip: %v, want only User-Agent", h)
 	}
 }
+
+var update = flag.Bool("update", false, "rewrite testdata/wire.golden from Wire.Encode")
+
+// golden is the Wire that testdata/wire.golden holds. The Rust test decodes the file and
+// compares it with these values.
+var golden = Wire{Agent: "claude-code", Proc: "p1", ClientProc: "p2", Session: "s1", Source: "header",
+	SessionHeader: "X-Claude-Code-Session-Id", ChainSession: "s0", RecordSession: "s1"}
+
+// The encoding of golden must equal the file that the gateway's decoder test reads.
+func TestWire_EncodeGolden(t *testing.T) {
+	path := filepath.Join("testdata", "wire.golden")
+	got := golden.Encode()
+	if *update {
+		if err := os.WriteFile(path, []byte(got+"\n"), 0o644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	want, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got != strings.TrimSpace(string(want)) {
+		t.Errorf("Encode() = %q, want %q", got, want)
+	}
+}
+
+// The JSON keys must be the camelCase names of the Rust Attribution struct.
+func TestWire_KeysMatchRust(t *testing.T) {
+	b, err := json.Marshal(golden)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m map[string]any
+	if err := json.Unmarshal(b, &m); err != nil {
+		t.Fatal(err)
+	}
+	var keys []string
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	want := []string{"agent", "chainSession", "clientProc", "proc", "recordSession", "session", "sessionHeader", "source"}
+	if !slices.Equal(keys, want) {
+		t.Errorf("keys = %v, want %v", keys, want)
+	}
+}
+
+// The Rust decoder reads base64url with no padding.
+func TestWire_EncodeIsUnpadded(t *testing.T) {
+	for _, w := range []Wire{golden, {Agent: "a"}, {Agent: "ab"}, {}} {
+		if s := w.Encode(); strings.Contains(s, "=") {
+			t.Errorf("Encode(%+v) = %q has padding", w, s)
+		}
+	}
+}

core/listener/forwardproxy/egress.go

@@ -0,0 +1,95 @@
+package forwardproxy
+
+import (
+	"errors"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+
+	"github.com/rossoctl/cortex/core/attribution"
+	"github.com/rossoctl/cortex/core/listener/egressguard"
+	"github.com/rossoctl/cortex/core/listener/httpx"
+	"github.com/rossoctl/cortex/core/model"
+	"github.com/rossoctl/cortex/core/pipeline"
+)
+
+// NewEgressHandler is the praxis-mode egress listener (spec Section 4, throwaway at phase B).
+//
+// cortex-gateway forwards each recorded request here with X-Cortex-Dest and
+// X-Cortex-Token. The handler checks the token, dials the origin that Dest
+// names through upstream, and relays the response with no buffering.
+func NewEgressHandler(token string, upstream http.RoundTripper) http.Handler {
+	rp := &httputil.ReverseProxy{
+		Transport:     upstream,
+		FlushInterval: -1,
+		Rewrite: func(pr *httputil.ProxyRequest) {
+			dest, _ := parseEgressDest(pr.In.Header.Get(headerDest)) // ServeHTTP checked it
+			pr.Out.URL.Scheme = dest.Scheme
+			pr.Out.URL.Host = dest.Host
+			pr.Out.Host = pr.In.Host
+			// Every hop and attribution header, not only Dest and Token: an origin must not
+			// get the session id or process names when the gateway's strip list is wrong.
+			stripHopHeaders(pr.Out.Header)
+			attribution.Strip(pr.Out.Header)
+			pr.Out.Header.Del(headerEgressError)
+		},
+		// The kind header is the egress listener's own signal. An origin must not set it.
+		ModifyResponse: func(resp *http.Response) error {
+			resp.Header.Del(headerEgressError)
+			return nil
+		},
+		ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
+			dest := pipeline.SanitizeLogValue(r.Header.Get(headerDest))
+			var refusal *egressguard.Refusal
+			if errors.As(err, &refusal) {
+				slog.Warn("egress: destination refused", "dest", dest, "error", pipeline.SanitizeLogValue(err.Error()))
+				w.Header().Set(headerEgressError, "destination_private")
+				// The same body as direct mode (serveOutbound), so the client sees one refusal shape.
+				httpx.WriteRejection(w, pipeline.DenyStatus(http.StatusForbidden, "destination_private",
+					fmt.Sprintf("refusing to reach private destination %s (%s)", refusal.Host, refusal.Addr)))
+				return
+			}
+			if r.Context().Err() != nil {
+				return // the client went away; nobody reads a status
+			}
+			fail := model.TransportError(err)
+			slog.Warn("egress: upstream failed", "dest", dest, "kind", fail.Kind, "error", pipeline.SanitizeLogValue(fail.Message))
+			w.Header().Set(headerEgressError, fail.Kind)
+			w.WriteHeader(http.StatusBadGateway)
+		},
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if !tokenMatches(r.Header.Get(headerToken), token) {
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+		raw := r.Header.Get(headerDest)
+		if raw == probeDest {
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		if _, ok := parseEgressDest(raw); !ok {
+			slog.Warn("egress: bad destination", "dest", pipeline.SanitizeLogValue(raw))
+			http.Error(w, "bad "+headerDest, http.StatusBadRequest)
+			return
+		}
+		rp.ServeHTTP(w, r)
+	})
+}
+
+// parseEgressDest accepts only "http://host:port" and "https://host:port" (Decision 1).
+func parseEgressDest(raw string) (*url.URL, bool) {
+	u, err := url.Parse(raw)
+	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.User != nil ||
+		u.Path != "" || u.RawQuery != "" || u.Fragment != "" || u.Opaque != "" {
+		return nil, false
+	}
+	host, port, err := net.SplitHostPort(u.Host)
+	if err != nil || host == "" || port == "" {
+		return nil, false
+	}
+	return u, true
+}