mudler--localai/core/services/cloudproxy/mitm/handler_test.go

package mitm

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/http/httptest"
	"net/url"
	"strings"

	"github.com/mudler/LocalAI/core/services/routing/pii"

	. "github.com/onsi/ginkgo/v2"
	. "github.com/onsi/gomega"
)

// substringDetector is a deterministic pii.NERDetector for tests: it
// reports an entity for every occurrence of each configured substring,
// with byte offsets into the scanned text. Lets the MITM tests drive
// request redaction without a real token-classification backend.
type substringDetector struct{ groups map[string]string } // substring -> entity group

func (d substringDetector) Detect(_ context.Context, text string) ([]pii.NEREntity, error) {
	var out []pii.NEREntity
	for sub, group := range d.groups {
		for idx := 0; ; {
			i := strings.Index(text[idx:], sub)
			if i < 0 {
				break
			}
			start := idx + i
			out = append(out, pii.NEREntity{Group: group, Start: start, End: start + len(sub), Score: 1})
			idx = start + len(sub)
		}
	}
	return out, nil
}

// testDetectorCfg flags emails (mask) and a known secret token (block).
func testDetectorCfg() pii.NERConfig {
	return pii.NERConfig{
		Detector: substringDetector{groups: map[string]string{
			"alice@example.com":                 "EMAIL",
			"bob@example.org":                   "EMAIL",
			"sk-abcdefghijklmnopqrstuvwxyz1234": "PASSWORD",
		}},
		EntityActions: map[string]pii.Action{"EMAIL": pii.ActionMask, "PASSWORD": pii.ActionBlock},
	}
}

// startPIITestRig plugs the production PII handler into a CONNECT proxy,
// with the upstream playing the role of api.anthropic.com. Request
// bodies bound for api.anthropic.com run through the NER detector above.
func startPIITestRig(upstream http.Handler) (*http.Client, string, *fakeStore, func()) {
	ts := httptest.NewTLSServer(upstream)
	upstreamCertPool := x509.NewCertPool()
	upstreamCertPool.AddCert(ts.Certificate())
	upstreamURL, _ := url.Parse(ts.URL)
	store := &fakeStore{}

	ca, err := NewInMemoryCA()
	ExpectWithOffset(1, err).NotTo(HaveOccurred())

	upstreamHost := upstreamURL.Host
	prodHandler := NewPIIHandler(PIIHandlerOptions{
		DetectorsByHost: map[string][]pii.NERConfig{
			"api.anthropic.com": {testDetectorCfg()},
		},
		EventStore: store,
		UpstreamTLS: &tls.Config{
			RootCAs:    upstreamCertPool,
			ServerName: "example.com",
		},
		DialHost: func(_ string) string { return upstreamHost },
	})

	srv, err := NewServer(Config{
		Addr:           "127.0.0.1:0",
		CA:             ca,
		InterceptHosts: []string{"api.anthropic.com"},
		Handler:        prodHandler,
		EventStore:     store,
	})
	ExpectWithOffset(1, err).NotTo(HaveOccurred())
	ExpectWithOffset(1, srv.Start()).To(Succeed())

	clientPool := x509.NewCertPool()
	clientPool.AddCert(ca.Cert())
	proxyURL, _ := url.Parse("http://" + srv.Addr())
	client := &http.Client{
		Transport: &http.Transport{
			Proxy:           http.ProxyURL(proxyURL),
			TLSClientConfig: &tls.Config{RootCAs: clientPool},
		},
	}

	cleanup := func() {
		srv.Stop()
		ts.Close()
	}
	return client, "https://api.anthropic.com", store, cleanup
}

type fakeStore struct{ events []pii.PIIEvent }

func (s *fakeStore) Record(_ context.Context, ev pii.PIIEvent) error {
	s.events = append(s.events, ev)
	return nil
}

func (s *fakeStore) List(_ context.Context, _ pii.ListQuery) ([]pii.PIIEvent, error) {
	return s.events, nil
}

func (s *fakeStore) Count(_ context.Context) (int, error) { return len(s.events), nil }
func (s *fakeStore) Close() error                         { return nil }

func (s *fakeStore) recorded() int { return len(s.events) }

