diff --git a/src/pubsub_middleware.ts b/src/pubsub_middleware.ts index 5ba43313..325a7365 100644 --- a/src/pubsub_middleware.ts +++ b/src/pubsub_middleware.ts @@ -150,6 +150,59 @@ const marshalPubSubRequestBody = ( }, }); +/** + * Attempt to restore a dropped CloudEvent that Eventarc sent as a Raw PubSub Message. + * + * This occurs when Eventarc's Push Subscription fails to unwrap CloudEvent HTTP + * headers. For example, if a document ID contains a trailing space, the `ce-subject` + * HTTP header is considered invalid by Node's strict HTTP parser and is stripped. + * Eventarc gracefully falls back to delivering a raw Pub/Sub message body where + * the CloudEvent metadata is buried inside the `attributes` dictionary. + * + * This method rescues the event by mutating the Express request headers (promoting + * the hidden attributes into standard `ce-` HTTP headers). This guarantees that + * downstream middlewares (like `isBinaryCloudEvent()`) will successfully validate + * the request as a Binary Cloud Event. + * + * @param req - Express request object to hydrate with CloudEvent headers + * @param body - An unmarshalled http request body from a Pub/Sub push subscription + * @param attributes - The Pub/Sub message attributes + */ +const rehydrateCloudEvent = ( + req: Request, + body: RawPubSubBody, + attributes: {[key: string]: string} +): void => { + // 1. Promote attributes to HTTP headers. The official CloudEvents Pub/Sub Protocol + // Binding dictates that attributes map without the `ce-` prefix. However, some + // Google Cloud systems attach them with the `ce-` prefix. We handle both. + for (const [key, value] of Object.entries(attributes)) { + if (key.startsWith('ce-')) { + req.headers[key.toLowerCase()] = value; + } else if (['type', 'source', 'subject', 'id', 'time', 'specversion', 'datacontenttype'].includes(key.toLowerCase())) { + req.headers['ce-' + key.toLowerCase()] = value; + } + } + + // 2. Extract the binary payload. + const dataBuf = Buffer.from(body.message.data || '', 'base64'); + const contentType = req.headers['ce-datacontenttype'] || req.headers['content-type'] || ''; + + // 3. Attempt JSON parse. If it fails (e.g. malformed JSON or a mismatched content-type + // like application/protobuf), fallback to the raw Buffer. This mirrors `bodyParser.raw()` + // and safely passes the binary payload to the user's function without crashing the HTTP stream. + if (typeof contentType === 'string' && contentType.includes('application/json')) { + try { + req.body = JSON.parse(dataBuf.toString('utf8')); + return; + } catch { + req.body = dataBuf; + return; + } + } + req.body = dataBuf; +}; + /** * Express middleware used to marshal the HTTP request body received directly from a * Pub/Sub subscription into the format that is expected downstream by wrapEventFunction @@ -164,7 +217,12 @@ export const legacyPubSubEventMiddleware = ( ) => { const {body, path} = req; if (isRawPubSubRequestBody(body) && !isBinaryCloudEvent(req)) { - req.body = marshalPubSubRequestBody(body, path); + const pubsubAttributes = body.message.attributes || {}; + if ((pubsubAttributes['ce-type'] || pubsubAttributes['type']) && (pubsubAttributes['ce-source'] || pubsubAttributes['source'])) { + rehydrateCloudEvent(req, body, pubsubAttributes); + } else { + req.body = marshalPubSubRequestBody(body, path); + } } next(); }; diff --git a/test/pubsub_middleware.ts b/test/pubsub_middleware.ts index c9fbe6dc..ff01941f 100644 --- a/test/pubsub_middleware.ts +++ b/test/pubsub_middleware.ts @@ -65,6 +65,7 @@ describe('legacyPubSubEventMiddleware', () => { path: string; body: object; expectedBody: () => object; + expectedHeaders?: () => object; } const testData: TestData[] = [ @@ -86,6 +87,52 @@ describe('legacyPubSubEventMiddleware', () => { body: {foo: 'bar'}, expectedBody: () => ({foo: 'bar'}), }, + { + name: 'Eventarc fallback CloudEvent (JSON payload)', + path: `/${PUB_SUB_TOPIC}`, + body: { + subscription: 'projects/FOO/subscriptions/BAR_SUB', + message: { + data: Buffer.from('{"hello":"world"}', 'utf8').toString('base64'), + messageId: '1', + attributes: { + 'ce-type': 'google.cloud.firestore.document.v1.written', + 'ce-source': '//firestore.googleapis.com/...', + 'ce-subject': 'documents/... ', + 'ce-datacontenttype': 'application/json', + }, + }, + }, + expectedBody: () => ({hello: 'world'}), + expectedHeaders: () => ({ + 'ce-type': 'google.cloud.firestore.document.v1.written', + 'ce-source': '//firestore.googleapis.com/...', + 'ce-subject': 'documents/... ', + 'ce-datacontenttype': 'application/json', + }), + }, + { + name: 'Eventarc fallback CloudEvent (Unprefixed attributes & Binary payload)', + path: `/${PUB_SUB_TOPIC}`, + body: { + subscription: 'projects/FOO/subscriptions/BAR_SUB', + message: { + data: Buffer.from('bad-json', 'utf8').toString('base64'), + messageId: '1', + attributes: { + 'type': 'google.cloud.storage.object.v1.finalized', + 'source': '//storage.googleapis.com/...', + 'datacontenttype': 'application/protobuf', + }, + }, + }, + expectedBody: () => Buffer.from('bad-json', 'utf8'), + expectedHeaders: () => ({ + 'ce-type': 'google.cloud.storage.object.v1.finalized', + 'ce-source': '//storage.googleapis.com/...', + 'ce-datacontenttype': 'application/protobuf', + }), + }, ]; testData.forEach((test: TestData) => { @@ -95,11 +142,15 @@ describe('legacyPubSubEventMiddleware', () => { const request = { path: test.path, body: test.body, + headers: {} as {[key: string]: string}, // eslint-disable-next-line @typescript-eslint/no-unused-vars header: (_: string) => '', }; legacyPubSubEventMiddleware(request as Request, {} as Response, next); assert.deepStrictEqual(request.body, test.expectedBody()); + if (test.expectedHeaders) { + assert.deepStrictEqual(request.headers, test.expectedHeaders()); + } assert.strictEqual(next.called, true); clock.restore(); });