fix(ingest): remove webhook response filtering (#440)

* revert filtering on webhook response

* test(ingest): cover cross-address webhook output

* fix(ingest): retain webhook response safeguards

Author: matheus1lva

packages/ingest/extract/webhook.spec.ts

@@ -1,11 +1,18 @@
 import { strict as assert } from 'node:assert'
-import { selectValidOutputs, readJsonCapped, MAX_OUTPUTS_PER_VAULT } from './webhook'
+import { mq } from 'lib'
+import {
+  MAX_OUTPUT_GROUPS,
+  MAX_OUTPUTS_PER_ADDRESS,
+  MAX_TOTAL_OUTPUTS,
+  readJsonCapped,
+  selectValidOutputs,
+  WebhookExtractor
+} from './webhook'
 import type { Data } from './webhook'
 import type { Output } from 'lib/types'
 import type { WebhookSubscription } from 'lib/subscriptions'
 
 const VAULT_A = '0x1111111111111111111111111111111111111111'
-const VAULT_B = '0x2222222222222222222222222222222222222222'
 const OUT_OF_SCOPE = '0x3333333333333333333333333333333333333333'
 
 const subscription: WebhookSubscription = {
@@ -40,35 +47,61 @@ function output(address: string, label = 'apr', chainId = 1): Output {
 }
 
 describe('selectValidOutputs', () => {
-  it('keeps outputs for requested vaults with allowed labels', () => {
-    const valid = selectValidOutputs([output(VAULT_A), output(VAULT_B)], data([VAULT_A, VAULT_B]))
-    assert.equal(valid.length, 2)
+  it('keeps configured-label outputs for an address other than the triggering vault', () => {
+    assert.deepEqual(selectValidOutputs([output(OUT_OF_SCOPE)], data([VAULT_A])), [output(OUT_OF_SCOPE)])
   })
 
-  it('drops outputs for vaults that were not requested', () => {
-    const valid = selectValidOutputs([output(OUT_OF_SCOPE)], data([VAULT_A]))
-    assert.equal(valid.length, 0)
+  it('drops outputs for a different chain than requested', () => {
+    assert.deepEqual(selectValidOutputs([output(OUT_OF_SCOPE, 'apr', 10)], data([VAULT_A])), [])
   })
 
-  it('drops outputs for a different chain than requested', () => {
-    const valid = selectValidOutputs([output(VAULT_A, 'apr', 10)], data([VAULT_A], 1))
-    assert.equal(valid.length, 0)
+  it('drops a group with an unexpected label', () => {
+    assert.deepEqual(selectValidOutputs([output(OUT_OF_SCOPE, 'not-allowed')], data([VAULT_A])), [])
+  })
+
+  it('drops a group over the per-address cap', () => {
+    const many = Array.from({ length: MAX_OUTPUTS_PER_ADDRESS + 1 }, () => output(OUT_OF_SCOPE))
+    assert.deepEqual(selectValidOutputs(many, data([VAULT_A])), [])
   })
 
-  it('matches vault addresses case-insensitively', () => {
-    const valid = selectValidOutputs([output(VAULT_A.toUpperCase().replace('0X', '0x'))], data([VAULT_A.toLowerCase()]))
-    assert.equal(valid.length, 1)
+  it('drops a response over the total output cap', () => {
+    const many = Array.from({ length: MAX_TOTAL_OUTPUTS + 1 }, () => output(OUT_OF_SCOPE))
+    assert.deepEqual(selectValidOutputs(many, data([VAULT_A])), [])
   })
 
-  it('drops a vault group with an unexpected label', () => {
-    const valid = selectValidOutputs([output(VAULT_A, 'not-allowed')], data([VAULT_A]))
-    assert.equal(valid.length, 0)
+  it('drops a response over the output group cap', () => {
+    const many = Array.from({ length: MAX_OUTPUT_GROUPS + 1 }, (_, i) => output(`0x${i.toString(16).padStart(40, '0')}`))
+    assert.deepEqual(selectValidOutputs(many, data([VAULT_A])), [])
   })
+})
+
+describe('WebhookExtractor', () => {
+  it('loads a cross-address output without composition lookup', async () => {
+    const originalFetch = globalThis.fetch
+    const originalSecret = process.env.WEBHOOK_SECRET_S_TEST
+    const responseOutput = {
+      ...output(OUT_OF_SCOPE),
+      blockNumber: '1',
+      blockTime: '1'
+    }
+    const add = vi.spyOn(mq, 'add').mockResolvedValue({} as never)
+    globalThis.fetch = vi.fn().mockResolvedValue(new Response(JSON.stringify([responseOutput])))
+    process.env.WEBHOOK_SECRET_S_TEST = 'secret'
+
+    try {
+      await new WebhookExtractor().extract(data([VAULT_A]))
 
-  it('drops a vault group over the per-vault cap', () => {
-    const many = Array.from({ length: MAX_OUTPUTS_PER_VAULT + 1 }, () => output(VAULT_A))
-    const valid = selectValidOutputs(many, data([VAULT_A]))
-    assert.equal(valid.length, 0)
+      assert.equal(add.mock.calls.length, 1)
+      assert.deepEqual(add.mock.calls[0], [
+        mq.job.load.output,
+        { batch: [output(OUT_OF_SCOPE)] }
+      ])
+    } finally {
+      add.mockRestore()
+      globalThis.fetch = originalFetch
+      if (originalSecret === undefined) delete process.env.WEBHOOK_SECRET_S_TEST
+      else process.env.WEBHOOK_SECRET_S_TEST = originalSecret
+    }
   })
 })
 

packages/ingest/extract/webhook.ts

@@ -1,8 +1,8 @@
-import { z } from 'zod'
 import { createHmac } from 'crypto'
 import { mq, sentry } from 'lib'
-import { Output, OutputSchema, zhexstring } from 'lib/types'
 import { WebhookSubscription, WebhookSubscriptionSchema } from 'lib/subscriptions'
+import { Output, OutputSchema, zhexstring } from 'lib/types'
+import { z } from 'zod'
 
 export const DataSchema = z.object({
   abiPath: z.string(),
@@ -81,33 +81,53 @@ export async function readJsonCapped(response: Response, maxBytes = MAX_RESPONSE
   return JSON.parse(Buffer.concat(chunks).toString('utf8'))
 }
 
-export const MAX_OUTPUTS_PER_VAULT = 100
+export const MAX_OUTPUTS_PER_ADDRESS = 100
+export const MAX_OUTPUT_GROUPS = 1_000
+export const MAX_TOTAL_OUTPUTS = 10_000
 
-// Drop any receiver output that isn't for the chain/vaults we asked it to compute,
-// exceeds the per-vault cap, or carries an unexpected label, before it reaches the
-// load queue. Scope filtering also bounds total groups to the requested vault set
-// (FINDINGS.md findings 4-5, CWE-20 / CWE-400).
+// A configured webhook is a trusted producer for its configured labels. It may
+// publish for any address on the requested chain, but each response remains
+// bounded before it reaches the load queue.
 export function selectValidOutputs(outputs: Output[], data: Data): Output[] {
   const { subscription } = data
-  const requestedVaults = new Set(data.vaults.map(v => v.toLowerCase()))
-  const grouped = Map.groupBy(outputs, o => `${o.chainId}:${o.address}`)
+  if (outputs.length > MAX_TOTAL_OUTPUTS) {
+    console.error(`🤬 ${subscription.id} skipping response: ${outputs.length} outputs > ${MAX_TOTAL_OUTPUTS}`)
+    sentry.captureMessage('WEBHOOK_TOTAL_OUTPUTS_OVER_LIMIT', {
+      level: 'warning',
+      tags: { component: 'ingest', job: 'extract.webhook' },
+      extra: { subscriptionId: subscription.id, outputs: outputs.length, max: MAX_TOTAL_OUTPUTS }
+    })
+    return []
+  }
+
+  const grouped = Map.groupBy(outputs, o => `${o.chainId}:${o.address.toLowerCase()}`)
+  if (grouped.size > MAX_OUTPUT_GROUPS) {
+    console.error(`🤬 ${subscription.id} skipping response: ${grouped.size} output groups > ${MAX_OUTPUT_GROUPS}`)
+    sentry.captureMessage('WEBHOOK_OUTPUT_GROUPS_OVER_LIMIT', {
+      level: 'warning',
+      tags: { component: 'ingest', job: 'extract.webhook' },
+      extra: { subscriptionId: subscription.id, groups: grouped.size, max: MAX_OUTPUT_GROUPS }
+    })
+    return []
+  }
+
   return [...grouped].flatMap(([key, group]) => {
     const [first] = group
-    if (first.chainId !== data.chainId || !requestedVaults.has(first.address.toLowerCase())) {
-      console.error(`🤬 ${subscription.id} skipping ${key}: out of requested scope`)
-      sentry.captureMessage('WEBHOOK_OUT_OF_SCOPE', {
+    if (first.chainId !== data.chainId) {
+      console.error(`🤬 ${subscription.id} skipping ${key}: unexpected chain`)
+      sentry.captureMessage('WEBHOOK_UNEXPECTED_CHAIN', {
         level: 'warning',
         tags: { component: 'ingest', job: 'extract.webhook' },
         extra: { subscriptionId: subscription.id, key, requestedChainId: data.chainId }
       })
       return []
     }
-    if (group.length > MAX_OUTPUTS_PER_VAULT) {
-      console.error(`🤬 ${subscription.id} skipping ${key}: ${group.length} outputs > ${MAX_OUTPUTS_PER_VAULT}`)
+    if (group.length > MAX_OUTPUTS_PER_ADDRESS) {
+      console.error(`🤬 ${subscription.id} skipping ${key}: ${group.length} outputs > ${MAX_OUTPUTS_PER_ADDRESS}`)
       sentry.captureMessage('WEBHOOK_OUTPUTS_OVER_LIMIT', {
         level: 'warning',
         tags: { component: 'ingest', job: 'extract.webhook' },
-        extra: { subscriptionId: subscription.id, key, outputs: group.length, max: MAX_OUTPUTS_PER_VAULT }
+        extra: { subscriptionId: subscription.id, key, outputs: group.length, max: MAX_OUTPUTS_PER_ADDRESS }
       })
       return []
     }
@@ -144,9 +164,8 @@ export class WebhookExtractor {
 
       const body = await readJsonCapped(response)
       const outputs = OutputSchema.array().parse(body)
-      const valid = selectValidOutputs(outputs, data)
 
-      await mq.add(mq.job.load.output, { batch: valid })
+      await mq.add(mq.job.load.output, { batch: selectValidOutputs(outputs, data) })
     } finally {
       semaphore.release()
     }