var _ = Describe("PIIHandler", func() {
	It("redacts request email via NER", func() {
		var receivedBody []byte
		upstream := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			receivedBody, _ = io.ReadAll(r.Body)
			w.Header().Set("Content-Type", "application/json")
			_, _ = fmt.Fprint(w, `{"id":"msg_x","content":[{"type":"text","text":"ok"}]}`)
		})

		client, base, store, cleanup := startPIITestRig(upstream)
		defer cleanup()

		body := `{"model":"claude-3-5-sonnet","max_tokens":100,"messages":[{"role":"user","content":"my email is alice@example.com please reply"}]}`
		resp, err := client.Post(base+"/v1/messages", "application/json", strings.NewReader(body))
		Expect(err).NotTo(HaveOccurred(), "client.Post")
		defer func() { _ = resp.Body.Close() }()
		Expect(resp.StatusCode).To(Equal(200))

		Expect(string(receivedBody)).NotTo(ContainSubstring("alice@example.com"), "upstream received unredacted body")
		Expect(string(receivedBody)).To(ContainSubstring("[REDACTED:ner:EMAIL]"), "upstream did not see redaction marker")
		Expect(store.recorded()).NotTo(BeZero(), "no PIIEvent recorded for the email match")
	})

	It("refuses to follow an upstream redirect", func() {
		upstream := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			http.Redirect(w, r, "https://evil.example.com/steal", http.StatusFound)
		})

		client, base, _, cleanup := startPIITestRig(upstream)
		defer cleanup()

		body := `{"model":"claude-3-5-sonnet","max_tokens":100,"messages":[{"role":"user","content":"hello"}]}`
		resp, err := client.Post(base+"/v1/messages", "application/json", strings.NewReader(body))
		Expect(err).NotTo(HaveOccurred(), "client.Post")
		defer func() { _ = resp.Body.Close() }()
		Expect(resp.StatusCode).To(Equal(http.StatusBadGateway), "refused redirect must surface as 502, not be followed")
	})

	It("blocks a detected secret in the request", func() {
		upstreamCalled := false
		upstream := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			upstreamCalled = true
			w.WriteHeader(200)
		})

		client, base, _, cleanup := startPIITestRig(upstream)
		defer cleanup()

		body := `{"model":"claude-3-5-sonnet","max_tokens":100,"messages":[{"role":"user","content":"my key is sk-abcdefghijklmnopqrstuvwxyz1234"}]}`
		resp, err := client.Post(base+"/v1/messages", "application/json", strings.NewReader(body))
		Expect(err).NotTo(HaveOccurred(), "client.Post")
		defer func() { _ = resp.Body.Close() }()
		Expect(resp.StatusCode).To(Equal(400), "PASSWORD entity action is block")
		Expect(upstreamCalled).To(BeFalse(), "upstream was called despite block — proxy should short-circuit")
		body2, _ := io.ReadAll(resp.Body)
		Expect(string(body2)).To(ContainSubstring("pii_blocked"))
	})

	It("non-chat path passes through", func() {
		var receivedBody []byte
		upstream := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			receivedBody, _ = io.ReadAll(r.Body)
			w.Header().Set("Content-Type", "application/json")
			_, _ = fmt.Fprint(w, `{"ok":true}`)
		})

		client, base, _, cleanup := startPIITestRig(upstream)
		defer cleanup()

		body := `{"email":"alice@example.com"}`
		resp, err := client.Post(base+"/oauth/callback", "application/json", strings.NewReader(body))
		Expect(err).NotTo(HaveOccurred())
		defer func() { _ = resp.Body.Close() }()
		Expect(string(receivedBody)).To(Equal(body), "body forwarded with mutation")
	})
})

var _ = Describe("redactRequest", func() {
	It("handles anthropic shape", func() {
		body := []byte(`{"model":"claude","max_tokens":10,"messages":[{"role":"user","content":"reach me at bob@example.org"}]}`)

		d := &piiDispatcher{}
		out, blocked, err := d.redactRequest(context.Background(), body, shapeAnthropicMessages, []pii.NERConfig{testDetectorCfg()}, "corr-1")
		Expect(err).NotTo(HaveOccurred())
		Expect(blocked).To(BeFalse(), "EMAIL is mask, not block — blocked should be false")
		var parsed map[string]any
		Expect(json.Unmarshal(out, &parsed)).To(Succeed())
		msgs := parsed["messages"].([]any)
		first := msgs[0].(map[string]any)
		content, _ := first["content"].(string)
		Expect(content).NotTo(ContainSubstring("bob@example.org"), "redaction did not run")
	})
})

