Context

Marketing and sales both want access to information about completed contracts. They want this information to be up to date. This tutorial shows how Conduktor can facilitate data encryption and data access within the same pipeline. Sales accesses decrypted data, but marketing can only access data with sensitive information encrypted.

Different teams, different encryption requirements; a solution in real-time using Conduktor Gateway

Your company is thriving and signs contracts constantly. Multiple teams need access to this information. Sales needs real-time access to signed contracts for tracking progress with quarterly targets Marketing also want access, as the team will tailor their campaigns to industries with interest in your products. Both teams want this data to be up to date. However, details such as client contacts and email addresses need to be encrypted for marketing, but not for sales.

Duplicating data for each team can cause headaches and, even worse, data discrepancy. Such discrepancy left unchecked leads to inconsistent insights into company performance and market trends. Encrypting data-in-motion further complicates your data architecture by adding new client side applications.

Through this contract pipeline tutorial we see how Conduktor can seamlessly facilitate data encryption and data access within the same pipeline with only config files and networking, bypassing the need to add more applications. Messages containing sensitive information can be accessed by different teams, and sensitive data will be encrypted for all teams except sales.

In this tutorial we will use a local terminal and the Conduktor Console UI, and we’ll see the versatility of working with Conduktor Console and Gateway in combination.

Set-up

Let’s run through the contract pipeline solution together. The prerequisites for this tutorial are:

• A Github account

• Docker

• 4GB of RAM

• jq

First clone this Github repo which contains the docker.compose.yml file:

git clone https://github.com/conduktor/tutorial-encryption-access.git

Next move into the project directory and build the docker yaml file for the pipeline:

cd tutorial-encryption-access/ && docker compose up --detach

--detach runs the process in the background, not in your terminal window.

This will start Conduktor Console, Conduktor Gateway, PostgreSQL, and a Kafka cluster. It will also set up networking between them.

Downloading these images will take a few minutes. In addition, the kafkabroker service will take up to a minute to start up as it runs health checks.

This interval gives you enough time to join Conduktor’s Slack community!

Once the images are downloaded head to Conduktor Console at http://localhost:8080 and enter your details. There’s no credit card sign up and we’ll get access to features used later on in the tutorial!

Within our Docker container we have spun up the Kafka Cluster underpinning both Conduktor Gateway and Conduktor Console. With the docker container created let’s check that the cluster is up and running with this command:

docker compose ps

Within the response we can see that we are running an apache/kafka container listening on port 9092: 

apache/kafka:latest              9092/tcp   kafkabroker

Virtual Clusters

Conduktor introduces the notion of Virtual clusters (vClusters from hereon). They allow for the creation of conceptually distinct and isolated logical clusters which can rely on just one physical Kafka cluster running locally or in the cloud; they are means of simplifying infrastructure. The default vCluster is called the  ‘passthrough’ cluster.

In this diagram you can see the vClusters we will be creating, the topics we will create within them and the physical cluster topic which they will be mapped to.

In order to connect to any vCluster we first must connect to the default vCluster, and for this we’ll need credentials (a JWT). A POST request to the Gateway will provide us with them. We also set the duration of this JWT to 90 days. Using the tee command the resulting gateway credentials will be printed in the terminal and stored in the gateway-admin.properties file for reference. 

curl --request POST --silent \
  'http://localhost:8888/admin/vclusters/v1/vcluster/passthrough/username/admin' \
    --user 'admin:conduktor' \
      --header "Content-Type: application/json" \
      --data-raw '{"lifeTimeSeconds": 7776000}' |
        jq --raw-output .token | tee gateway-admin.properties

The resulting string starting with ‘ey’ is your password (it’s a JWT, base64-encoded), and is the password of your vCluster. 

In Conduktor Console, edit the cluster cdk-gateway:

Switch to SASL_PLAINTEXT + PLAIN. The password field is the JWT you’ve generated in the previous step. In addition, make sure to add the Gateway bootstrap server: “conduktor-gateway:6969”

Note: if you restart your docker Gateway service you’ll need to re-enter your Gateway token, create your topics and send your contracts again.

