Merge pull request #24 from kikuomax/quarantine

Quarantine unsuccessful inbox payloads

cdk/lambda/receive_inbound_activity/index.py

@@ -7,11 +7,16 @@
 * ``OBJECTS_BUCKET_NAME``: name of the S3 bucket that stores received objects.
 * ``DOMAIN_NAME_PARAMETER_PATH``: path to the parameter that stores the domain
   name in Parameter Store on AWS Systems Manager.
+* ``QUARANTINE_BUCKET_NAME``: name of the S3 bucket that stores quarantined
+  payloads.
 """
 
+import base64
+import hashlib
 import json
 import logging
 import os
+from typing import Any, Optional
 import boto3
 from libactivitypub.activity import Activity
 from libactivitypub.actor import Actor
@@ -27,7 +32,7 @@
 )
 from libmumble.parameters import get_domain_name
 from libmumble.user_table import UserTable
-from libmumble.utils import to_urlsafe_base64
+from libmumble.utils import current_yyyymmdd_hhmmss_ssssss, to_urlsafe_base64
 import requests
 
 
@@ -46,6 +51,9 @@
 # bucket for objects
 OBJECTS_BUCKET_NAME = os.environ['OBJECTS_BUCKET_NAME']
 
+# bucket for quarantined payloads
+QUARANTINE_BUCKET_NAME = os.environ['QUARANTINE_BUCKET_NAME']
+
 
 def save_activity_in_inbox(activity_data: str, digest: str, recipient: str):
     """Saves a given activity in the inbox of a specified user.
@@ -77,6 +85,32 @@ def save_activity_in_inbox(activity_data: str, digest: str, recipient: str):
     LOGGER.debug('saved activity: %s', res)
 
 
+def quarantine(tag: str, payload: Any, options: Optional[Any]=None):
+    """Saves a given payload in the quarantine bucket.
+    """
+    data = {
+        'tag': tag,
+        'datetime': current_yyyymmdd_hhmmss_ssssss(),
+        'action': 'receive_inbound_activity',
+        'payload': payload,
+    }
+    if options is not None:
+        data['options'] = options
+    json_data = json.dumps(data).encode('utf-8')
+    digest = hashlib.sha256(json_data).digest()
+    object_name = base64.urlsafe_b64encode(digest).decode('utf-8').rstrip('=')
+    object_key = f'inbox/{object_name}.json'
+    LOGGER.debug('saving quarantined payload: %s', object_key)
+    s3_client = boto3.client('s3')
+    res = s3_client.put_object(
+        Bucket=QUARANTINE_BUCKET_NAME,
+        Key=object_key,
+        Body=json_data,
+        ChecksumSHA256=base64.b64encode(digest).decode('utf-8'),
+    )
+    LOGGER.debug('quarantined payload: %s', res)
+
+
 def lambda_handler(event, _context):
     """Runs on AWS Lambda.
 
@@ -119,24 +153,29 @@ def lambda_handler(event, _context):
     try:
         signature = parse_signature(event['signature'])
     except ValueError as exc:
+        quarantine('bad_signature', event)
         raise UnauthorizedError(f'bad signature: {exc}') from exc
 
     LOGGER.debug('resolving signer: %s', signature['key_id'])
     try:
         signer = Actor.resolve_uri(signature['key_id'])
     except requests.HTTPError as exc:
+        quarantine('bad_signer', event)
         raise UnauthorizedError(
             f'failed to resolve signer: {signature["key_id"]}',
         ) from exc
     except ValueError as exc:
+        quarantine('bad_signer_format', event)
         raise UnauthorizedError(f'invalid actor: {exc}') from exc
 
     LOGGER.debug('loading public key')
     try:
         public_key = signer.public_key
     except (AttributeError, TypeError) as exc:
+        quarantine('bad_signer_format', event, signer.to_dict())
         raise UnauthorizedError(f'invalid actor: {exc}') from exc
     if public_key['id'] != signature['key_id']:
+        quarantine('bad_signer_format', event, signer.to_dict())
         raise UnauthorizedError(f'key ID mismatch: {signature["key_id"]}')
 
     LOGGER.debug('verifying signature')
@@ -155,26 +194,31 @@ def lambda_handler(event, _context):
             },
         )
     except (KeyError, ValueError, VerificationError) as exc:
+        quarantine('invalid_signature', event)
         raise UnauthorizedError(f'failed to authenticate: {exc}') from exc
 
     # once the signature is verified, we can parse the body
     LOGGER.debug('parsing activity')
     try:
         activity = Activity.parse_object(json.loads(body))
     except ValueError as exc:
+        quarantine('invalid_activity', event)
         raise BadRequestError(f'{exc}') from exc
     if signer.id != activity.actor_id:
+        quarantine('invalid_activity', event, activity.to_dict())
         raise UnauthorizedError(
             f'signer and actor mismatch: {signer.id} != {activity.actor_id}',
         )
 
     LOGGER.debug('looking up user: %s', username)
     user = USER_TABLE.find_user_by_username(username, DOMAIN_NAME)
     if user is None:
+        quarantine('bad_recipient', event)
         raise NotFoundError(f'no such user: {username}')
 
     # saves the activity in an S3 bucket
     try:
         save_activity_in_inbox(body, event['digest'], username)
     except ValueError as exc:
+        quarantine('bad_signature', event)
         raise BadRequestError(f'{exc}') from exc

cdk/lib/mumble-api.ts

@@ -138,6 +138,7 @@ export class MumbleApi extends Construct {
           OBJECTS_BUCKET_NAME: objectStore.objectsBucket.bucketName,
           DOMAIN_NAME_PARAMETER_PATH:
             systemParameters.domainNameParameter.parameterName,
+          QUARANTINE_BUCKET_NAME: objectStore.quarantineBucket.bucketName,
         },
         memorySize: 256,
         timeout: Duration.seconds(20),
@@ -148,6 +149,7 @@ export class MumbleApi extends Construct {
     systemParameters.domainNameParameter.grantRead(
       receiveInboundActivityLambda,
     );
+    objectStore.quarantineBucket.grantPut(receiveInboundActivityLambda);
     // - receives an activity or object posted to the outbox of a given user
     const receiveOutboundObjectLambda = new PythonFunction(
       this,

cdk/lib/object-store.ts

@@ -88,6 +88,8 @@ export interface Props {
 export class ObjectStore extends Construct {
   /** S3 bucket for objects. */
   readonly objectsBucket: s3.IBucket;
+  /** S3 bucket for quarantined objects. */
+  readonly quarantineBucket: s3.IBucket;
   /** DynamoDB table to manage metadata and the history of objects. */
   readonly objectTable: dynamodb.Table;
   /**
@@ -151,6 +153,13 @@ export class ObjectStore extends Construct {
       removalPolicy: RemovalPolicy.RETAIN,
     });
 
+    // S3 bucket for quarantined objects
+    this.quarantineBucket = new s3.Bucket(this, 'QuarantineBucket', {
+      blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
+      encryption: s3.BucketEncryption.S3_MANAGED,
+      removalPolicy: RemovalPolicy.RETAIN,
+    });
+
     // DynamoDB table to manage metadata and the history of objects
     const billingSettings = deploymentStage === 'production' ? {
       billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,