Redacting Text in Amazon Kinesis Data Firehose

Amazon Kinesis Firehose is a managed streaming service designed to take large amounts of data from one place to another. For example, you can take data from sources such as Amazon CloudWatch, AWS IoT, and custom applications using the AWS SDK to destinations Amazon S3, Amazon Redshift, Amazon Elasticsearch, and other services. In this post we will use Amazon S3 as the firehose’s destination.

In some cases you may need to manipulate the data as it goes through the firehose to remove sensitive information. In this guide we will show how Amazon Kinesis Firehose and AWS Lambda can be used in conjunction with Philter to remove sensitive information (PII and PHI) from the text as it travels through the firehose.

Philter is software that redacts PII, PHI, and other sensitive information from text. Philter runs entirely within your private cloud and does not require any external connectivity. Your data never leaves your private cloud and is not sent to any third-party. In fact, you can run Philter without any external network connectivity and we recommend doing so.

Prerequisites

You must have a running instance of Philter 4.0. If you don’t already have one, launch it from the AWS Marketplace with the AWS quick start, which also creates an API key for your application, or from a template as in deploying Philter with CloudFormation.

It’s not required that the instance of Philter be running in AWS but it is required that the instance of Philter be accessible from your AWS Lambda function. Running Philter and your AWS Lambda function in your own VPC allows your Lambda function to communicate locally with Philter from the function. This keeps your sensitive information from being sent over the public internet and keeps the network traffic inside your VPC.

Setting up the Amazon Kinesis Firehose transformation

There is no need to duplicate an excellent blog post on creating an Amazon Kinesis Firehose Data Transformation with AWS Lambda. Instead, refer to the AWS post on Firehose data transformation and substitute the Python code below for the code in that post.

Configuring the firehose and the Lambda function

To start, create an AWS Firehose and configure an AWS Lambda transformation. When creating the AWS Lambda function, select Python 3.x and use the following code. It uses only the Python standard library, so it needs no layer:

import base64
import os
import ssl
import urllib.request

PHILTER_URL = os.environ["PHILTER_URL"]  # for example https://philter.internal:8080/api/filter
PHILTER_API_KEY = os.environ["PHILTER_API_KEY"]

# Verifies Philter's certificate against the system trust store, or against PHILTER_CA_FILE
# when Philter's certificate comes from a private certificate authority.
SSL_CONTEXT = ssl.create_default_context(cafile=os.environ.get("PHILTER_CA_FILE"))


def redact(text: bytes) -> bytes:
    request = urllib.request.Request(
        PHILTER_URL,
        data=text,
        method="POST",
        headers={
            "Authorization": f"Bearer {PHILTER_API_KEY}",
            "Content-Type": "text/plain",
        },
    )
    with urllib.request.urlopen(request, context=SSL_CONTEXT, timeout=20) as response:
        return response.read()


def handler(event, context):
    output = []

    for record in event["records"]:
        filtered = redact(base64.b64decode(record["data"]))

        output.append({
            "recordId": record["recordId"],
            "result": "Ok",
            "data": base64.b64encode(filtered + b"\n").decode("utf-8"),
        })

    return {"records": output}

Set three environment variables on the function:

• PHILTER_URL: Philter’s filter endpoint, such as https://philter.internal:8080/api/filter.
• PHILTER_API_KEY: an API key for the function, ideally one holding only the redact scope.
• PHILTER_CA_FILE (optional): a certificate file to trust, if Philter’s certificate comes from a private certificate authority. Philter’s default self-signed certificate is not valid for its host name, so give Philter a certificate issued for the name the function uses, as described in using a signed SSL certificate with Philter. Do not turn off certificate verification instead.

The following Kinesis Firehose test event can be used to test the function:

{
  "invocationId": "invocationIdExample",
  "deliveryStreamArn": "arn:aws:kinesis:EXAMPLE",
  "region": "us-east-1",
  "records": [
    {
      "recordId": "49546986683135544286507457936321625675700192471156785154",
      "approximateArrivalTimestamp": 1495072949453,
      "data": "R2VvcmdlIFdhc2hpbmd0b24gbGl2ZWQgaW4gOTAyMTAgYW5kIGhpcyBTU04gd2FzIDEyMy00NS02Nzg5Lg=="
    },
    {
      "recordId": "49546986683135544286507457936321625675700192471156785155",
      "approximateArrivalTimestamp": 1495072949453,
      "data": "R2VvcmdlIFdhc2hpbmd0b24gbGl2ZWQgaW4gOTAyMTAgYW5kIGhpcyBTU04gd2FzIDEyMy00NS02Nzg5Lg=="
    }
  ]
}

This test event contains 2 records, and the data of each is the base64 encoding of “George Washington lived in 90210 and his SSN was 123-45-6789.” When the test is executed the response will be:

{
  "records": [
    {
      "recordId": "49546986683135544286507457936321625675700192471156785154",
      "result": "Ok",
      "data": "e3t7UkVEQUNURUQtbmFtZX19fSBsaXZlZCBpbiA5MDIxMCBhbmQgaGlzIFNTTiB3YXMge3t7UkVEQUNURUQtc3NufX19Lgo="
    },
    {
      "recordId": "49546986683135544286507457936321625675700192471156785155",
      "result": "Ok",
      "data": "e3t7UkVEQUNURUQtbmFtZX19fSBsaXZlZCBpbiA5MDIxMCBhbmQgaGlzIFNTTiB3YXMge3t7UkVEQUNURUQtc3NufX19Lgo="
    }
  ]
}

Each record’s data decodes to:

{{{REDACTED-name}}} lived in 90210 and his SSN was {{{REDACTED-ssn}}}.

When running the test, the AWS Lambda function extracts the data from each record in the firehose, submits it to Philter for filtering, and returns the redacted records to Firehose. The request names no policy, so Philter applies its default policy, which redacts names, SSNs, and email addresses; add a p query parameter to PHILTER_URL to apply another. If a call to Philter fails, the function raises an error rather than passing the record through unredacted.

When data is now published to the Amazon Kinesis Data Firehose stream, the data will be processed by the AWS Lambda function and Philter prior to exiting the firehose at its configured destination.

Processing data

We can use the AWS CLI to publish data to our Amazon Kinesis Firehose stream called sensitive-text:

aws firehose put-record \
    --delivery-stream-name sensitive-text \
    --cli-binary-format raw-in-base64-out \
    --record '{"Data": "George Washington lived in 90210 and his SSN was 123-45-6789."}'

Check the destination S3 bucket and you will have a single object with the following line:

{{{REDACTED-name}}} lived in 90210 and his SSN was {{{REDACTED-ssn}}}.

Conclusion

In this guide we have created an Amazon Kinesis Data Firehose pipeline that uses an AWS Lambda function to remove PII and PHI from the text in the streaming pipeline.

Philter is available from the AWS Marketplace. Not using AWS? Philter is also available from the Google Cloud Marketplace and the Microsoft Azure Marketplace.

Resources

• Amazon Kinesis Data Firehose
• Amazon Kinesis Data Firehose Data Transformation