Now we’re going to create a vCluster for our sales team and another vCluster for our marketing team, setting our admin privileges in the same way.

curl --request POST 'http://localhost:8888/admin/vclusters/v1/vcluster/sales/username/admin' \--user 'admin:conduktor' \--header 'Content-Type: application/json' \--silent \--data-raw '{"lifeTimeSeconds": 7776000}' | jq -r '.token' | tee sales-admin.properties

Note: /vclusters/v1/vcluster/sales/username/admin

After receiving our credentials for the sales virtual cluster we will create the virtual cluster within Conduktor console. First we go to ‘Manage Clusters’ to add a new cluster.

We will authenticate our sales cluster in the same way as we authenticated Conduktor Gateway, by using SASL_PLAINTEXT, PLAIN, and adding the Gateway bootstrap server. However, remember to add the sales vCluster token in the password field, the token generated by the previous command.

Let’s now do the same with marketing:

curl --request POST 'http://localhost:8888/admin/vclusters/v1/vcluster/marketing/username/admin' \--header 'Content-Type: application/json' \--user 'admin:conduktor' \--silent \--data-raw '{"lifeTimeSeconds": 7776000}' | jq --raw-output ".token" |tee marketing-admin.properties

Following through these steps for both virtual clusters we will now find them within the clusters section of the UI:

Creating Contract Topic and Messages

Now head over to Conduktor Console at http://localhost:8080/console/cdk-gateway/topics to create a topic within the UI!

• Create a topic named ‘signed-contracts’, using the default configuration

• In this topic you can now produce messages:

Paste and send this test data in the “Value” field and hit “Produce”:

{
    "title": "......",
    "date": "......",
    "company": "......",
    "industry": "......",
    "contract_value": "......",
    "contract_duration": "......",
    "contact": "......",
    "contact_email": "......"
}

Back to the consume tab, if you click on the message you can see its contents, header and metadata:

Encryption Interceptor

Let’s now create the interceptor JSON file. Within the UI head to Kafka Gateway and click ‘Create my first interceptor’ 

Find ‘On the fly field level encryption’ within the Data-security category.

Replace the default JSON with our interceptor configuration:

{
    "name": "PII-Encryption",
    "pluginClass": "io.conduktor.gateway.interceptor.EncryptPlugin",
    "priority": 100,
    "config": {
      "topic": "signed_contracts",
      "recordValue": {
        "fields": [
          {
            "fieldName": "contact",
            "keySecretId": "name-secret",
            "algorithm": "AES128_GCM"
          },
          {
            "fieldName": "contact_email",
            "keySecretId": "email-secret",
            "algorithm": "AES128_GCM"
          }
        ]
      }
    }
  }

Here are the config properties we need to understand for this tutorial: the name we are giving to the interceptor, the topic to which the interceptor will be applied and the ‘fields’ dictionary which details, for each chosen message field to be encrypted, an identifier and an algorithm to be applied. For example, the interceptor will look for a ‘contact’ field within the messages and encrypt the contacts name using the AES128_GCM algorithm.

Once we have deployed the interceptor it becomes visible in the UI. Returning to the ‘Topics’ window let’s now send a new contract:

{    "title": "contract",    "date": "01/07/2024",    "company": "Lego",    "industry": "Toy",    "contract_value": "£2,500,000",    "contract_duration": "3 years",    "contact": "Niels B. Christiansen",    "contact_email": "nchrisiansen@lego.com"}

Click on the message within the consume tab of the signed_contract topic and you will see the above contract with the contact details encrypted:

http://localhost:8080/console/cdk-gateway/topics/

All signed contracts entering the data pipeline now have personal details encrypted at-rest. Whoever consumes this data will now see those details encrypted. Our PII_encryption interceptor sits on Conduktor Gateway’s Kafka proxy where it acts on each message produced within the signed_contracts topic. 

Now we will link our virtual clusters to this topic and add a specific interceptor to the sales cluster, allowing the team to re-access (i.e. decrypt) its sensitive information.

