fix: Address review feedback for SSE streaming and dynamic upstreams

docs/config.md

@@ -1230,10 +1230,14 @@ Authorization: Bearer <agent-key>
 
 Patterns use glob-style matching with scheme enforcement:
 
+- `*` matches exactly **one** hostname segment (between dots)
+- `**` matches **one or more** hostname segments (across dots)
+
 | Pattern | Matches | Does Not Match |
 |---------|---------|----------------|
 | `https://api.example.com` | `https://api.example.com/any/path` | `http://api.example.com` (scheme mismatch) |
-| `https://*.googleapis.com` | `https://storage.googleapis.com` | `https://googleapis.com` (no subdomain) |
+| `https://*.googleapis.com` | `https://storage.googleapis.com` | `https://googleapis.com` (no subdomain), `https://evil.com.googleapis.com` (`*` is single segment) |
+| `https://**.googleapis.com` | `https://storage.googleapis.com`, `https://intended.com.googleapis.com` | `https://googleapis.com` (`**` requires at least one segment) |
 | `https://api.example.com/v1` | `https://api.example.com/v1/users` | `https://api.example.com/v1-admin` (path segment boundary) |
 
 Hostname matching is case-insensitive. Path matching enforces segment boundaries (`/v1` matches `/v1/foo` but not `/v1-admin`).

internal/proxy/sse.go

@@ -20,7 +20,6 @@ type sseFilterReader struct {
 }
 
 func (r *sseFilterReader) Read(p []byte) (int, error) {
-	// Return buffered data first
 	if r.buf.Len() > 0 {
 		return r.buf.Read(p)
 	}
@@ -29,21 +28,18 @@ func (r *sseFilterReader) Read(p []byte) (int, error) {
 		return 0, io.EOF
 	}
 
-	// Initialize scanner on first read
 	if r.scanner == nil {
 		r.scanner = bufio.NewScanner(r.reader)
 		r.scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) // 1MB max line
 	}
 
