feat(pluginsdk): support external HTTP credential plugins (#656)

Add a generic SDK + gRPC runtime so credential plugins can expose HTTP
injection, dashboard secret slots, agent env pushdown, and OAuth flow
metadata out of tree, keeping provider-specific logic out of core.

* pluginsdk.CredentialDef can declare HTTP injection and return
  per-instance metadata (secret slots, env vars, OAuth,
  disambiguators) from Build, and may implement InjectHTTP to stamp
  request headers at proxy time.
* The external-plugin adapter maps that metadata onto the existing
  secret-slot, env-pushdown, OAuth-flow, and HTTP-credential
  interfaces; the built-in HTTPS endpoint routes placeholder-matched
  requests through the plugin's InjectHTTP, applies the returned
  header mutations, and consumes returned redactions for audit
  samples.
* Dynamic MCP OAuth is selected by the credential flow (dynamic_mcp)
  rather than provider-hostname checks; the auth-code exchange uses
  public-client PKCE and tolerates pasted callback URLs and reordered
  query parameters.
* Plugin callbacks (Build, InjectHTTP, HandleConn, tunnel
  Open/Close/Dial) recover from panics, header mutations are
  validated, plugin-returned metadata shape is checked, and derived
  secrets are redacted from audit samples.

Author: dhruvkelawala

cmd/clawpatrol/extplugin_http_credential_test.go

@@ -114,8 +114,13 @@ func TestFetchEnvPushdownErrors(t *testing.T) {
 // flat list.
 func TestEnvPushdownVarsServerDriven(t *testing.T) {
 	prev := envPushdownGatewayFetcher
+	prevDaemon := envPushdownDaemonFetcher
 	envPushdownGatewayFetcher = fetchEnvPushdownFromGateway
-	t.Cleanup(func() { envPushdownGatewayFetcher = prev })
+	envPushdownDaemonFetcher = nil
+	t.Cleanup(func() {
+		envPushdownGatewayFetcher = prev
+		envPushdownDaemonFetcher = prevDaemon
+	})
 
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		_ = json.NewEncoder(w).Encode(map[string]any{
@@ -162,8 +167,13 @@ func TestEnvPushdownVarsErrorReturnsCAOnly(t *testing.T) {
 	// the host's running NE happens to have to say, instead of the
 	// gateway-URL-missing error this test exercises.
 	prev := envPushdownGatewayFetcher
+	prevDaemon := envPushdownDaemonFetcher
 	envPushdownGatewayFetcher = fetchEnvPushdownFromGateway
-	t.Cleanup(func() { envPushdownGatewayFetcher = prev })
+	envPushdownDaemonFetcher = nil
+	t.Cleanup(func() {
+		envPushdownGatewayFetcher = prev
+		envPushdownDaemonFetcher = prevDaemon
+	})
 
 	dir := t.TempDir()
 	caPath := filepath.Join(dir, "ca.crt")

cmd/clawpatrol/integrations_test.go

@@ -2417,8 +2417,16 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c
 						}
 					case wantsHTTP:
 						reqBodySecretRedactions = appendCredentialSecretRedactions(reqBodySecretRedactions, sec)
+						// Match existing request-signing behavior: an injection failure is logged,
+						// then the request continues with the agent's placeholder. The upstream
+						// service should reject that placeholder without exposing gateway secrets.
 						if err := injector.InjectHTTP(req.Context(), req, sec); err != nil {
-							log.Printf("inject %s: %v", cc.Credential.Symbol.Name, err)
+							log.Printf("inject %s: %v; forwarding without injection", cc.Credential.Symbol.Name, err)
+						}
+						if rp, ok := injector.(runtime.HTTPCredentialRedactionProvider); ok {
+							for _, secret := range rp.ConsumeHTTPRedactions(req) {
+								reqBodySecretRedactions = appendCredentialSecretRedaction(reqBodySecretRedactions, secret)
+							}
 						}
 					}
 					if wantsWS && isWSUpgrade(req) {
@@ -2598,7 +2606,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c
 			}
 		}
 		ev.Status = resp.StatusCode
-		ev.ReqHeaders = flatHeaders(req.Header)
+		ev.ReqHeaders = flatHeadersRedacted(req.Header, reqBodySecretRedactions)
 		ev.In = reqS.n
 		ev.Out = respS.n
 		ev.ReqSha = reqS.sha()

cmd/clawpatrol/main.go

@@ -56,11 +56,12 @@ type oauthState struct {
 	displayName string // human-readable name (e.g. github login)
 	avatarURL   string // dashboard pfp
 	// clientID is the dynamically-registered OAuth client_id for flows
-	// that use RFC 7591 (notion_mcp). Static-ClientID flows (github,
-	// anthropic, codex) leave this empty and use cfg.ClientID. Persisted
-	// in the credentials table alongside the tokens so refresh works
-	// across gateway restarts.
+	// that use RFC 7591 (notion_mcp/dynamic_mcp). Static-ClientID flows
+	// (github, anthropic, codex) leave this empty and use cfg.ClientID.
+	// Persisted in the credentials table alongside the tokens so refresh
+	// works across gateway restarts.
 	clientID string
+	flow     string
 	db       *sql.DB
 	mu       sync.Mutex
 }
@@ -218,8 +219,8 @@ func (r *OAuthRegistry) Set(ctx context.Context, id string, tok *oauth2.Token) e
 
 // SetWithClient is Set + a dynamically registered client_id. Pass empty
 // clientID for static-ClientID flows; pass the per-credential client_id
-// for RFC 7591 dynamic registration flows (notion_mcp). The clientID is
-// stamped onto the in-memory state and persisted alongside the tokens so
+// for RFC 7591 dynamic registration flows (notion_mcp/dynamic_mcp). The
+// clientID is stamped onto the in-memory state and persisted alongside the tokens so
 // refresh continues to work after restart.
 //
 // The userinfo fetch (fetchOAuthProfile) runs OUTSIDE the registry lock
@@ -335,6 +336,7 @@ func newState(it *OAuthIntegration, db *sql.DB) *oauthState {
 		header: header,
 		prefix: prefix,
 		id:     it.ID,
+		flow:   it.Flow,
 		db:     db,
 	}
 }
@@ -347,11 +349,13 @@ func (s *oauthState) setToken(tok *oauth2.Token) {
 		// (returns "Invalid request format" otherwise). Stdlib oauth2
 		// only sends form-urlencoded.
 		base = &anthropicRefreshSource{cfg: s.cfg, current: tok}
-	case isNotionMCPTokenURL(s.cfg.Endpoint.TokenURL):
-		// Notion's MCP token endpoint refreshes via form-urlencoded body
-		// and expects the dynamically registered client_id (no static
-		// ClientSecret — PKCE-only public client).
-		base = &notionMCPRefreshSource{cfg: s.cfg, current: tok}
+	case s.flow == "dynamic_mcp" || s.flow == "notion_mcp":
+		// Hosted MCP token endpoints refresh via form-urlencoded body and
+		// expect the dynamically registered client_id (no static
+		// ClientSecret — PKCE-only public client). The flow, not the
+		// provider hostname, selects this behavior so external credential
+		// plugins can supply their own MCP OAuth endpoints.
+		base = &dynamicMCPRefreshSource{cfg: s.cfg, current: tok}
 	default:
 		base = s.cfg.TokenSource(context.Background(), tok)
 	}
@@ -535,7 +539,7 @@ func (r *OAuthRegistry) loadFromDB() error {
 		}
 		s := newState(it, r.db)
 		// Restore the dynamically-registered client_id BEFORE setToken
-		// so the refresh source (notion_mcp) picks it up via s.cfg.
+		// so dynamic MCP refresh sources pick it up via s.cfg.
 		if clientID.Valid && clientID.String != "" {
 			s.clientID = clientID.String
 			s.cfg.ClientID = clientID.String
@@ -566,8 +570,9 @@ type oauthSession struct {
 	id       string
 	created  time.Time
 	// dynClientID is the RFC 7591 client_id this session registered at
-	// start time (notion_mcp). Empty for static-ClientID flows. Stamped
-	// onto the credential row at exchange time so refresh can replay it.
+	// start time (dynamic MCP flows). Empty for static-ClientID flows.
+	// Stamped onto the credential row at exchange time so refresh can
+	// replay it.
 	dynClientID string
 }
 
@@ -646,8 +651,8 @@ func (w *webMux) apiOAuthStart(rw http.ResponseWriter, r *http.Request) {
 		w.startOpenAIDeviceFlow(rw, r, id, flow)
 		return
 	}
-	if flow.Flow == "notion_mcp" {
-		w.startNotionMCPFlow(rw, r, id, flow)
+	if flow.Flow == "notion_mcp" || flow.Flow == "dynamic_mcp" {
+		w.startDynamicMCPFlow(rw, r, id, flow)
 		return
 	}
 
@@ -677,14 +682,18 @@ func (w *webMux) apiOAuthStart(rw http.ResponseWriter, r *http.Request) {
 	writeJSON(rw, map[string]string{"auth_url": authURL, "state": state})
 }
 
-// startNotionMCPFlow drives the mcp.notion.com auth-code flow with RFC
-// 7591 dynamic client registration. Notion accepts arbitrary redirect
-// URIs at registration, so we register the dashboard's own
-// /oauth/callback page — the page auto-exchanges via /api/oauth/exchange
-// when it loads, with copy-paste from the URL bar as a fallback.
-func (w *webMux) startNotionMCPFlow(rw http.ResponseWriter, r *http.Request, id string, flow *OAuthIntegration) {
-	redirectURI := w.dashboardRedirectURI(r, "/oauth/callback")
-	clientID, err := registerOAuthClient(r.Context(), flow.OAuth.RegisterURL, redirectURI)
+// startDynamicMCPFlow drives auth-code OAuth flows with RFC 7591
+// dynamic client registration. Plugins may pin a provider-accepted
+// redirect URI (Amplitude uses a localhost loopback URI); otherwise we
+// register the dashboard's own /oauth/callback page, which
+// auto-exchanges via /api/oauth/exchange when it loads, with copy-paste
+// from the URL bar as a fallback.
+func (w *webMux) startDynamicMCPFlow(rw http.ResponseWriter, r *http.Request, id string, flow *OAuthIntegration) {
+	redirectURI := strings.TrimSpace(flow.OAuth.RedirectURI)
+	if redirectURI == "" {
+		redirectURI = w.dashboardRedirectURI(r, "/oauth/callback")
+	}
+	clientID, err := registerOAuthClient(r.Context(), flow.OAuth.RegisterURL, redirectURI, flow.OAuth.Scopes)
 	if err != nil {
 		http.Error(rw, "dynamic client registration: "+err.Error(), http.StatusBadGateway)
 		return
@@ -697,7 +706,11 @@ func (w *webMux) startNotionMCPFlow(rw http.ResponseWriter, r *http.Request, id
 		ClientID:    clientID,
 		Scopes:      flow.OAuth.Scopes,
 		RedirectURL: redirectURI,
-		Endpoint:    oauth2.Endpoint{AuthURL: flow.OAuth.AuthURL, TokenURL: flow.OAuth.TokenURL},
+		Endpoint: oauth2.Endpoint{
+			AuthURL:   flow.OAuth.AuthURL,
+			TokenURL:  flow.OAuth.TokenURL,
+			AuthStyle: oauth2.AuthStyleInParams,
+		},
 	}
 	authURL := cfg.AuthCodeURL(state,
 		oauth2.SetAuthURLParam("code_challenge", challenge),
@@ -740,14 +753,18 @@ func (w *webMux) dashboardRedirectURI(r *http.Request, path string) string {
 // registerOAuthClient performs RFC 7591 dynamic client registration
 // against `registerURL`, asking for a public PKCE client bound to the
 // given redirect URI. Returns the issued client_id.
-func registerOAuthClient(ctx context.Context, registerURL, redirectURI string) (string, error) {
-	body, _ := json.Marshal(map[string]any{
+func registerOAuthClient(ctx context.Context, registerURL, redirectURI string, scopes []string) (string, error) {
+	bodyMap := map[string]any{
 		"client_name":                "clawpatrol",
 		"redirect_uris":              []string{redirectURI},
 		"grant_types":                []string{"authorization_code", "refresh_token"},
 		"response_types":             []string{"code"},
 		"token_endpoint_auth_method": "none",
-	})
+	}
+	if len(scopes) > 0 {
+		bodyMap["scope"] = strings.Join(scopes, " ")
+	}
+	body, _ := json.Marshal(bodyMap)
 	req, err := http.NewRequestWithContext(ctx, "POST", registerURL, bytes.NewReader(body))
 	if err != nil {
 		return "", err
@@ -775,6 +792,70 @@ func registerOAuthClient(ctx context.Context, registerURL, redirectURI string) (
 	return rr.ClientID, nil
 }
 
+func normalizeOAuthExchangeInput(input string) string {
+	code, _ := parseOAuthExchangeInput(input)
+	return code
+}
+
+func parseOAuthExchangeInput(input string) (code, oauthErr string) {
+	s := strings.TrimSpace(input)
+	if s == "" {
+		return "", ""
+	}
+
+	// Operators often paste the whole callback URL after providers
+	// redirect to a loopback URI (for example
+	// localhost:8900/callback?code=...&state=...) or just its raw query
+	// string, with parameters in any order. Try query-shaped inputs
+	// first, then fall back to a bare opaque code.
+	if code, oauthErr, ok := parseOAuthRawQuery(s); ok {
+		return code, oauthErr
+	}
+	if strings.Contains(s, "?") {
+		candidate := s
+		if !strings.Contains(candidate, "://") && !strings.HasPrefix(candidate, "/") {
+			// Browsers omit the scheme when the URL is copied from the
+			// address bar; url.Parse needs one to find the query.
+			candidate = "http://" + candidate
+		}
+		if u, err := url.Parse(candidate); err == nil {
+			if code, oauthErr := oauthCodeOrError(u.Query()); code != "" || oauthErr != "" {
+				return code, oauthErr
+			}
+		}
+	}
+
+	// Otherwise treat the input as a bare code. Preserve '&' and '=' because
+	// opaque provider codes may contain them; only strip fragment/query suffixes.
+	if i := strings.IndexAny(s, "#?"); i > 0 {
+		s = s[:i]
+	}
+	return strings.TrimSpace(s), ""
+}
+
+func parseOAuthRawQuery(input string) (code, oauthErr string, ok bool) {
+	raw := strings.TrimPrefix(input, "?")
+	vals, err := url.ParseQuery(raw)
+	if err != nil {
+		return "", "", false
+	}
+	code, oauthErr = oauthCodeOrError(vals)
+	return code, oauthErr, code != "" || oauthErr != ""
+}
+
+func oauthCodeOrError(vals url.Values) (code, oauthErr string) {
+	if code := strings.TrimSpace(vals.Get("code")); code != "" {
+		return code, ""
+	}
+	if errCode := strings.TrimSpace(vals.Get("error")); errCode != "" {
+		if desc := strings.TrimSpace(vals.Get("error_description")); desc != "" {
+			return "", fmt.Sprintf("%s: %s", errCode, desc)
+		}
+		return "", errCode
+	}
+	return "", ""
+}
+
 func (w *webMux) apiOAuthExchange(rw http.ResponseWriter, r *http.Request) {
 	if r.Method != "POST" {
 		http.Error(rw, "POST", http.StatusMethodNotAllowed)
@@ -788,14 +869,16 @@ func (w *webMux) apiOAuthExchange(rw http.ResponseWriter, r *http.Request) {
 		http.Error(rw, err.Error(), 400)
 		return
 	}
-	body.Code = strings.TrimSpace(body.Code)
+	var oauthErr string
+	body.Code, oauthErr = parseOAuthExchangeInput(body.Code)
+	if oauthErr != "" {
+		http.Error(rw, "oauth error: "+oauthErr, 400)
+		return
+	}
 	if body.Code == "" || body.State == "" {
 		http.Error(rw, "missing code/state", 400)
 		return
 	}
-	if i := strings.IndexAny(body.Code, "#&?"); i > 0 {
-		body.Code = body.Code[:i]
-	}
 
 	w.mu.Lock()
 	sess, ok := w.sessions[body.State]
@@ -1155,56 +1238,52 @@ func isAnthropicTokenURL(u string) bool {
 	return strings.Contains(u, "anthropic.com/")
 }
 
-func isNotionMCPTokenURL(u string) bool {
-	return strings.Contains(u, "mcp.notion.com/")
-}
-
-// notionMCPRefreshSource refreshes Notion MCP OAuth tokens via the
+// dynamicMCPRefreshSource refreshes hosted MCP OAuth tokens via the
 // form-urlencoded body the spec mandates. The client_id is read from
 // cfg.ClientID, which the registry restores from the persisted
 // credentials.client_id column on boot. Stateful: holds the current
 // token (with refresh_token) and rotates it on each refresh.
-type notionMCPRefreshSource struct {
+type dynamicMCPRefreshSource struct {
 	mu      sync.Mutex
 	cfg     *oauth2.Config
 	current *oauth2.Token
 }
 
-func (n *notionMCPRefreshSource) Token() (*oauth2.Token, error) {
-	n.mu.Lock()
-	defer n.mu.Unlock()
-	if n.current.Valid() {
-		return n.current, nil
+func (d *dynamicMCPRefreshSource) Token() (*oauth2.Token, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.current.Valid() {
+		return d.current, nil
 	}
-	if n.current.RefreshToken == "" {
-		return nil, fmt.Errorf("notion_mcp refresh: no refresh_token")
+	if d.current.RefreshToken == "" {
+		return nil, fmt.Errorf("dynamic_mcp refresh: no refresh_token")
 	}
-	if n.cfg.ClientID == "" {
-		return nil, fmt.Errorf("notion_mcp refresh: no client_id (dynamic registration was never persisted)")
+	if d.cfg.ClientID == "" {
+		return nil, fmt.Errorf("dynamic_mcp refresh: no client_id (dynamic registration was never persisted)")
 	}
 	form := url.Values{}
 	form.Set("grant_type", "refresh_token")
-	form.Set("refresh_token", n.current.RefreshToken)
-	form.Set("client_id", n.cfg.ClientID)
+	form.Set("refresh_token", d.current.RefreshToken)
+	form.Set("client_id", d.cfg.ClientID)
 	// See anthropicRefreshSource.Token: oauth2 has no ctx on Token(),
-	// so we bound the upstream round-trip here so n.mu stays available
-	// even if Notion's token endpoint hangs.
+	// so we bound the upstream round-trip here so d.mu stays available
+	// even if the MCP token endpoint hangs.
 	ctx, cancel := context.WithTimeout(context.Background(), oauthUpstreamTimeout)
 	defer cancel()
-	req, err := http.NewRequestWithContext(ctx, "POST", n.cfg.Endpoint.TokenURL, strings.NewReader(form.Encode()))
+	req, err := http.NewRequestWithContext(ctx, "POST", d.cfg.Endpoint.TokenURL, strings.NewReader(form.Encode()))
 	if err != nil {
-		return nil, fmt.Errorf("notion_mcp refresh: build request: %w", err)
+		return nil, fmt.Errorf("dynamic_mcp refresh: build request: %w", err)
 	}
 	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
 	req.Header.Set("Accept", "application/json")
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		return nil, fmt.Errorf("notion_mcp refresh: %w", err)
+		return nil, fmt.Errorf("dynamic_mcp refresh: %w", err)
 	}
 	defer func() { _ = resp.Body.Close() }()
 	respBytes, _ := io.ReadAll(io.LimitReader(resp.Body, oauthResponseLimit))
 	if resp.StatusCode != 200 {
-		return nil, fmt.Errorf("notion_mcp refresh %d: %s", resp.StatusCode, string(respBytes))
+		return nil, fmt.Errorf("dynamic_mcp refresh %d: %s", resp.StatusCode, string(respBytes))
 	}
 	var tr struct {
 		AccessToken  string `json:"access_token"`
@@ -1221,12 +1300,12 @@ func (n *notionMCPRefreshSource) Token() (*oauth2.Token, error) {
 		TokenType:    tr.TokenType,
 	}
 	if t.RefreshToken == "" {
-		t.RefreshToken = n.current.RefreshToken
+		t.RefreshToken = d.current.RefreshToken
 	}
 	if tr.ExpiresIn > 0 {
 		t.Expiry = time.Now().Add(time.Duration(tr.ExpiresIn) * time.Second)
 	}
-	n.current = t
+	d.current = t
 	return t, nil
 }
 
@@ -1319,7 +1398,7 @@ func (w *webMux) apiOAuthRevoke(rw http.ResponseWriter, r *http.Request) {
 
 // oauthCallbackHTML is served at GET /oauth/callback for
 // dynamic-registration flows that redirect back to the dashboard
-// (notion_mcp). The inline JS extracts ?code & ?state, POSTs them to
+// (notion_mcp/dynamic_mcp). The inline JS extracts ?code & ?state, POSTs them to
 // /api/oauth/exchange so the original ConnectModal sees the credential
 // connect itself, and shows the code prominently as a copy-paste
 // fallback if the auto-exchange fails (e.g. dashboard secret expired,