package service import ( "bytes" "context" "crypto/hmac" "crypto/sha256" "crypto/subtle" "encoding/hex" "encoding/json" "errors" "io" "net/http" "net/url" "strconv" "strings" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" db "github.com/multica-ai/multica/server/pkg/db/generated" "github.com/multica-ai/multica/server/pkg/plugincontract" "github.com/multica-ai/multica/server/pkg/remotemcp" ) // The hook engine: the one place Multica calls OUT to a plugin's own server. // // Everything before this ran the other way — a sandboxed surface asked the host // and the host acted on the signed-in user's session, so no request ever left // our infrastructure on a plugin's say-so. A hook does leave, which is why the // checks here are about the destination rather than the caller: the host must // be one the manifest declared through a `net:` scope, it must resolve to a // public address at dial time, and the body must be signed so the receiver can // tell our call from anyone else's. const ( // Signature scheme version, carried in the header so the scheme can change // without every handler guessing which one it is looking at. hookSignatureVersion = "v1" // How far a request timestamp may be from our clock. Bounds replay: a // captured request is only reusable inside this window, and a handler that // also remembers signatures can close it entirely. hookTimestampTolerance = 5 * time.Minute // Fallback when a manifest omits timeout_ms. hookDefaultTimeout = 10 * time.Second // Response bodies are read for the caller, so they are capped. A hook that // answers with a gigabyte should fail, not consume the host. hookMaxResponseBytes = 1 << 20 // Event retry schedule. Three attempts total, then the call is abandoned. hookEventAttempts = 3 hookEventBackoff = 2 * time.Second // Circuit breaker. After this many failures inside the window, background // delivery for the hook stops until the window rolls off — an endpoint that // has been down for a minute does not need one request per event or schedule // occurrence to notice. hookBreakerThreshold = 5 hookBreakerWindow = 5 * time.Minute // Per-hook rate limit, counted in attempts. hookRateLimit = 120 hookRateWindow = time.Minute ) // HookResult is what a completed hook call returns to whoever invoked it. type HookResult struct { Status string `json:"status"` Output json.RawMessage `json:"output,omitempty"` Error string `json:"error,omitempty"` Latency int `json:"latency_ms"` HookKey string `json:"hook_key"` Trigger string `json:"trigger"` Attempts int `json:"attempts"` } // HookActor is who the resulting writes belong to. // // This is the whole identity model for hooks, and it follows the trigger rather // than the plugin: a ui/manual hook runs because a person pressed something, so // its writes stay that person's (marked via_plugin_id); event/schedule hooks // have no person behind them and must not borrow the last one who touched the // resource. type HookActor struct { Type string ID pgtype.UUID } // HookInvocation is one call the engine has been asked to make. type HookInvocation struct { // ID is allocated before the outbound request. The receiver can log it and // correlate a response with the durable invocation row even when the call // itself fails. ID pgtype.UUID Installation db.PluginInstallation Hook plugincontract.Hook Trigger string // EventType is set only for the event trigger, and names what happened. EventType string Actor HookActor // IssueID is the issue this call is about, when there is one. It narrows the // callback token so a handler answering about one issue cannot use the same // grant to reach across the workspace. IssueID pgtype.UUID Input any // DeliveryID is stable across retries of the same scheduled occurrence. // PlannedAt is the canonical UTC cron occurrence, not delivery wall time. DeliveryID string PlannedAt pgtype.Timestamptz Attempt int } // hookRequestBody is the wire format. Stable and small on purpose: a hook // handler written against v1 should not have to care what else the host learns // how to send later. type hookRequestBody struct { Version int `json:"version"` InvocationID string `json:"invocation_id"` DeliveryID string `json:"delivery_id,omitempty"` Attempt int `json:"attempt"` OccurredAt time.Time `json:"occurred_at"` HookKey string `json:"hook_key"` Trigger string `json:"trigger"` EventType string `json:"event_type,omitempty"` WorkspaceID string `json:"workspace_id"` Installation string `json:"installation_id"` // IssueID is the issue this call is about, as resolved and permission-checked // by the host. Sent because the alternative is every handler reading it out // of client-supplied `input` — unvalidated, and absent entirely for the event // trigger, where no client was involved at all. IssueID string `json:"issue_id,omitempty"` Actor hookRequestActor `json:"actor"` Input json.RawMessage `json:"input,omitempty"` // Config is the installation's non-secret configuration, the values an // administrator typed into the host-rendered form. Sent because the handler // has no other way to read them — the Action API deliberately has no config // endpoint — and a plugin forced to keep its own second copy would make the // manifest's config block decorative. // // Secret-typed fields are NEVER included. They are the plugin's credentials // for ITS OWN services; it already has them, and putting them on the wire // would hand every value to whoever holds the endpoint. Config map[string]any `json:"config,omitempty"` // CallbackToken lets the handler call the Action API back for the few // minutes it is valid. Narrower than the installation's own token and tied // to this one call, so a handler that leaks it leaks a few minutes of the // scopes it was already using, not standing access. CallbackToken string `json:"callback_token,omitempty"` CallbackURL string `json:"callback_url,omitempty"` Schedule *hookRequestSchedule `json:"schedule,omitempty"` } type hookRequestActor struct { Type string `json:"type"` ID string `json:"id,omitempty"` } type hookRequestSchedule struct { PlannedAt time.Time `json:"planned_at"` } // FindHook returns the named hook from an installation's consented manifest. // // Read from the installation row, never from the manifest the source URL serves // right now: the admin consented to a specific set of endpoints, and a plugin // that later repoints its own hook at somewhere else must go through the // upgrade flow to have that take effect. func FindHook(installation db.PluginInstallation, hookKey string) (plugincontract.Hook, error) { manifest, err := ParseInstallationManifest(installation) if err != nil { return plugincontract.Hook{}, pluginErrf(PluginErrorInvalid, "plugin manifest is unreadable") } for _, hook := range manifest.Contributes.Hooks { if hook.Key == hookKey { return hook, nil } } return plugincontract.Hook{}, pluginErrf(PluginErrorNotFound, "this Plugin has no hook named %q", hookKey) } // HookAllowsTrigger reports whether the hook declared this trigger. A trigger // the manifest did not list is not a call site the host may invent. func HookAllowsTrigger(hook plugincontract.Hook, trigger string) bool { for _, declared := range hook.Triggers { if declared == trigger { return true } } return false } // InvokeHook performs one call and records it. // // Callers pick the blocking behaviour, not this function: ui/manual await the // result because a person is watching, event delivery runs outside the host // request, and schedule delivery runs under the durable scheduler lease. func (s *PluginService) InvokeHook(ctx context.Context, invocation HookInvocation, attempt int) (HookResult, error) { if !invocation.ID.Valid { invocation.ID = pgtype.UUID{Bytes: uuid.New(), Valid: true} } if attempt < 1 { attempt = 1 } invocation.Attempt = attempt hook := invocation.Hook result := HookResult{HookKey: hook.Key, Trigger: invocation.Trigger, Attempts: attempt} if !invocation.Installation.Enabled { return result, pluginErrf(PluginErrorForbidden, "this Plugin is disabled") } if !HookAllowsTrigger(hook, invocation.Trigger) { return result, pluginErrf(PluginErrorForbidden, "hook %q does not declare the %s trigger", hook.Key, invocation.Trigger) } if hook.Transport.Type != plugincontract.TransportHTTP { return result, pluginErrf(PluginErrorIncompatible, "hook transport %q is not supported yet", hook.Transport.Type) } if err := s.checkHookRate(ctx, invocation.Installation.ID, hook.Key); err != nil { return result, err } started := time.Now() output, callErr := s.callHookEndpoint(ctx, invocation) latency := int(time.Since(started).Milliseconds()) result.Latency = latency status := "ok" message := "" if callErr != nil { status = hookFailureStatus(callErr) message = redactHookError(callErr) result.Error = message } else { result.Output = output } result.Status = status s.recordInvocation(ctx, invocation, status, attempt, latency, message) if callErr != nil { return result, callErr } return result, nil } // callHookEndpoint is the network half: validate the destination, sign, send. func (s *PluginService) callHookEndpoint(ctx context.Context, invocation HookInvocation) (json.RawMessage, error) { installation := invocation.Installation granted := decodeScopes(installation.GrantedScopes) // The destination must be inside the consented `net:` set. Passing the // granted domains as the allowlist means the same string the admin approved // on the consent screen is what bounds the request — there is no second // list to fall out of sync. domains := plugincontract.NetDomains(granted) if len(domains) == 0 { return nil, pluginErrf(PluginErrorForbidden, "this Plugin was granted no net: scope, so it cannot call out") } // Two ways to reach an endpoint, and only the network guard differs between // them. The consent check does not: a destination outside the granted `net:` // set is refused on both paths, because that is what the admin approved and // no deployment setting may widen it. var endpoint *url.URL client := &http.Client{} if s.isDevOrigin(invocation.Hook.Transport.URL) { // The operator named this exact origin in MULTICA_PLUGIN_DEV_ORIGINS — // the same opt-in that lets a manifest be served from a local dev // server, for the same reason: an author building a hook has nowhere // public to point it yet. parsed, err := url.Parse(invocation.Hook.Transport.URL) if err != nil { return nil, &PluginError{Kind: PluginErrorInvalid, Message: "hook endpoint is not a valid URL", Err: err} } if !hostInNetScopes(parsed.Hostname(), domains) { return nil, pluginErrf(PluginErrorForbidden, "hook endpoint host is not covered by a net: scope") } endpoint = parsed if s.HookClient != nil { client = s.HookClient } } else { validated, err := remotemcp.ValidatePublicHTTPSEndpoint(ctx, invocation.Hook.Transport.URL, domains, nil) if err != nil { return nil, &PluginError{Kind: PluginErrorForbidden, Message: "hook endpoint is not allowed", Err: err} } endpoint = validated // NewSecureHTTPClient re-resolves at dial and refuses a non-public // address, so a hostname that passed validation and then flipped to a // private IP cannot be used to reach inside the network. client = remotemcp.NewSecureHTTPClient(endpoint) } body, err := s.buildHookBody(ctx, invocation) if err != nil { return nil, err } // The grant lives exactly as long as the call it was issued for. Without // this it would stay usable for the rest of its TTL after the handler has // already answered. if s.Callbacks != nil && body.CallbackToken != "" { defer s.Callbacks.Revoke(body.CallbackToken) } encoded, err := json.Marshal(body) if err != nil { return nil, &PluginError{Kind: PluginErrorInvalid, Message: "encode hook request", Err: err} } timeout := hookDefaultTimeout if invocation.Hook.TimeoutMs > 0 { timeout = time.Duration(invocation.Hook.TimeoutMs) * time.Millisecond } callCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() request, err := http.NewRequestWithContext(callCtx, http.MethodPost, endpoint.String(), bytes.NewReader(encoded)) if err != nil { return nil, &PluginError{Kind: PluginErrorInvalid, Message: "build hook request", Err: err} } timestamp := strconv.FormatInt(time.Now().Unix(), 10) signature, err := s.SignHookPayload(installation.ID, timestamp, encoded) if err != nil { return nil, err } request.Header.Set("Content-Type", "application/json") request.Header.Set("X-Multica-Timestamp", timestamp) request.Header.Set("X-Multica-Signature", hookSignatureVersion+"="+signature) request.Header.Set("X-Multica-Plugin-Installation", uuidString(installation.ID)) request.Header.Set("User-Agent", "Multica-Hooks/1") response, err := client.Do(request) if err != nil { return nil, &PluginError{Kind: PluginErrorUnavailable, Message: "hook endpoint did not answer", Err: err} } defer func() { _ = response.Body.Close() }() payload, err := io.ReadAll(io.LimitReader(response.Body, hookMaxResponseBytes)) if err != nil { return nil, &PluginError{Kind: PluginErrorUnavailable, Message: "read hook response", Err: err} } if response.StatusCode < 200 || response.StatusCode >= 300 { return nil, pluginErrf(PluginErrorUnavailable, "hook endpoint returned %d", response.StatusCode) } if len(bytes.TrimSpace(payload)) == 0 { return nil, nil } if !json.Valid(payload) { return nil, pluginErrf(PluginErrorInvalid, "hook endpoint returned a non-JSON body") } return json.RawMessage(payload), nil } // nonSecretConfig reads the installation's stored configuration and drops // anything the manifest declared as a secret. // // Filtered against the MANIFEST rather than against the stored shape: secrets // live in their own table and should never appear in the config column at all, // so this is the check that would catch it if one ever did. func nonSecretConfig(installation db.PluginInstallation) map[string]any { if len(installation.Config) == 0 { return nil } values := map[string]any{} if err := json.Unmarshal(installation.Config, &values); err != nil { return nil } manifest, err := ParseInstallationManifest(installation) if err != nil { // Without a manifest there is no way to tell which keys are secret, so // send none of them. return nil } for key := range values { field, declared := manifest.Config.Field(key) if !declared || field.Type == plugincontract.ConfigSecret { delete(values, key) } } if len(values) == 0 { return nil } return values } func (s *PluginService) buildHookBody(ctx context.Context, invocation HookInvocation) (hookRequestBody, error) { attempt := invocation.Attempt if attempt < 1 { attempt = 1 } body := hookRequestBody{ Version: 1, InvocationID: uuidString(invocation.ID), DeliveryID: invocation.DeliveryID, Attempt: attempt, OccurredAt: time.Now().UTC(), HookKey: invocation.Hook.Key, Trigger: invocation.Trigger, EventType: invocation.EventType, WorkspaceID: uuidString(invocation.Installation.WorkspaceID), Installation: uuidString(invocation.Installation.ID), Actor: hookRequestActor{Type: invocation.Actor.Type, ID: uuidString(invocation.Actor.ID)}, } if invocation.PlannedAt.Valid { body.Schedule = &hookRequestSchedule{PlannedAt: invocation.PlannedAt.Time.UTC()} } if invocation.IssueID.Valid { body.IssueID = uuidString(invocation.IssueID) } body.Config = nonSecretConfig(invocation.Installation) if invocation.Input != nil { encoded, err := json.Marshal(invocation.Input) if err != nil { return hookRequestBody{}, &PluginError{Kind: PluginErrorInvalid, Message: "encode hook input", Err: err} } body.Input = encoded } if s.Callbacks != nil { token, err := s.Callbacks.Issue(ctx, invocation) if err != nil { return hookRequestBody{}, err } body.CallbackToken = token body.CallbackURL = s.CallbackBaseURL } return body, nil } // SignHookPayload produces the hex HMAC a receiver checks. // // The signed string joins the timestamp and the body with a separator that // cannot appear in the timestamp. Signing the body alone would let a captured // request be replayed forever; signing a plain concatenation would let a // crafted timestamp and body pair swap bytes between the two fields. func (s *PluginService) SignHookPayload(installationID pgtype.UUID, timestamp string, body []byte) (string, error) { key, err := s.hookSigningKey(installationID) if err != nil { return "", err } mac := hmac.New(sha256.New, key) mac.Write([]byte(timestamp)) mac.Write([]byte(".")) mac.Write(body) return hex.EncodeToString(mac.Sum(nil)), nil } // hookSigningKey derives this installation's signing secret from the deployment // key. // // Derived rather than stored, and this is the asymmetry that decides it: the // host must PRODUCE this value to sign with it, so a one-way hash is not an // option, and storing a recoverable copy of it in every installation row would // put a usable secret in reach of any database read. The install token goes the // other way — the plugin produces it and the host only ever verifies — so that // one is stored hashed. Same deployment key, opposite directions. func (s *PluginService) hookSigningKey(installationID pgtype.UUID) ([]byte, error) { if len(s.DeploymentKey) == 0 { return nil, pluginErrf(PluginErrorUnavailable, "hooks are disabled: MULTICA_PLUGIN_SECRET_KEY is not configured") } if len(s.DeploymentKey) != 32 { return nil, pluginErrf(PluginErrorUnavailable, "hooks are disabled: MULTICA_PLUGIN_SECRET_KEY must decode to 32 bytes") } mac := hmac.New(sha256.New, s.DeploymentKey) mac.Write([]byte("multica-plugin-hook-signature:v1:")) mac.Write([]byte(uuidString(installationID))) return mac.Sum(nil), nil } // HookSigningSecret is the same value in the form an author configures on their // own server. Shown once at install time next to the install token. func (s *PluginService) HookSigningSecret(installationID pgtype.UUID) (string, error) { key, err := s.hookSigningKey(installationID) if err != nil { return "", err } return "whsec_" + hex.EncodeToString(key), nil } // VerifyHookSignature is the receiver's half, exported so a plugin written in Go // and our own tests check signatures the same way — a scheme only one side can // implement is a scheme nobody can review. func VerifyHookSignature(secretHex, timestamp string, body []byte, presented string, now time.Time) error { secretHex = strings.TrimPrefix(secretHex, "whsec_") key, err := hex.DecodeString(secretHex) if err != nil { return errors.New("signing secret is not valid hex") } seconds, err := strconv.ParseInt(timestamp, 10, 64) if err != nil { return errors.New("timestamp is not an integer") } drift := now.Sub(time.Unix(seconds, 0)) if drift < 0 { drift = -drift } if drift > hookTimestampTolerance { return errors.New("timestamp is outside the accepted window") } mac := hmac.New(sha256.New, key) mac.Write([]byte(timestamp)) mac.Write([]byte(".")) mac.Write(body) expected := hex.EncodeToString(mac.Sum(nil)) presented = strings.TrimPrefix(presented, hookSignatureVersion+"=") // Constant time: a byte-at-a-time comparison leaks how much of a guessed // signature was right, which is enough to find the rest. if subtle.ConstantTimeCompare([]byte(expected), []byte(presented)) != 1 { return errors.New("signature does not match") } return nil } // checkHookRate caps how often one hook may be called. func (s *PluginService) checkHookRate(ctx context.Context, installationID pgtype.UUID, hookKey string) error { since := pgtype.Timestamptz{Time: time.Now().Add(-hookRateWindow), Valid: true} count, err := s.Queries.CountRecentPluginInvocations(ctx, db.CountRecentPluginInvocationsParams{ InstallationID: installationID, HookKey: hookKey, CreatedAt: since, }) if err != nil { // A telemetry read that fails must not take the feature down with it. return nil } if count >= hookRateLimit { return pluginErrf(PluginErrorQuota, "hook %q exceeded %d calls per minute", hookKey, hookRateLimit) } return nil } // HookBreakerOpen reports whether background delivery for this hook is // currently suspended after repeated failures. func (s *PluginService) HookBreakerOpen(ctx context.Context, installationID pgtype.UUID, hookKey string) bool { since := pgtype.Timestamptz{Time: time.Now().Add(-hookBreakerWindow), Valid: true} failures, err := s.Queries.CountRecentPluginFailures(ctx, db.CountRecentPluginFailuresParams{ InstallationID: installationID, HookKey: hookKey, CreatedAt: since, }) if err != nil { return false } return failures >= hookBreakerThreshold } func (s *PluginService) recordInvocation(ctx context.Context, invocation HookInvocation, status string, attempt, latency int, message string) { params := db.CreatePluginInvocationParams{ ID: invocation.ID, InstallationID: invocation.Installation.ID, WorkspaceID: invocation.Installation.WorkspaceID, HookKey: invocation.Hook.Key, Trigger: invocation.Trigger, Status: status, Attempt: int32(attempt), LatencyMs: int32(latency), } if invocation.DeliveryID != "" { params.DeliveryID = pgtype.Text{String: invocation.DeliveryID, Valid: true} } if invocation.PlannedAt.Valid { params.PlannedAt = invocation.PlannedAt } if invocation.EventType != "" { params.EventType = pgtype.Text{String: invocation.EventType, Valid: true} } if message != "" { params.Error = pgtype.Text{String: truncate(message, 500), Valid: true} } // Best effort: telemetry must never fail the call it is describing. _, _ = s.Queries.CreatePluginInvocation(ctx, params) } func hookFailureStatus(err error) string { var pluginErr *PluginError if errors.As(err, &pluginErr) { switch pluginErr.Kind { case PluginErrorForbidden, PluginErrorQuota, PluginErrorIncompatible: return "refused" } } if errors.Is(err, context.DeadlineExceeded) { return "timeout" } return "failed" } // redactHookError keeps the host's own description and drops anything the // remote end supplied. An endpoint that echoes its input could otherwise write // issue content into a table that has no deletion path for it. func redactHookError(err error) string { var pluginErr *PluginError if errors.As(err, &pluginErr) { return truncate(pluginErr.Message, 500) } if errors.Is(err, context.DeadlineExceeded) { return "hook endpoint timed out" } return "hook call failed" } // hostInNetScopes is the exact-host check the consent screen promises. Never a // suffix match: one scope string must mean the same thing here as it does in // the surface CSP and on the authorization screen. func hostInNetScopes(host string, domains []string) bool { host = strings.ToLower(strings.TrimSuffix(host, ".")) for _, domain := range domains { if host == strings.ToLower(domain) { return true } } return false } func truncate(value string, limit int) string { if len(value) <= limit { return value } return value[:limit] }