-	// Accumulate lines until we have a complete SSE message (blank line delimiter)
 	var lines []string
 	gotMessage := false
 
 	for r.scanner.Scan() {
 		line := r.scanner.Text()
 		lines = append(lines, line)
 
-		// Empty line marks end of an SSE message
 		if line == "" {
 			gotMessage = true
 			break
@@ -52,7 +48,8 @@ func (r *sseFilterReader) Read(p []byte) (int, error) {
 
 	if err := r.scanner.Err(); err != nil {
 		r.done = true
-		return 0, err
+		r.buf.WriteString("event: error\ndata: {\"error\":\"stream line too long\"}\n\n")
+		return r.buf.Read(p)
 	}
 
 	if !gotMessage && len(lines) == 0 {

internal/proxy/sse_test.go

@@ -240,3 +240,28 @@ func TestSSEFilterReader_EmptyStream(t *testing.T) {
 		t.Errorf("expected empty output for empty stream, got %q", string(output))
 	}
 }
+
+func TestSSEFilterReader_OversizedLine(t *testing.T) {
+	// Create a line that exceeds the 1MB scanner buffer limit.
+	// The scanner has a 1MB max token size; build a single line larger than that.
+	oversized := "data: " + strings.Repeat("x", 1024*1024+1) + "\n\n"
+	f := newTestFilter(t, filter.ActionRedact)
+
+	reader := &sseFilterReader{
+		reader: io.NopCloser(strings.NewReader(oversized)),
+		filter: f,
+	}
+
+	output, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	result := string(output)
+	if !strings.Contains(result, "event: error") {
+		t.Error("expected SSE error event for oversized line")
+	}
+	if !strings.Contains(result, "stream line too long") {
+		t.Error("expected 'stream line too long' message in error event")
+	}
+}

internal/proxy/upstream.go

@@ -8,34 +8,34 @@ import (
 
 // MatchUpstream checks if a target URL is allowed by any of the given patterns.
 // Patterns are glob-style with scheme (e.g., "https://*.googleapis.com").
+//
+// Hostname glob rules:
+//   - "*" matches exactly one hostname segment (between dots)
+//   - "**" matches one or more hostname segments (across dots)
+//
 // Returns false for URLs with userinfo, non-HTTP schemes, or malformed input.
 func MatchUpstream(rawURL string, patterns []string) bool {
 	u, err := url.Parse(rawURL)
 	if err != nil {
 		return false
 	}
 
-	// Reject non-HTTP schemes
 	if u.Scheme != "http" && u.Scheme != "https" {
 		return false
 	}
 
-	// Reject userinfo (SSRF vector: http://user@internal-host)
 	if u.User != nil {
 		return false
 	}
 
-	// Reject URLs with query parameters or fragments (keep upstream clean)
 	if u.RawQuery != "" || u.Fragment != "" {
 		return false
 	}
 
-	// Must have a host
 	if u.Host == "" {
 		return false
 	}
 
-	// Match against each pattern
 	for _, pattern := range patterns {
 		if matchUpstreamPattern(u, pattern) {
 			return true
@@ -46,32 +46,31 @@ func MatchUpstream(rawURL string, patterns []string) bool {
 }
 
 // matchUpstreamPattern matches a parsed URL against a single pattern.
-// Pattern format: "https://*.example.com" or "https://api.example.com"
+// Pattern format: "https://*.example.com" or "https://**.example.com"
 func matchUpstreamPattern(u *url.URL, pattern string) bool {
-	// Replace * with WILDCARD so url.Parse doesn't treat globs as invalid path chars.
-	p, err := url.Parse(strings.ReplaceAll(pattern, "*", "WILDCARD"))
+	// Replace ** first, then * with distinct placeholders for url.Parse.
+	sanitized := strings.ReplaceAll(pattern, "**", "DOUBLEWILD")
+	sanitized = strings.ReplaceAll(sanitized, "*", "SINGLEWILD")
+
+	p, err := url.Parse(sanitized)
 	if err != nil {
 		return false
 	}
 
-	// Scheme must match exactly
-	patternScheme := strings.ReplaceAll(p.Scheme, "WILDCARD", "*")
+	patternScheme := strings.ReplaceAll(p.Scheme, "DOUBLEWILD", "**")
+	patternScheme = strings.ReplaceAll(patternScheme, "SINGLEWILD", "*")
 	if patternScheme != u.Scheme {
 		return false
 	}
 
-	// Match hostname using path.Match (glob matching where * matches any non-/ chars).
-	// Since hostnames contain no /, * matches across dots (e.g., *.example.com).
-	// Normalize to lowercase because DNS hostnames are case-insensitive.
-	patternHost := strings.ToLower(strings.ReplaceAll(p.Host, "WILDCARD", "*"))
-	matched, err := path.Match(patternHost, strings.ToLower(u.Host))
-	if err != nil || !matched {
+	patternHost := strings.ReplaceAll(p.Host, "DOUBLEWILD", "**")
+	patternHost = strings.ReplaceAll(patternHost, "SINGLEWILD", "*")
+	if !matchHostname(strings.ToLower(patternHost), strings.ToLower(u.Host)) {
 		return false
 	}
 
-	// If pattern has a path beyond "/", the URL path must match exactly or
-	// as a segment prefix (e.g., /v1 matches /v1/foo but not /v1-admin).
-	patternPath := strings.ReplaceAll(p.Path, "WILDCARD", "*")
+	patternPath := strings.ReplaceAll(p.Path, "DOUBLEWILD", "**")
+	patternPath = strings.ReplaceAll(patternPath, "SINGLEWILD", "*")
 	if patternPath != "" && patternPath != "/" {
 		if u.Path != patternPath && !strings.HasPrefix(u.Path, patternPath+"/") {
 			return false
@@ -80,3 +79,56 @@ func matchUpstreamPattern(u *url.URL, pattern string) bool {
 
 	return true
 }
+
+// matchHostname matches a hostname against a pattern where:
+//   - "*" matches exactly one dot-separated segment
+//   - "**" matches one or more dot-separated segments
+func matchHostname(pattern, host string) bool {
+	// Fast path: no wildcards, use path.Match for simple glob chars (e.g., ?)
+	if !strings.Contains(pattern, "*") {
+		matched, err := path.Match(pattern, host)
+		return err == nil && matched
+	}
+
+	patternParts := strings.Split(pattern, ".")
+	hostParts := strings.Split(host, ".")
+
+	return matchSegments(patternParts, hostParts)
+}
+
+// matchSegments recursively matches pattern segments against host segments.
+func matchSegments(pattern, host []string) bool {
+	for len(pattern) > 0 {
+		seg := pattern[0]
+
+		if seg == "**" {
+			// ** must match at least one segment
+			if len(host) == 0 {
+				return false
+			}
+			rest := pattern[1:]
+			// Try consuming 1..N host segments
+			for i := 1; i <= len(host); i++ {
+				if matchSegments(rest, host[i:]) {
+					return true
+				}
+			}
+			return false
+		}
+
+		if len(host) == 0 {
+			return false
+		}
+
+		// Single segment match: * matches one segment, otherwise literal/glob
+		matched, err := path.Match(seg, host[0])
+		if err != nil || !matched {
+			return false
+		}
+
+		pattern = pattern[1:]
+		host = host[1:]
+	}
+
+	return len(host) == 0
+}

internal/proxy/upstream_test.go

@@ -23,8 +23,32 @@ func TestMatchUpstream_GlobSubdomain(t *testing.T) {
 		{"https://storage.googleapis.com", true},
 		{"https://compute.googleapis.com", true},
 		{"https://googleapis.com", false},
-		{"https://evil.com.googleapis.com", true}, // single * matches one segment
-		{"http://storage.googleapis.com", false},  // scheme mismatch
+		{"https://evil.com.googleapis.com", false}, // * matches one segment only
+		{"http://storage.googleapis.com", false},   // scheme mismatch
+	}
+
+	for _, tt := range tests {
+		got := MatchUpstream(tt.url, patterns)
+		if got != tt.want {
+			t.Errorf("MatchUpstream(%q) = %v, want %v", tt.url, got, tt.want)
+		}
+	}
+}
+
+func TestMatchUpstream_DoubleStarSubdomain(t *testing.T) {
+	patterns := []string{"https://**.googleapis.com"}
+
+	tests := []struct {
+		url  string
+		want bool
+	}{
+		{"https://storage.googleapis.com", true},
+		{"https://compute.googleapis.com", true},
+		{"https://googleapis.com", false},                     // ** requires at least one segment
+		{"https://intended.com.googleapis.com", true},         // ** matches multiple segments
+		{"https://deep.nested.sub.googleapis.com", true},      // ** matches many segments
+		{"http://storage.googleapis.com", false},              // scheme mismatch
+		{"https://storage.googleapis.com/some/path", true},    // path allowed
 	}
 
 	for _, tt := range tests {