var _ = Describe("Proxy events", func() {
	It("emits connect and traffic events", func() {
		upstream := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			w.Header().Set("Content-Type", "application/json")
			_, _ = fmt.Fprint(w, `{"id":"msg_x","content":[{"type":"text","text":"ok"}]}`)
		})

		client, base, store, cleanup := startPIITestRig(upstream)
		defer cleanup()

		body := `{"model":"claude-3-5-sonnet","max_tokens":10,"messages":[{"role":"user","content":"hi"}]}`
		resp, err := client.Post(base+"/v1/messages", "application/json", strings.NewReader(body))
		Expect(err).NotTo(HaveOccurred(), "client.Post")
		defer func() { _ = resp.Body.Close() }()
		_, _ = io.Copy(io.Discard, resp.Body)

		var connect, traffic *pii.PIIEvent
		for i := range store.events {
			ev := &store.events[i]
			switch ev.ResolvedKind() {
			case pii.KindProxyConnect:
				connect = ev
			case pii.KindProxyTraffic:
				traffic = ev
			}
		}

		Expect(connect).NotTo(BeNil(), "no proxy_connect event recorded")
		Expect(connect.Host).To(Equal("api.anthropic.com"))
		Expect(connect.Intercepted).NotTo(BeNil())
		Expect(*connect.Intercepted).To(BeTrue(), "connect.Intercepted should be true for an allowlisted host")

		Expect(traffic).NotTo(BeNil(), "no proxy_traffic event recorded")
		Expect(traffic.Host).To(Equal("api.anthropic.com"))
		Expect(traffic.BytesSent).To(BeNumerically(">", 0))
		Expect(traffic.BytesReceived).To(BeNumerically(">", 0))
		Expect(traffic.StatusCode).To(Equal(200))
	})

	It("tunneled host emits connect event only", func() {
		upstream := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			_, _ = fmt.Fprint(w, "passthrough")
		})
		ts := httptest.NewTLSServer(upstream)
		defer ts.Close()
		upstreamURL, _ := url.Parse(ts.URL)
		upstreamHost, _, _ := net.SplitHostPort(upstreamURL.Host)

		ca, _ := NewInMemoryCA()
		store := &fakeStore{}
		srv, err := NewServer(Config{
			Addr:           "127.0.0.1:0",
			CA:             ca,
			InterceptHosts: []string{"some-other-host"},
			Handler:        func(w http.ResponseWriter, r *http.Request, h string) {},
			EventStore:     store,
		})
		Expect(err).NotTo(HaveOccurred())
		Expect(srv.Start()).To(Succeed())
		defer srv.Stop()

		upstreamCertPool := x509.NewCertPool()
		upstreamCertPool.AddCert(ts.Certificate())
		proxyURL, _ := url.Parse("http://" + srv.Addr())
		client := &http.Client{
			Transport: &http.Transport{
				Proxy: http.ProxyURL(proxyURL),
				TLSClientConfig: &tls.Config{
					RootCAs:    upstreamCertPool,
					ServerName: upstreamHost,
				},
			},
		}
		resp, err := client.Get(ts.URL)
		Expect(err).NotTo(HaveOccurred(), "Get through tunnel")
		_ = resp.Body.Close()

		var connect *pii.PIIEvent
		for i := range store.events {
			ev := &store.events[i]
			Expect(ev.ResolvedKind()).NotTo(Equal(pii.KindProxyTraffic), "unexpected proxy_traffic event for tunneled host: %+v", ev)
			if ev.ResolvedKind() == pii.KindProxyConnect {
				connect = ev
			}
		}
		Expect(connect).NotTo(BeNil(), "no proxy_connect event recorded for tunneled host")
		Expect(connect.Intercepted).NotTo(BeNil())
		Expect(*connect.Intercepted).To(BeFalse(), "connect.Intercepted should be false (tunneled)")
		Expect(connect.Host).NotTo(BeEmpty())
	})
})

var _ = Describe("classifyRequestShape", func() {
	cases := []struct {
		host string
		path string
		want requestShape
	}{
		{"api.anthropic.com", "/v1/messages", shapeAnthropicMessages},
		{"api.openai.com", "/v1/chat/completions", shapeOpenAIChat},
		{"api.anthropic.com", "/v1/oauth/token", shapeUnknown},
		{"api.openai.com", "/v1/embeddings", shapeUnknown},
		{"example.com", "/v1/messages", shapeUnknown},
	}
	for _, c := range cases {
		It(fmt.Sprintf("classifies (%q, %q)", c.host, c.path), func() {
			Expect(classifyRequestShape(c.host, c.path)).To(Equal(c.want))
		})
	}
})