Connecting the vClusters to the Topic

To give marketing access to the signed_contracts topic run this POST request:

curl --request POST 'http://localhost:8888/admin/vclusters/v1/vcluster/marketing/topics/encrypted_signed_contracts' \
--user 'admin:conduktor' \
--header "Content-Type: application/json" \
--silent \
--data-raw '{ "physicalTopicName": "signed_contracts", "readOnly": true, "type": "alias" }' | jq

Note: we can rename the topic with an alias, like here `encrypted_signed_contracts` to hide the underlying physical topic name to the marketing team. We create this new name at the end of the URL: /marketing/topics/encrypted_signed_contracts

You can now see the topic in your marketing vCluster. Whenever going between clusters it is important to reopen the topics tab. Because we are giving the topics different alias’ we have to look at the unique topics for each vCluster.

Now let’s create the topic mapping in the same way for the sales cluster:

curl --request POST 'http://localhost:8888/admin/vclusters/v1/vcluster/sales/topics/decrypted_signed_contracts' \
--user 'admin:conduktor' \
--header "Content-Type: application/json" \
--silent \
--data-raw '{ "physicalTopicName": "signed_contracts", "readOnly": true, "type": "alias" }' | jq

Note: vcluster/sales/topics/decrypted_signed_contracts

Decryption Interceptor

If you consume data in the sales vCluster, you can see the data of the topic is still encrypted. It’s time to set this vCluster to decrypt the data so that sales can see the contact and contact email for a given signed contract.

In order to do so we need to connect the sales vCluster on Conduktor Console to the sales vCluster on Conduktor Gateway. Let’s manage the sales vCluster once more, heading to the ‘Provider’ tab:

This tab is going to run a curl command behind the scenes just as we ran earlier on. Therefore we input the same URL and user credentials: ‘http://conduktor-gateway:8888’/  ‘admin:conduktor’, but we specify the sales vCluster upon which our decryption interceptor will sit.

Heading back to ‘Kafka Gateway’ we will now add our decryption interceptor.

Notice that we already see an interceptor, this interceptor maps the decrypted_signed_contracts to the signed_contracts topic and was created with the previous POST command.

Now let’s add the decryption interceptor, choosing the ‘field level decryption’ plugin. Populate the interceptor with this JSON:

{
    "name": "PII-Decryption",
    "pluginClass": "io.conduktor.gateway.interceptor.DecryptPlugin",
    "priority": 100,
    "config": {
      "topic": "decrypted_signed_contracts",
      "recordValueFields": [
        "contact",
        "contact_email"
      ]
    }
}

Just like the encryption interceptor before, this decryption interceptor targets the specific ‘contact’ and ‘contact_email’ fields.

We can now see our interceptor within the sales vCluster, ready to decrypt the messages consumed through the sales vCluster.

Completed Pipeline

If we produce a second contract within the Gateway cluster the first interceptor will encrypt the sensitive fields as normal, but when we consume the messages within the sales cluster the second interceptor decrypts the sensitive fields for us to see.

Second contract to send:

{
    "title": "contract",    
    "date": "01/06/2024",
    "company": "Disney",
    "industry": "Entertainment",
    "contract_value": "£10,000,000",
    "contract_duration": "24 months",
    "contact": "Walt Disney",
    "contact_email": "wdisney@disney.com"
}

Now you should see that, for the sales team, the contact information is unencrypted:

There you have it! Contracts which enter your new data pipeline are now encrypted as they enter. Marketing can access these contracts, but sensitive information will be encrypted. Sales may also access these contracts with all sensitive information decrypted.

You can find extra contracts: ‘bonus_contracts’ within the tutorial repo. Produce those contracts to the signed_contracts topic if you’d like to see further examples. To take the tutorial a step further you have a couple of options. First, you could add an engineering team to Conduktor Console and subscribe that team to the signed_contracts topic. Perhaps for that team you’d want all of the data encrypted, and so, second, you could try out the Encryption interceptor and see the range of KMS which are supported. If you get stuck at all you can reach us on Conduktor’s Slack space.