feat: Rest hook Lambda

.eslintrc.js

@@ -25,6 +25,11 @@ module.exports = {
         'no-empty-function': 'off',
         '@typescript-eslint/no-empty-function': 'error',
         'import/no-extraneous-dependencies': ['error', { devDependencies: ['**/*.test.ts', 'integration-tests/*'] }],
+        // @types/aws-lambda is special since aws-lambda is not the name of a package that we take as a dependency.
+        // Making eslint recognize it would require several additional plugins and it's not worth setting it up right now.
+        // See https://github.com/typescript-eslint/typescript-eslint/issues/1624
+        // eslint-disable-next-line import/no-unresolved
+        'import/no-unresolved': ['error', { ignore: ['aws-lambda'] }],
         'no-shadow': 'off', // replaced by ts-eslint rule below
         '@typescript-eslint/no-shadow': 'error',
     },

cloudformation/subscriptions.yaml

@@ -96,4 +96,43 @@ Resources:
       Protocol: sqs
       FilterPolicy:
         channelType:
-          - 'rest-hook'
+          - 'rest-hook'
+
+  RestHookLambdaRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+          - Effect: 'Allow'
+            Principal:
+              Service: 'lambda.amazonaws.com'
+            Action: 'sts:AssumeRole'
+      Policies:
+        - PolicyName: 'restHookLambdaPolicy'
+          PolicyDocument:
+            Version: '2012-10-17'
+            Statement:
+              - Effect: Allow
+                Action:
+                  - logs:CreateLogStream
+                  - logs:CreateLogGroup
+                  - logs:PutLogEvents
+                Resource: !Sub 'arn:${AWS::Partition}:logs:${AWS::Region}:*:*'
+              - Effect: Allow
+                Action:
+                  - 'xray:PutTraceSegments'
+                  - 'xray:PutTelemetryRecords'
+                Resource:
+                  - '*'
+              - Effect: Allow
+                Action:
+                  - 'kms:Decrypt'
+                Resource:
+                  - !GetAtt SubscriptionsKey.Arn
+              - Effect: Allow
+                Action:
+                  - 'sqs:DeleteMessage'
+                  - 'sqs:ReceiveMessage'
+                  - 'sqs:GetQueueAttributes'
+                Resource: !GetAtt RestHookQueue.Arn

package.json

@@ -33,13 +33,15 @@
     "serverless-info": "serverless info"
   },
   "dependencies": {
+    "@types/aws-lambda": "^8.10.92",
     "aws-sdk": "^2.1000.0",
     "axios": "^0.21.4",
     "fhir-works-on-aws-authz-rbac": "5.0.0",
     "fhir-works-on-aws-interface": "11.3.0",
     "fhir-works-on-aws-persistence-ddb": "3.9.0",
     "fhir-works-on-aws-routing": "6.3.0",
     "fhir-works-on-aws-search-es": "3.9.2",
+    "lodash": "^4.17.21",
     "serverless-http": "^2.7.0",
     "yargs": "^16.2.0"
   },
@@ -62,7 +64,6 @@
     "jest-circus": "^26.6.3",
     "jest-mock-extended": "^1.0.8",
     "jsonwebtoken": "^8.5.1",
-    "lodash": "^4.17.21",
     "prettier": "^2.4.1",
     "qs": "^6.10.1",
     "serverless": "2.64.1",

serverless.yaml

@@ -192,6 +192,17 @@ functions:
       ELASTICSEARCH_DOMAIN_ENDPOINT: !Join ['', ['https://', !GetAtt ElasticSearchDomain.DomainEndpoint]]
       NUMBER_OF_SHARDS: !If [isDev, 1, 3] # 133 indices, one per resource types
 
