diff --git a/go/cmd/compass-server/main.go b/go/cmd/compass-server/main.go index 680cccbc..a98798b1 100644 --- a/go/cmd/compass-server/main.go +++ b/go/cmd/compass-server/main.go @@ -382,6 +382,7 @@ type forgeFlags struct { installationID *string appKeySecret *string appWebhook *string + linearWebhook *string } // registerForgeFlags declares the forge flags on the given FlagSet and returns @@ -414,6 +415,12 @@ func registerForgeFlags(fs *flag.FlagSet) forgeFlags { "Declared server_only secret NAME holding the webhook signing secret the "+ "ingress verifies deliveries against. Defaults to "+ "$COMPASS_FORGE_APP_WEBHOOK_SECRET."), + linearWebhook: fs.String("forge-linear-webhook-secret", "", + "Declared server_only secret NAME holding the Linear webhook signing "+ + "secret the shared POST /webhooks/linear ingress verifies deliveries "+ + "against. Defaults to $COMPASS_FORGE_LINEAR_WEBHOOK_SECRET. The Linear "+ + "data-change/session arm runs iff this is declared, independent of the "+ + "GitHub App gate."), } } @@ -428,6 +435,7 @@ func (f forgeFlags) resolve() (server.ForgeConfig, error) { firstNonEmpty(*f.installationID, os.Getenv("COMPASS_FORGE_INSTALLATION_ID")), firstNonEmpty(*f.appKeySecret, os.Getenv("COMPASS_FORGE_APP_KEY_SECRET")), firstNonEmpty(*f.appWebhook, os.Getenv("COMPASS_FORGE_APP_WEBHOOK_SECRET")), + firstNonEmpty(*f.linearWebhook, os.Getenv("COMPASS_FORGE_LINEAR_WEBHOOK_SECRET")), ) } @@ -438,7 +446,7 @@ func (f forgeFlags) resolve() (server.ForgeConfig, error) { // so Owner/Name and owner/name collapse to one target. appID/installationID are // parsed as int64 when set (garbage is a startup error); empty leaves them zero. // Empty host/secret/App-secret NAMEs default server-side. -func resolveForge(repos, secret, host, appID, installationID, appKeySecret, appWebhook string) (server.ForgeConfig, error) { +func resolveForge(repos, secret, host, appID, installationID, appKeySecret, appWebhook, linearWebhook string) (server.ForgeConfig, error) { seed, err := parseForgeRepos(repos) if err != nil { return server.ForgeConfig{}, err @@ -452,9 +460,10 @@ func resolveForge(repos, secret, host, appID, installationID, appKeySecret, appW return server.ForgeConfig{}, err } return server.ForgeConfig{ - Host: host, - SeedRepos: seed, - SecretName: secret, + Host: host, + SeedRepos: seed, + SecretName: secret, + LinearWebhookSecretName: linearWebhook, App: server.ForgeAppConfig{ AppID: id, InstallationID: instID, diff --git a/go/cmd/compass-server/main_forge_test.go b/go/cmd/compass-server/main_forge_test.go index 7bb816e7..9b90ef93 100644 --- a/go/cmd/compass-server/main_forge_test.go +++ b/go/cmd/compass-server/main_forge_test.go @@ -19,7 +19,7 @@ import ( func TestResolveForgeMapping(t *testing.T) { t.Run("disabled default: no repos, no App", func(t *testing.T) { - fc, err := resolveForge("", "", "", "", "", "", "") + fc, err := resolveForge("", "", "", "", "", "", "", "") if err != nil { t.Fatalf("resolveForge: %v", err) } @@ -40,7 +40,7 @@ func TestResolveForgeMapping(t *testing.T) { t.Run("full flag mapping", func(t *testing.T) { fc, err := resolveForge("owner/repo, foo/bar", "MY_TOKEN", "ghe.example.com", - "12345", "678", "APP_KEY", "WEBHOOK_SECRET") + "12345", "678", "APP_KEY", "WEBHOOK_SECRET", "LINEAR_WEBHOOK_SECRET") if err != nil { t.Fatalf("resolveForge: %v", err) } @@ -57,6 +57,9 @@ func TestResolveForgeMapping(t *testing.T) { t.Fatalf("App secrets = %q/%q, want APP_KEY/WEBHOOK_SECRET", fc.App.AppPrivateKeySecret, fc.App.AppWebhookSecretName) } + if fc.LinearWebhookSecretName != "LINEAR_WEBHOOK_SECRET" { + t.Fatalf("LinearWebhookSecretName = %q, want LINEAR_WEBHOOK_SECRET", fc.LinearWebhookSecretName) + } want := []string{"owner/repo", "foo/bar"} if len(fc.SeedRepos) != len(want) { t.Fatalf("SeedRepos = %v, want %v", fc.SeedRepos, want) @@ -69,7 +72,7 @@ func TestResolveForgeMapping(t *testing.T) { }) t.Run("case normalization: Owner/Name lowercases to one target", func(t *testing.T) { - fc, err := resolveForge("Owner/Name", "", "", "", "", "", "") + fc, err := resolveForge("Owner/Name", "", "", "", "", "", "", "") if err != nil { t.Fatalf("resolveForge: %v", err) } @@ -91,7 +94,7 @@ func TestResolveForgeRejectsGarbage(t *testing.T) { } for _, tc := range garbage { t.Run(tc.name, func(t *testing.T) { - _, err := resolveForge(tc.repos, "", "", "", "", "", "") + _, err := resolveForge(tc.repos, "", "", "", "", "", "", "") if err == nil { t.Fatalf("resolveForge(%q) = nil error, want a startup error", tc.repos) } @@ -110,7 +113,7 @@ func TestResolveForgeRejectsBadAppID(t *testing.T) { {"non-numeric installation id", "123", "nope", "--forge-installation-id"}, } { t.Run(tc.name, func(t *testing.T) { - _, err := resolveForge("owner/repo", "", "", tc.appID, tc.installID, "", "") + _, err := resolveForge("owner/repo", "", "", tc.appID, tc.installID, "", "", "") if err == nil { t.Fatalf("resolveForge = nil error, want a startup error") } diff --git a/go/server/cors_pgtest_test.go b/go/server/cors_pgtest_test.go index 83209c83..2a133ca4 100644 --- a/go/server/cors_pgtest_test.go +++ b/go/server/cors_pgtest_test.go @@ -73,7 +73,7 @@ func buildDoorHandler(t *testing.T, corsOrigin string) http.Handler { if err != nil { t.Fatalf("otelconnect.NewInterceptor: %v", err) } - srv, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC, nil, nil) + srv, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC, nil, nil, nil) if err != nil { t.Fatalf("buildNetworkServer: %v", err) } diff --git a/go/server/linear_webhook.go b/go/server/linear_webhook.go new file mode 100644 index 00000000..7fe31148 --- /dev/null +++ b/go/server/linear_webhook.go @@ -0,0 +1,239 @@ +//go:build unix + +// The Linear webhook ingress: a single shared http.Handler for +// POST /webhooks/linear (symmetric with /webhooks/github, DL-302) that +// type-dispatches the top-level envelope. "AgentSessionEvent" routes to the +// RIG-2717 responder seam (a local SessionEventSink satisfied by +// *linearagent.Dispatcher); "Issue"/"Comment" data-change events route to an +// injected ForgeEventSink. It mirrors github_webhook.go's fail-closed order +// (secret -> HMAC -> parse) and its ack-fast discipline; T7d mounts it and +// supplies the sinks. The data sink is injected-and-nil-for-now (DL-302): a +// Linear-provider-bound notify lane wires the real sink later, so a nil sink +// acks-and-drops rather than mis-routing into the GitHub-coordinate fanout. +// +// Dedup is NOT done here: RIG-2717's design (design.md:167-170) rides the comms +// rail's own idempotency for sessions (the dispatcher's client_request_id) and +// the notify router's idempotency for data, so — unlike the GitHub handler's +// X-GitHub-Delivery LRU — this mount adds no delivery-id LRU. +package server + +import ( + "context" + "encoding/json" + "errors" + "io" + "log/slog" + "net/http" + "time" + + "github.com/RigelBuild/compass/go/internal/linearagent" +) + +const ( + // linearWebhookPath is the shared Linear ingress path this handler mounts at + // — /webhooks/linear, for symmetry with /webhooks/github (DL-302, Matt's + // 2026-08-30 amendment). + linearWebhookPath = "/webhooks/linear" + + // linearSignatureHeader carries the HMAC-SHA256 hex of the raw body under + // the webhook secret (RIG-2717 design §134). + linearSignatureHeader = "Linear-Signature" + + // linearWebhookSkew bounds how stale a webhookTimestamp may be before the + // delivery is acked-and-dropped (RIG-2717 §145): the timestamp is inside the + // signed body, so a replay is already signature-valid; a stale timestamp is + // dropped with a 200, never a 400. + linearWebhookSkew = 60 * time.Second + + // Envelope type discriminants (RIG-2717 §134, data arm design.md:660-666). + linearTypeSession = "AgentSessionEvent" + linearTypeIssue = "Issue" + linearTypeComment = "Comment" +) + +// SessionEventSink receives a verified Linear session event for asynchronous +// dispatch. *linearagent.Dispatcher satisfies it via Enqueue (dispatcher.go:145). +// Defined locally so the server can wire a real dispatcher OR pass nil when the +// RIG-2717 responder is not assembled — the handler logs-and-drops in that case. +type SessionEventSink interface { + // Enqueue offers a verified session event to the dispatcher without + // blocking; a full queue returns linearagent.ErrQueueFull, which the handler + // maps to a 500 so Linear retries the delivery. + Enqueue(ev *linearagent.SessionEvent) error +} + +// linearWebhookHandler serves POST /webhooks/linear. +type linearWebhookHandler struct { + secret func(ctx context.Context) ([]byte, error) + dataSink ForgeEventSink + sessionSink SessionEventSink + maxBody int64 + skew time.Duration + now func() time.Time + log *slog.Logger +} + +// NewLinearWebhookHandler returns the POST /webhooks/linear handler and the +// path it mounts at. secret lazily resolves the Linear webhook secret (TTL-cached by the +// caller); dataSink receives every accepted data-change event; sessionSink +// receives session events (nil when the responder is unassembled — session +// events are then logged-and-dropped). Mirrors NewGitHubWebhookHandler's +// nil-log default and field init. +func NewLinearWebhookHandler( + secret func(ctx context.Context) ([]byte, error), + dataSink ForgeEventSink, + sessionSink SessionEventSink, + log *slog.Logger, +) (string, http.Handler) { + if log == nil { + log = slog.Default() + } + return linearWebhookPath, &linearWebhookHandler{ + secret: secret, + dataSink: dataSink, + sessionSink: sessionSink, + maxBody: githubWebhookMaxBody, + skew: linearWebhookSkew, + now: time.Now, + log: log, + } +} + +// linearEnvelope is the minimal peek this handler routes on: the top-level type +// (Issue/Comment/AgentSessionEvent) and the webhookTimestamp both envelopes +// carry. The full body is re-parsed by the branch parser after routing. +type linearEnvelope struct { + Type string `json:"type"` + WebhookTimestamp int64 `json:"webhookTimestamp"` +} + +func (h *linearWebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + ctx := r.Context() + + // Bound the body read before buffering it (oversized-body rejection). A body + // over the cap trips MaxBytesReader mid-read, so a hostile Content-Length + // never forces the full allocation. + r.Body = http.MaxBytesReader(w, r.Body, h.maxBody) + body, err := io.ReadAll(r.Body) + if err != nil { + var mbe *http.MaxBytesError + if errors.As(err, &mbe) { + http.Error(w, "payload too large", http.StatusRequestEntityTooLarge) + return + } + http.Error(w, "bad request", http.StatusBadRequest) + return + } + + // Fail-closed order: secret -> HMAC -> parse. An attacker who reaches this + // unauthenticated endpoint must clear the signature before any body parsing. + secret, err := h.secret(ctx) + if err != nil { + h.log.ErrorContext(ctx, "linear webhook secret unavailable", "err", err) + http.Error(w, "webhook unavailable", http.StatusServiceUnavailable) + return + } + if !linearagent.VerifySignature(secret, body, r.Header.Get(linearSignatureHeader)) { + // Fail-closed: an unverifiable delivery is a 400, never processed. + http.Error(w, "invalid signature", http.StatusBadRequest) + return + } + + // Peek just enough to route (type) and freshness-check (timestamp); the + // branch parser re-parses the full body. + var env linearEnvelope + if err := json.Unmarshal(body, &env); err != nil { + h.log.WarnContext(ctx, "linear webhook envelope parse error", "err", err) + w.WriteHeader(http.StatusOK) // verified-but-unparseable: ack-and-drop. + return + } + + // Stale timestamp -> 200-with-drop, NOT 400 (RIG-2717 §145-155): the + // timestamp is inside the signed body, so a replay is signature-valid; a 400 + // would burn Linear's retry ladder on a legitimate late retry. + if !linearagent.CheckTimestamp(env.WebhookTimestamp, h.now(), h.skew) { + h.log.WarnContext(ctx, "linear webhook stale timestamp, dropped", + "type", env.Type, "webhook_timestamp", env.WebhookTimestamp) + w.WriteHeader(http.StatusOK) + return + } + + switch env.Type { + case linearTypeSession: + h.serveSession(ctx, w, body) + case linearTypeIssue, linearTypeComment: + h.serveData(ctx, w, body) + default: + // An envelope type this mount does not handle (e.g. "Reaction"): ignored. + w.WriteHeader(http.StatusOK) + } +} + +// serveSession routes a verified AgentSessionEvent to the responder seam. Unlike +// the data branch, Enqueue is itself non-blocking (a bounded try-send) and its +// return maps to the status code, so the status is written AFTER Enqueue: a full +// queue is a 500 (Linear retries), everything else a 200. +func (h *linearWebhookHandler) serveSession(ctx context.Context, w http.ResponseWriter, body []byte) { + ev, err := linearagent.ParseSessionEvent(body) + if err != nil { + h.log.WarnContext(ctx, "linear session event parse error", "err", err) + w.WriteHeader(http.StatusOK) // verified-but-malformed: ack-and-drop. + return + } + if h.sessionSink == nil { + // The RIG-2717 responder assembly wires a real dispatcher here; until + // then session events are acked-and-dropped so Linear does not retry. + h.log.WarnContext(ctx, "linear session responder not wired, dropping event", + "action", ev.Action, "session", ev.AgentSession.ID) + w.WriteHeader(http.StatusOK) + return + } + if err := h.sessionSink.Enqueue(ev); err != nil { + if errors.Is(err, linearagent.ErrQueueFull) { + // A full queue is a 500 so Linear retries rather than the event + // being silently dropped (dispatcher.go:30-33). + http.Error(w, "queue full", http.StatusInternalServerError) + return + } + h.log.ErrorContext(ctx, "linear session enqueue failed", "err", err) + http.Error(w, "enqueue failed", http.StatusInternalServerError) + return + } + w.WriteHeader(http.StatusOK) +} + +// serveData routes a verified Issue/Comment data-change event to the shared +// fanout sink. It acks 200 (and flushes) BEFORE the non-blocking Enqueue so the +// ack is never on the sink's latency path (mirrors github_webhook.go:181-192). +func (h *linearWebhookHandler) serveData(ctx context.Context, w http.ResponseWriter, body []byte) { + ev, ok, perr := linearagent.ParseLinearDataEvent(body) + if perr != nil { + h.log.WarnContext(ctx, "linear data event parse error", "err", perr) + w.WriteHeader(http.StatusOK) // verified-but-malformed: ack-and-drop. + return + } + // Ack fast, enqueue after: the sink must never be on the ack's latency path. + w.WriteHeader(http.StatusOK) + if !ok { + return // ignored action (e.g. Issue remove): counted-and-dropped. + } + // Flush the ack onto the wire before handing off: WriteHeader only records + // the status, so a blocking Enqueue would otherwise delay the client-visible + // 200 until ServeHTTP returns. + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + if h.dataSink == nil { + // dataSink is injected-and-nil-for-now by driver decision (DL-302): the + // GitHub-bound fanout sink would mis-route Linear events (Linear subs + // looked up under a GitHub coordinate), so the data branch acks-and-drops + // until a Linear-provider-bound notify lane injects a real sink here. + h.log.WarnContext(ctx, "linear data-change routing pending the Linear notify lane, dropping event") + return + } + h.dataSink.Enqueue(ctx, ev) +} diff --git a/go/server/linear_webhook_test.go b/go/server/linear_webhook_test.go new file mode 100644 index 00000000..169c9941 --- /dev/null +++ b/go/server/linear_webhook_test.go @@ -0,0 +1,354 @@ +//go:build unix + +// Unit tests for the shared Linear /webhooks ingress handler: signature +// fail-closed, the type-dispatch (Issue/Comment -> data sink, AgentSessionEvent +// -> session sink), the ErrQueueFull -> 500 retry signal, the 200-with-drop +// stale-timestamp rule, the nil-sessionSink logged-drop, and the ignored-type / +// ignored-action drops. +package server + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/linearagent" +) + +// recordingSessionSink records enqueued session events; enqErr (when set) is +// returned by every Enqueue so the ErrQueueFull -> 500 path is exercisable. +type recordingSessionSink struct { + mu sync.Mutex + events []*linearagent.SessionEvent + enqErr error +} + +func (s *recordingSessionSink) Enqueue(ev *linearagent.SessionEvent) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.enqErr != nil { + return s.enqErr + } + s.events = append(s.events, ev) + return nil +} + +func (s *recordingSessionSink) count() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.events) +} + +func linSign(secret, body []byte) string { + mac := hmac.New(sha256.New, secret) + mac.Write(body) + return hex.EncodeToString(mac.Sum(nil)) +} + +// linHandler builds a handler with a fixed test secret and injectable now, plus +// the two recording sinks. sessionSink may be nil to exercise the logged-drop. +func linHandler(t *testing.T, secret []byte, now time.Time, sessionNil bool) (*linearWebhookHandler, *recordingSink, *recordingSessionSink) { + t.Helper() + data := &recordingSink{} + session := &recordingSessionSink{} + var sess SessionEventSink = session + if sessionNil { + sess = nil + } + _, h := NewLinearWebhookHandler( + func(context.Context) ([]byte, error) { return secret, nil }, + data, sess, nil, + ) + lh := h.(*linearWebhookHandler) + lh.now = func() time.Time { return now } + return lh, data, session +} + +func linPost(h http.Handler, sig string, body []byte) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodPost, linearWebhookPath, strings.NewReader(string(body))) + if sig != "" { + req.Header.Set(linearSignatureHeader, sig) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +// fresh returns a webhookTimestamp (ms epoch) equal to now, so CheckTimestamp +// passes with room to spare. +func freshTS(now time.Time) int64 { return now.UnixMilli() } + +func TestLinearWebhookHandler_BadSignature(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + body := fmt.Appendf(nil, `{"type":"Issue","action":"create","webhookTimestamp":%d,"data":{"number":1,"team":{"key":"RIG"},"url":"u"}}`, freshTS(now)) + + rec := linPost(h, "deadbeef", body) + if rec.Code != http.StatusBadRequest { + t.Fatalf("code = %d, want 400", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (fail-closed)", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_IssueCreate(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + body := fmt.Appendf(nil, `{"type":"Issue","action":"create","webhookTimestamp":%d,"data":{"number":7,"team":{"key":"RIG"},"url":"https://linear.app/i/7"}}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200", rec.Code) + } + if data.count() != 1 { + t.Fatalf("data enqueued = %d, want 1", data.count()) + } + ev := data.events[0] + if ev.Provider != compassv1.ForgeProvider_FORGE_PROVIDER_LINEAR { + t.Errorf("Provider = %v, want LINEAR", ev.Provider) + } + if ev.Repo != "RIG" || ev.Number != 7 { + t.Errorf("coordinate = %q/#%d, want RIG/#7", ev.Repo, ev.Number) + } + if ev.Change != compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_OPENED { + t.Errorf("Change = %v, want OPENED", ev.Change) + } + if session.count() != 0 { + t.Errorf("session enqueued = %d, want 0", session.count()) + } +} + +func TestLinearWebhookHandler_CommentCreate(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + body := fmt.Appendf(nil, `{"type":"Comment","action":"create","webhookTimestamp":%d,"data":{"id":"c1","body":"hello","user":{"displayName":"Ann"},"issue":{"number":3,"url":"https://linear.app/i/3","team":{"key":"RIG"}}}}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200", rec.Code) + } + if data.count() != 1 { + t.Fatalf("data enqueued = %d, want 1", data.count()) + } + if ev := data.events[0]; ev.Change != compassv1internal.ForgeNotificationKind_FORGE_NOTIFICATION_KIND_COMMENT { + t.Errorf("Change = %v, want COMMENT", ev.Change) + } + if session.count() != 0 { + t.Errorf("session enqueued = %d, want 0", session.count()) + } +} + +func TestLinearWebhookHandler_SessionEvent(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + body := fmt.Appendf(nil, `{"type":"AgentSessionEvent","action":"created","webhookTimestamp":%d,"agentSession":{"id":"s1"}}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200", rec.Code) + } + if session.count() != 1 { + t.Fatalf("session enqueued = %d, want 1", session.count()) + } + if got := session.events[0].AgentSession.ID; got != "s1" { + t.Errorf("session id = %q, want s1", got) + } + if data.count() != 0 { + t.Errorf("data enqueued = %d, want 0", data.count()) + } +} + +func TestLinearWebhookHandler_SessionQueueFull(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + session.enqErr = linearagent.ErrQueueFull + body := fmt.Appendf(nil, `{"type":"AgentSessionEvent","action":"created","webhookTimestamp":%d,"agentSession":{"id":"s1"}}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusInternalServerError { + t.Fatalf("code = %d, want 500 (Linear retries on full queue)", rec.Code) + } + if data.count() != 0 { + t.Errorf("data enqueued = %d, want 0", data.count()) + } +} + +func TestLinearWebhookHandler_StaleTimestamp(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + // A timestamp 5 minutes in the past: well beyond the 60s skew. + stale := now.Add(-5 * time.Minute).UnixMilli() + body := fmt.Appendf(nil, `{"type":"Issue","action":"create","webhookTimestamp":%d,"data":{"number":7,"team":{"key":"RIG"},"url":"u"}}`, stale) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200 (stale -> 200-with-drop, not 400)", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (stale drop)", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_NilSessionSink(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, _ := linHandler(t, secret, now, true) // sessionSink nil + body := fmt.Appendf(nil, `{"type":"AgentSessionEvent","action":"created","webhookTimestamp":%d,"agentSession":{"id":"s1"}}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200 (logged-and-dropped)", rec.Code) + } + if data.count() != 0 { + t.Errorf("data enqueued = %d, want 0", data.count()) + } +} + +func TestLinearWebhookHandler_IgnoredType(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + body := fmt.Appendf(nil, `{"type":"Reaction","action":"create","webhookTimestamp":%d}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200 (ignored type)", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_IssueRemoveDrops(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + body := fmt.Appendf(nil, `{"type":"Issue","action":"remove","webhookTimestamp":%d,"data":{"number":7,"team":{"key":"RIG"},"url":"u"}}`, freshTS(now)) + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200 (remove -> ok=false drop)", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (remove drop)", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_NilDataSink(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + session := &recordingSessionSink{} + // dataSink injected nil (DL-302): the data branch must ack-and-drop, not + // panic, until a Linear-provider-bound notify lane injects a real sink. + _, h := NewLinearWebhookHandler( + func(context.Context) ([]byte, error) { return secret, nil }, + nil, session, nil, + ) + lh := h.(*linearWebhookHandler) + lh.now = func() time.Time { return now } + body := fmt.Appendf(nil, `{"type":"Issue","action":"create","webhookTimestamp":%d,"data":{"number":7,"team":{"key":"RIG"},"url":"u"}}`, freshTS(now)) + + rec := linPost(lh, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200 (nil dataSink -> 200-with-drop)", rec.Code) + } + if session.count() != 0 { + t.Errorf("session enqueued = %d, want 0", session.count()) + } +} + +func TestLinearWebhookHandler_MethodNotAllowed(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + + req := httptest.NewRequest(http.MethodGet, linearWebhookPath, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("code = %d, want 405", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (non-POST rejected)", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_BodyTooLarge(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + // A correctly-signed body just over the 1 MiB cap: MaxBytesReader must trip + // mid-read and 413 BEFORE any parse, so the memory-amplification guard on + // this unauthenticated endpoint has a regression test. + body := bytes.Repeat([]byte("a"), (1<<20)+1) + + // No signature needed: MaxBytesReader trips during io.ReadAll, before the + // HMAC verify is ever reached. + rec := linPost(h, "", body) + if rec.Code != http.StatusRequestEntityTooLarge { + t.Fatalf("code = %d, want 413 (body over 1 MiB cap)", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (over-cap rejected)", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_SecretUnavailable(t *testing.T) { + now := time.Unix(1_700_000_000, 0) + data := &recordingSink{} + session := &recordingSessionSink{} + // The secret resolver faults: fail-closed with a 503, never parsing the body. + _, h := NewLinearWebhookHandler( + func(context.Context) ([]byte, error) { return nil, errors.New("secretspec load failed") }, + data, session, nil, + ) + lh := h.(*linearWebhookHandler) + lh.now = func() time.Time { return now } + body := fmt.Appendf(nil, `{"type":"Issue","action":"create","webhookTimestamp":%d,"data":{"number":7,"team":{"key":"RIG"},"url":"u"}}`, freshTS(now)) + + // No signature needed: the secret resolver faults before the HMAC verify. + rec := linPost(lh, "", body) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("code = %d, want 503 (secret resolve fault)", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (fail-closed on secret fault)", data.count(), session.count()) + } +} + +func TestLinearWebhookHandler_VerifiedUnparseable(t *testing.T) { + secret := []byte("shh") + now := time.Unix(1_700_000_000, 0) + h, data, session := linHandler(t, secret, now, false) + // A correctly-signed but non-JSON body: verified-but-unparseable envelope + // acks 200-and-drops (Linear should not retry a malformed-but-authentic body). + body := []byte("not json at all") + + rec := linPost(h, linSign(secret, body), body) + if rec.Code != http.StatusOK { + t.Fatalf("code = %d, want 200 (verified-but-unparseable -> ack-and-drop)", rec.Code) + } + if data.count() != 0 || session.count() != 0 { + t.Errorf("enqueued data=%d session=%d, want 0/0 (unparseable drop)", data.count(), session.count()) + } +} diff --git a/go/server/network_door.go b/go/server/network_door.go index 4bf7a2dc..66b5be91 100644 --- a/go/server/network_door.go +++ b/go/server/network_door.go @@ -246,6 +246,7 @@ func buildNetworkServer( otelIC *otelconnect.Interceptor, webhookSink ForgeEventSink, webhookSecret func(ctx context.Context) ([]byte, error), + linearWebhookHandler http.Handler, ) (*http.Server, error) { handle := cfg.resolvedAdminHandle() stateDir := cfg.StateDir @@ -323,6 +324,17 @@ func buildNetworkServer( webhookPath, webhookHandler := NewGitHubWebhookHandler(webhookSecret, webhookSink, slog.Default()) netMux.Handle(webhookPath, webhookHandler) } + + // The internet-facing Linear webhook ingress (RIG-2732 T7d / RIG-2717), + // mounted only when the Linear webhook secret is declared + // (linearWebhookHandler != nil) — an App-INDEPENDENT gate. Like the GitHub + // ingress it sits OUTSIDE the bearer + admin-gate interceptors: Linear signs + // each delivery with the webhook secret (Linear-Signature), not a bearer + // token, so the handler's own VerifySignature is its whole authentication. It + // inherits withBodyReadDeadline + ReadHeaderTimeout for free (same mux). + if linearWebhookHandler != nil { + netMux.Handle(linearWebhookPath, linearWebhookHandler) + } var netRoot http.Handler = netMux if cfg.CORSAllowedOrigin != "" { // Network door defaults closed: CORS only for the one explicit diff --git a/go/server/otel_emission_pgtest_test.go b/go/server/otel_emission_pgtest_test.go index dbd8da26..0d1ac4e8 100644 --- a/go/server/otel_emission_pgtest_test.go +++ b/go/server/otel_emission_pgtest_test.go @@ -213,7 +213,7 @@ func TestNetworkDoorExposesTraceResponseHeader(t *testing.T) { srv, err := buildNetworkServer(ctx, ServeConfig{ StateDir: t.TempDir(), CORSAllowedOrigin: corsOriginForTest, - }, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC, nil, nil) + }, svc, commsSvc, secretsSvc, nil, st, admin, nil, nil, otelIC, nil, nil, nil) if err != nil { t.Fatalf("buildNetworkServer: %v", err) } diff --git a/go/server/serve.go b/go/server/serve.go index a8b47351..68efac18 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -150,6 +150,13 @@ type ForgeConfig struct { // (F1). The agent forge-WRITE path is enabled iff BOTH this and SecretName // resolve to a declared secret (Matt's 2026-08-19 ruling). ReviewerSecretName string + // LinearWebhookSecretName is the declared server_only secret NAME holding + // the Linear webhook signing secret the shared POST /webhooks ingress + // verifies deliveries against (the VALUE never crosses config or a flag). + // The Linear data-change arm runs iff this resolves to a declared secret — + // INDEPENDENT of the GitHub App gate (a deployment can run Linear + // notifications without a GitHub App and vice versa). + LinearWebhookSecretName string } // ForgeAppConfig is the GitHub App credential the board webhook-ingestion lane @@ -729,6 +736,22 @@ func buildDoors( devServer = &http.Server{Handler: devCORS().Handler(devMux), Protocols: cleartextHTTP2()} //nolint:gosec // G112: loopback dev-only door (off on the shipped path), so the Slowloris ReadHeaderTimeout does not apply here either } + // The Linear webhook ingress (RIG-2732 T7d / RIG-2717): a shared + // POST /webhooks/linear handler (DL-302) built iff the Linear webhook secret + // is declared — an App-INDEPENDENT gate (a deployment can run Linear + // notifications without a GitHub App). Its data-change arm's sink is + // injected-and-nil-for-now (DL-302): feeding the GitHub-coordinate fanout + // would mis-route Linear events, so the data branch acks-and-drops until a + // Linear-provider-bound notify lane injects a real sink. Its session arm is + // left unwired (nil sessionSink -> logged-drop) until the RIG-2717 responder + // assembly wires a *linearagent.Dispatcher here. Built here (not gated on the + // net door) so a resolve fault fail-fasts startup regardless of --listen; the + // handler is mounted only on the net door below, when one exists. + linearWebhookHandler, err := buildLinearWebhookWiring(ctx, cfg, resolver, nil, slog.Default()) + if err != nil { + return serveDoors{}, err + } + // Authenticated network door, built only when --listen is given. It mints and // writes the bootstrap token 0600 under the state dir (so a socket-only start // leaves none behind) and mounts the CompassService + CommsService behind the @@ -739,7 +762,7 @@ func buildDoors( // error the listeners this Serve bound are still ours to close. var netServer *http.Server if netListener != nil { - s, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, hub, st, adminID, netTLS, resolver, otelIC, webhookSink, webhookSecret) + s, err := buildNetworkServer(ctx, cfg, svc, commsSvc, secretsSvc, hub, st, adminID, netTLS, resolver, otelIC, webhookSink, webhookSecret, linearWebhookHandler) if err != nil { return serveDoors{}, err } @@ -900,6 +923,49 @@ func buildBoardWebhookWiring( return lane, notifyLane, sink, secret, nil } +// buildLinearWebhookWiring builds the shared Linear POST /webhooks/linear +// handler (DL-302) when the Linear webhook secret is declared — an +// App-INDEPENDENT gate (a deployment can run Linear notifications without a +// GitHub App, so this does NOT check boardIngestionEnabled). When the secret is +// undeclared it returns a nil handler and buildNetworkServer mounts no +// /webhooks/linear route. A resolve FAULT fails startup (the same fail-fast as +// forgeSecretDeclared); an absent name is the clean off-state, not an error. +// +// dataSink is the data-change arm's sink, injected-and-nil-for-now by driver +// decision (DL-302): feeding the GitHub-coordinate notify+board fanoutSink would +// mis-route Linear events (Linear subs looked up under a GitHub coordinate), so +// the handler's data branch acks-and-drops on a nil sink until a +// Linear-provider-bound notify lane injects a real sink here. The session arm is +// left unwired (nil sessionSink -> the handler logs-and-drops session events +// with a 200): the RIG-2717 responder assembly, a separate in-flight lane, wires +// a real *linearagent.Dispatcher once it assembles one in Serve. The secret is +// TTL-cached (newCachedWebhookSecret): /webhooks/linear is an internet-facing, +// unauthenticated endpoint whose secret is resolved on EVERY request BEFORE the +// HMAC check, so an uncached resolve would let a garbage POST force a full +// secretspec Load ahead of authentication. +func buildLinearWebhookWiring( + ctx context.Context, + cfg ServeConfig, + resolver secrets.Resolver, + dataSink ForgeEventSink, + log *slog.Logger, +) (http.Handler, error) { + name := cfg.Forge.LinearWebhookSecretName + if name == "" { + return nil, nil //nolint:nilnil // undeclared secret is a valid off-state: a nil handler is the signal buildNetworkServer guards on, not an ambiguous nil-nil. + } + declared, err := forgeSecretDeclared(ctx, resolver, name) + if err != nil { + return nil, err + } + if !declared { + return nil, nil //nolint:nilnil // a set-but-undeclared name is the operator's off-state; the write path's forgeSecretDeclared treats an absent optional name the same way. + } + secret := newCachedWebhookSecret(resolver, name) + _, handler := NewLinearWebhookHandler(secret, dataSink, nil, log) + return handler, nil +} + // buildBoardIngestLane assembles the App-only board webhook-ingestion lane // (RIG-2883 T5). It runs iff the GitHub App is configured (AppID != 0) AND both // App secrets — the PEM private key and the webhook signing secret — are