+  subscriptionsRestHook:
+    timeout: 20
+    runtime: nodejs14.x
+    description: 'Send rest-hook notification for subscription'
+    role: RestHookLambdaRole
+    handler: src/subscriptions/index.handler
+    events:
+      - sqs:
+          arn:
+            !GetAtt RestHookQueue.Arn
+
 stepFunctions:
   stateMachines:
     BulkExportStateMachine: ${file(bulkExport/state-machine-definition.yaml)}

src/config.ts

@@ -23,9 +23,9 @@ import {
 import JsonSchemaValidator from 'fhir-works-on-aws-routing/lib/router/validation/jsonSchemaValidator';
 import HapiFhirLambdaValidator from 'fhir-works-on-aws-routing/lib/router/validation/hapiFhirLambdaValidator';
 import SubscriptionValidator from 'fhir-works-on-aws-routing/lib/router/validation/subscriptionValidator';
+import getAllowListedSubscriptionEndpoints from './subscriptions/allowList';
 import RBACRules from './RBACRules';
 import { loadImplementationGuides } from './implementationGuides/loadCompiledIGs';
-import getAllowListedSubscriptionEndpoints from './subscriptions/allowList';
 
 const { IS_OFFLINE, ENABLE_MULTI_TENANCY, ENABLE_SUBSCRIPTIONS } = process.env;
 

src/index.ts

@@ -32,3 +32,5 @@ exports.handler = async (event: any = {}, context: any = {}): Promise<any> => {
     await ensureAsyncInit(serverlessHandler);
     return (await serverlessHandler)(event, context);
 };
+
+export default ensureAsyncInit;

src/subscriptions/allowListUtil.ts

@@ -0,0 +1,85 @@
+import { SubscriptionEndpoint } from 'fhir-works-on-aws-routing/lib/router/validation/subscriptionValidator';
+import { groupBy } from 'lodash';
+import { makeLogger } from 'fhir-works-on-aws-interface';
+import getAllowListedSubscriptionEndpoints from './allowList';
+
+const SINGLE_TENANT_ALLOW_LIST_KEY = 'SINGLE_TENANT_ALLOW_LIST_KEY';
+const logger = makeLogger({ component: 'subscriptions' });
+
+export interface AllowListInfo {
+    allowList: (string | RegExp)[];
+    headerMap: { [key: string]: string[] };
+}
+
+const extractAllowListInfo = (subscriptionEndpoints: SubscriptionEndpoint[]): AllowListInfo => {
+    const allowList: (string | RegExp)[] = [];
+    const headerMap: { [key: string]: string[] } = {};
+    subscriptionEndpoints.forEach((allowEndpoint: SubscriptionEndpoint) => {
+        allowList.push(allowEndpoint.endpoint);
+        headerMap[allowEndpoint.endpoint.toString()] = allowEndpoint.headers || [];
+    });
+    return { allowList, headerMap };
+};
+
+export async function getAllowListInfo({
+    enableMultitenancy = false,
+}: {
+    enableMultitenancy: boolean;
+}): Promise<{ [key: string]: AllowListInfo }> {
+    const originalAllowList = await getAllowListedSubscriptionEndpoints();
+    logger.debug(originalAllowList);
+    if (!enableMultitenancy) {
+        return { [SINGLE_TENANT_ALLOW_LIST_KEY]: extractAllowListInfo(originalAllowList) };
+    }
+    const allowListInfo: { [key: string]: AllowListInfo } = {};
+    const endpointsGroupByTenant: { [key: string]: SubscriptionEndpoint[] } = groupBy(
+        originalAllowList,
+        (allowEndpoint: SubscriptionEndpoint) => allowEndpoint.tenantId,
+    );
+    Object.entries(endpointsGroupByTenant).forEach(([key, value]) => {
+        allowListInfo[key] = extractAllowListInfo(value);
+    });
+    return allowListInfo;
+}
+
+/**
+ * Verify endpoint is allow listed
+ * Return allow list headers if endpoint is allow listed
+ * Throw error if endpoint is not allow listed
+ * @param allowListInfoMap
+ * @param endpoint
+ * @param tenantId
+ * @param enableMultitenancy
+ */
+export const getAllowListHeaders = (
+    allowListInfoMap: { [key: string]: AllowListInfo },
+    endpoint: string,
+    { enableMultitenancy = false, tenantId }: { enableMultitenancy: boolean; tenantId: string | undefined },
+): string[] => {
+    const getHeaders = (allowListInfo: AllowListInfo): string[] => {
+        if (allowListInfo) {
+            const { allowList, headerMap } = allowListInfo;
+            // eslint-disable-next-line no-restricted-syntax
+            for (const allowedEndpoint of allowList) {
+                if (allowedEndpoint instanceof RegExp && allowedEndpoint.test(endpoint)) {
+                    return headerMap[allowedEndpoint.toString()];
+                }
+                if (allowedEndpoint === endpoint) {
+                    return headerMap[allowedEndpoint];
+                }
+            }
+        }
+        throw new Error(`Endpoint ${endpoint} is not allow listed.`);
+    };
+
+    if (enableMultitenancy) {
+        if (tenantId) {
+            return getHeaders(allowListInfoMap[tenantId]);
+        }
+        throw new Error('This instance has multi-tenancy enabled, but the incoming request is missing tenantId');
+    }
+    if (!tenantId) {
+        return getHeaders(allowListInfoMap[SINGLE_TENANT_ALLOW_LIST_KEY]);
+    }
+    throw new Error('This instance has multi-tenancy disabled, but the incoming request has a tenantId');
+};

src/subscriptions/index.ts

@@ -0,0 +1,14 @@
+import RestHookHandler from './restHook';
+import { AllowListInfo, getAllowListInfo } from './allowListUtil';
+
+const enableMultitenancy = process.env.ENABLE_MULTI_TENANCY === 'true';
+
+const allowListPromise: Promise<{ [key: string]: AllowListInfo }> = getAllowListInfo({
+    enableMultitenancy,
+});
+
+const restHookHandler = new RestHookHandler({ enableMultitenancy });
+
+exports.handler = async (event: any) => {
+    return restHookHandler.sendRestHookNotification(event, allowListPromise);
+};

src/subscriptions/restHook.test.ts

@@ -0,0 +1,137 @@
+import axios from 'axios';
+import RestHookHandler from './restHook';
+import { AllowListInfo, getAllowListInfo } from './allowListUtil';
+
+jest.mock('axios');
+// This mock only works on the file level for once
+// Separating multi-tenant tests to a separate file to use other mock value
+jest.mock('./allowList', () => ({
+    __esModule: true,
+    default: async () => [
+        {
+            endpoint: 'https://fake-end-point-1',
+            headers: ['header-name-1: header-value-1'],
+        },
+        {
+            endpoint: new RegExp('^https://fake-end-point-2'),
+            headers: ['header-name-2: header-value-2'],
+        },
+    ],
+}));
+
+const getEvent = ({
+    channelHeader = ['testKey:testValue'],
+    channelPayload = 'application/fhir+json',
+    endpoint = 'https://fake-end-point-1',
+    tenantId = null as any,
+} = {}) => ({
+    Records: [
+        {
+            messageId: 'fake-message-id',
+            receiptHandle: 'fake-receipt-Handle',
+            body: JSON.stringify({
+                Message: JSON.stringify({
+                    subscriptionId: 123456,
+                    channelType: 'rest-hook',
+                    tenantId,
+                    endpoint,
+                    channelPayload,
+                    channelHeader,
+                    matchedResource: {
+                        id: 1234567,
+                        resourceType: 'Patient',
+                        versionId: 2,
+                        lastUpdated: 'some-time-stamp',
+                    },
+                }),
+            }),
+            attributes: {
+                ApproximateReceiveCount: '1',
+                SentTimestamp: '123456789',
+                SenderId: 'FAKESENDERID',
+                MessageDeduplicationId: '1',
+                ApproximateFirstReceiveTimestamp: '123456789',
+            },
+            messageAttributes: {},
+            md5OfBody: '123456789012',
+            eventSource: 'aws:sqs',
+            eventSourceARN: 'arn:aws:sqs:us-east-2:123456789012:fhir-service-dev-RestHookQueue',
+            awsRegion: 'us-east-2',
+        },
+    ],
+});
+
+describe('Single tenant: Rest hook notification', () => {
+    const restHookHandler = new RestHookHandler({ enableMultitenancy: false });
+    const allowListPromise: Promise<{ [key: string]: AllowListInfo }> = getAllowListInfo({ enableMultitenancy: false });
+
+    beforeEach(() => {
+        axios.post = jest.fn().mockResolvedValueOnce({ data: { message: 'POST Successful' } });
+        axios.put = jest.fn().mockResolvedValueOnce({ data: { message: 'PUT Successful' } });
+    });
+
+    test('Empty POST notification is sent when channelPayload is null', async () => {
+        await expect(
+            restHookHandler.sendRestHookNotification(getEvent({ channelPayload: null as any }), allowListPromise),
+        ).resolves.toEqual([{ message: 'POST Successful' }]);
+        expect(axios.post).toHaveBeenCalledWith('https://fake-end-point-1', null, {
+            headers: { 'header-name-1': ' header-value-1', testKey: 'testValue' },
+        });
+    });
+
+    test('PUT notification with ID is sent when channelPayload is application/fhir+json', async () => {
+        await expect(
+            restHookHandler.sendRestHookNotification(
+                getEvent({ endpoint: 'https://fake-end-point-2-something' }),
+                allowListPromise,
+            ),
+        ).resolves.toEqual([{ message: 'PUT Successful' }]);
+        expect(axios.put).toHaveBeenCalledWith('https://fake-end-point-2-something/Patient/1234567', null, {
+            headers: { 'header-name-2': ' header-value-2', testKey: 'testValue' },
+        });
+    });
+
+    test('Header in channelHeader overrides header in allow list when there is duplicated header name', async () => {
+        await expect(
+            restHookHandler.sendRestHookNotification(
+                getEvent({
+                    channelHeader: ['header-name-2: header-value-2-something'],
+                    endpoint: 'https://fake-end-point-2-something',
+                }),
+                allowListPromise,
+            ),
+        ).resolves.toEqual([{ message: 'PUT Successful' }]);
+        expect(axios.put).toHaveBeenCalledWith('https://fake-end-point-2-something/Patient/1234567', null, {
+            headers: { 'header-name-2': ' header-value-2-something' },
+        });
+    });
+
+    test('Header string without colon is sent as empty header', async () => {
+        await expect(
+            restHookHandler.sendRestHookNotification(
+                getEvent({ endpoint: 'https://fake-end-point-2-something', channelHeader: ['testKey'] }),
+                allowListPromise,
+            ),
+        ).resolves.toEqual([{ message: 'PUT Successful' }]);
+        expect(axios.put).toHaveBeenCalledWith('https://fake-end-point-2-something/Patient/1234567', null, {
+            headers: { 'header-name-2': ' header-value-2', testKey: '' },
+        });
+    });
+
+    test('Error thrown when endpoint is not allow listed', async () => {
+        await expect(
+            restHookHandler.sendRestHookNotification(
+                getEvent({ endpoint: 'https://fake-end-point-3' }),
+                allowListPromise,
+            ),
+        ).rejects.toThrow(new Error('Endpoint https://fake-end-point-3 is not allow listed.'));
+    });
+
+    test('Error thrown when tenantID is passed in', async () => {
+        await expect(
+            restHookHandler.sendRestHookNotification(getEvent({ tenantId: 'tenant1' }), allowListPromise),
+        ).rejects.toThrow(
+            new Error('This instance has multi-tenancy disabled, but the incoming request has a tenantId'),
+        );
+    });
+});