Add an Ockam Kafka Input and an Ockam Kafka Output #2616

mrinalwadhwa · 2024-05-29T02:12:32Z

Everyone on the Ockam team is thrilled to contribute two new Ockam Kafka plugins - an input and an output.

Ockam Kafka plugins for Redpanda Connect create Ockam Portals to send end-to-end encrypted messages through Kafka. These plugins combine Ockam and the Franz Kafka plugins to setup encrypted portals.

Messages from a producer are encrypted to a specific Consumer. Only that specific Consumer can decrypt these messages. This guarantees that your data cannot be observed or tampered as it passes through Kafka. Operators of the Kafka cluster only see end-to-end encrypted data. Any compromise of an operator's infrastructure cannot compromise your business data. Kafka brokers cannot see the data.

Examples

On one machine

Let's begin with a very simple local example that runs all the parts on one machine.

Consumer

The consumer uses ockam_kafka input to decrypt and read messages from Kafka. It then transforms messages using a processor and writes the transformed message to stdout.

# consumer.yaml

input:
  ockam_kafka:
    seed_brokers: [localhost:9092]
    topics: [topic_A]
    consumer_group: example_group

pipeline:
  processors:
    - bloblang: |
        root.doc = this | content().string()
        root.length = content().length()
        root.topic = meta("kafka_topic")

output:
  stdout: {}

Producer

The producer reads messages from stdin. It then uses ockam_kafka output to encrypt and write messages to Kafka.

# producer.yaml

input:
  stdin: {}

output:
  ockam_kafka:
    seed_brokers: [localhost:9092]
    topic: topic_A

Run

With a redpanda-connect binary compiled using this pull request:

# Setup Redpanda
brew install redpanda-data/tap/redpanda
rpk container start

# Start a Consumer
redpanda-connect -c ./consumer.yaml

# Start Producers in separate terminals and type some messages.
redpanda-connect -c ./producer.yaml --set "http.address=127.0.0.1:4196"
redpanda-connect -c ./producer.yaml --set "http.address=127.0.0.1:4197"
redpanda-connect -c ./producer.yaml --set "http.address=127.0.0.1:4198"

# See encrypted messages in Redpanda at http:https://localhost:8080

Between two machines

Consumer

# consumer.yaml

input:
  ockam_kafka:
    seed_brokers: [${IP_OF_REDPANDA_MACHINE}:9092]
    topics: [topic_A]
    consumer_group: example_group
    ockam_node_address: 0.0.0.0:6262
    ockam_allow_producer: ${OCKAM_ALLOW_PRODUCER}

pipeline:
  processors:
    - bloblang: |
        root.doc = this | content().string()
        root.length = content().length()
        root.topic = meta("kafka_topic")

output:
  stdout: {}

Producer

# producer.yaml

input:
  stdin: {}

output:
  ockam_kafka:
    seed_brokers: [${IP_OF_REDPANDA_MACHINE}:9092]
    topic: topic_A
    ockam_route_to_consumer: /ip4/${IP_OF_CONSUMER_MACHINE}/tcp/6262
    ockam_allow_consumer: ${OCKAM_ALLOW_CONSUMER}

Run

With a redpanda-connect binary compiled using this pull request:

# Redpanda
brew install redpanda-data/tap/redpanda
rpk container start

# Machine 1 - Producer
curl --proto '=https' --tlsv1.2 -sSfL https://install.command.ockam.io | bash && source "$HOME/.ockam/env"
ockam identity create
# copy the generated Identity's identifier to Machine 2 - I5fc86d2...

# Machine 2 - Consumer
curl --proto '=https' --tlsv1.2 -sSfL https://install.command.ockam.io | bash && source "$HOME/.ockam/env"
ockam identity create
IP_OF_REDPANDA_MACHINE=... OCKAM_ALLOW_PRODUCER=I5fc86d2... redpanda-connect -c ./consumer.yaml
# copy the generated Identity's identifier to Machine 1 - I94d64b91...

# Machine 1 - Producer
IP_OF_REDPANDA_MACHINE=... IP_OF_CONSUMER_MACHINE=... \
  OCKAM_ALLOW_CONSUMER=I94d64b91... redpanda-connect -c ./producer.yaml

Across several machines

Consumer

# consumer.yaml

input:
  ockam_kafka:
    seed_brokers: [rp-node-0:9092]
    topics: [topic_A]
    consumer_group: example_group
    ockam_allow_producer: producer
    ockam_relay: consumer_relay
    ockam_enrollment_ticket: ${OCKAM_ENROLLMENT_TICKET}

pipeline:
  processors:
    - bloblang: |
        root = this
        root.data.message = this.data.message.uppercase()

output:
  stdout: {}

Producer

# producer.yaml

input:
  generate:
    count: 1000
    interval: "@every 1s"
    mapping: |
      root = {
        "_producer": hostname(),
        "data": { "email": fake("email"), "message": fake("sentence") }
      }

output:
  ockam_kafka:
    seed_brokers: [rp-node-0:9092]
    topic: topic_A
    ockam_route_to_consumer: /project/default/service/forward_to_consumer_relay/secure/api
    ockam_allow_consumer: consumer
    ockam_enrollment_ticket: ${OCKAM_ENROLLMENT_TICKET}

Run

As a temporary step, until this pull request is merged, we compiled this PR's branch and pushed a docker image to ghcr.io/build-trust/redpanda-connect

# Setup Redpanda
brew install redpanda-data/tap/redpanda
rpk container start

# Setup Ockam
brew install build-trust/ockam/ockam
ockam enroll

# Setup Consumer
ockam project ticket --usage-count 1 --expires-in 10m --attribute consumer --relay consumer_relay > ticket
docker run --rm --network redpanda --name consumer -v "$(pwd)/consumer.yaml:/connect.yaml" \
  -e OCKAM_ENROLLMENT_TICKET="$(cat ticket)" ghcr.io/build-trust/redpanda-connect

# Setup Producers (in a new terminal window)
for i in $(seq 1 3); do
    ockam project ticket --usage-count 1 --expires-in 10m --attribute producer > ticket
    docker run --rm -d --network redpanda --name "producer$i" -v "$(pwd)/producer.yaml:/connect.yaml" \
        -e OCKAM_ENROLLMENT_TICKET="$(cat ./ticket)" ghcr.io/build-trust/redpanda-connect
done

# It can take a minute or so for the messages to start flowing.
# Observe that the above consumer can decrypt and transform messages.

# However, if you look at the messages inside redpanda console at localhost:8080
# or using the below rpk command, you'll notice that the messages are encrypted.
rpk topic consume topic_A

# Cleanup
docker stop $(docker ps -q --filter "name=producer" --filter "name=consumer")
rpk container purge

Across several machines - using Redpanda Serverless

Consumer

#consumer.yaml

input:
  ockam_kafka:
    seed_brokers: ["${REDPANDA_BOOTSTRAP_SERVER}"]
    topics: [topic_A]
    consumer_group: example_group
    tls:
      enabled: true
    sasl:
      - mechanism: "SCRAM-SHA-256"
        username: "${REDPANDA_USERNAME}"
        password: "${REDPANDA_PASSWORD}"
    ockam_allow_producer: producer
    ockam_relay: consumer_relay
    ockam_enrollment_ticket: "${OCKAM_ENROLLMENT_TICKET}"

pipeline:
  processors:
    - bloblang: |
        root = this
        root.data.message = this.data.message.uppercase()

output:
  stdout: {}

Producers

# producer.yaml

input:
  generate:
    count: 1000
    interval: "@every 1s"
    mapping: |
      root = {
        "_producer": hostname(),
        "data": { "email": fake("email"), "message": fake("sentence") }
      }

output:
  ockam_kafka:
    seed_brokers: ["${REDPANDA_BOOTSTRAP_SERVER}"]
    topic: topic_A
    sasl:
      - mechanism: "SCRAM-SHA-256"
        username: "${REDPANDA_USERNAME}"
        password: "${REDPANDA_PASSWORD}"
    tls:
      enabled: true
    ockam_route_to_consumer: /project/default/service/forward_to_consumer_relay/secure/api
    ockam_allow_consumer: consumer
    ockam_enrollment_ticket: ${OCKAM_ENROLLMENT_TICKET}

Run

# Setup Redpanda
#   1. Create a Redpanda serverless cluster - https://cloud.redpanda.com/
#   2. Create a new topic with name topic_A with default settings.
#   3. Create a new user with name tester.
#       - Select SCRAM-SHA-256 as SASL Mechanism and copy the password to use as environment variable below.
#       - Navigate to Security->ACLs, select tester, select "Allow all operations"
#   4. Copy the following values to your shell's and set them as environment variables:
export REDPANDA_BOOTSTRAP_SERVER="TODO:9092"
export REDPANDA_USERNAME="tester"
export REDPANDA_PASSWORD="TODO"

# Setup Ockam
#   1. Signup for Ockam https://ockam.io/signup
#   2. Setup Ockam Command on your development machine
brew install build-trust/ockam/ockam
ockam enroll

# Setup Consumer
ockam project ticket --usage-count 1 --expires-in 10m --attribute consumer --relay consumer_relay > ticket
docker run --rm --name consumer \
  -v "$(pwd)/consumer.yaml:/connect.yaml" \
  -e OCKAM_ENROLLMENT_TICKET="$(cat ticket)" \
  -e REDPANDA_BOOTSTRAP_SERVER="$REDPANDA_BOOTSTRAP_SERVER" \
  -e REDPANDA_USERNAME="$REDPANDA_USERNAME" \
  -e REDPANDA_PASSWORD="$REDPANDA_PASSWORD" \
  ghcr.io/build-trust/redpanda-connect

# Setup Producers (in a new terminal window)
for i in $(seq 1 3); do
    ockam project ticket --usage-count 1 --expires-in 10m --attribute producer > ticket
    docker run --rm -d --name "producer$i" \
        -v "$(pwd)/producer.yaml:/connect.yaml" \
        -e OCKAM_ENROLLMENT_TICKET="$(cat ./ticket)" \
        -e REDPANDA_BOOTSTRAP_SERVER="$REDPANDA_BOOTSTRAP_SERVER" \
        -e REDPANDA_USERNAME="$REDPANDA_USERNAME" \
        -e REDPANDA_PASSWORD="$REDPANDA_PASSWORD" \
        ghcr.io/build-trust/redpanda-connect
done

# It can take a minute or so for the messages to start flowing.
# Observe that the above consumer can decrypt and transform messages.

# However, if you look at the messages inside redpanda console or using
# the below rpk command, you'll notice that the messages are encrypted.
rpk cloud login
rpk topic consume topic_A

# Cleanup
docker stop $(docker ps -q --filter "name=producer" --filter "name=consumer")

mrinalwadhwa · 2024-05-29T07:02:45Z

internal/impl/ockam/input_kafka.go

+func ockamKafkaInputConfig() *service.ConfigSpec {
+ return kafka.FranzKafkaInputConfig().
+ Summary(`Ockam`).
+ Field(service.NewStringField("ockam_identity_name").Optional()).
+ Field(service.NewStringField("ockam_node_address").Default("127.0.0.1:6262")).
+ Field(service.NewStringField("ockam_allow_producer").Default("self")).
+ Field(service.NewStringField("ockam_enrollment_ticket").Optional()).
+ Field(service.NewStringField("ockam_relay").Optional())
+}


I'll push some documentation for these tomorrow with some examples on how to use them.

mrinalwadhwa · 2024-05-29T07:03:09Z

internal/impl/ockam/output_kafka.go

+ Field(service.NewStringField("ockam_enrollment_ticket").Optional()).
+ Field(service.NewStringField("ockam_identity_name").Optional()).
+ Field(service.NewStringField("ockam_allow_consumer").Default("self")).
+ Field(service.NewStringField("ockam_route_to_consumer").Default("/ip4/127.0.0.1/tcp/6262"))
+}


I'll push some documentation for these tomorrow with some examples on how to use them.

CLAassistant · 2024-05-30T13:57:22Z

All committers have signed the CLA.

Jeffail

Hey, left a couple of comments. Is this being distributed as an Apache V2 licensed plugin? If so we need headers added to each file. There's a handy script for it add_license_headers.sh, if you run that from the repository root directory it should add them automatically.

Jeffail · 2024-07-25T13:31:14Z

internal/impl/ockam/input_kafka.go

+ Example([]string{"foo:9092", "bar:9092"}).
+ Example([]string{"foo:9092,bar:9092"})).
+ Field(service.NewBoolField("disable_content_encryption").Default(false).Optional()).
+ Field(service.NewStringField("ockam_identity_name").Optional()).


Can we remove the ockam_ prefix of these fields? If there's a need to distinguish between ockam and kafka specific fields I'd say it's more reasonable to namespace the kafka fields under than object like this:

ockam_kafka: node_address: TODO allow_producer: self kafka: seed_brokers: ...

You can do it with service.NewObjectField.

Jeffail · 2024-07-25T13:31:55Z

internal/impl/ockam/input_kafka.go

+ Example([]string{"localhost:9092"}).
+ Example([]string{"foo:9092", "bar:9092"}).
+ Example([]string{"foo:9092,bar:9092"})).
+ Field(service.NewBoolField("disable_content_encryption").Default(false).Optional()).


Using both Default and Optional is redundant, as a field with a default is always specified. Remove the Optional call if the default value is valid. This goes for all fields.

Jeffail · 2024-07-25T13:32:58Z

internal/impl/ockam/input_kafka.go

+
+ if routeToKafkaOutlet == "self" {
+ // TODO: Handle other tls fields in kafka franz
+ _, tls, err := conf.FieldTLSToggled("tls")


It looks as though the tls field is never defined above.

This is now explicitly defined within the "kafka" namespace

Jeffail · 2024-07-25T13:48:28Z

internal/impl/ockam/input_kafka.go

+ if err != nil {
+ return nil, err
+ }
+ if len(seedBrokers) > 1 {


Suggested change

if len(seedBrokers) > 1 {

if len(seedBrokers) != 1 {

Looks as though exactly one broker is needed here.

Co-authored-by: Mrinal Wadhwa <[email protected]> Co-authored-by: Adrian Benavides <[email protected]> Co-authored-by: Pablo Polvorin <[email protected]> Co-authored-by: etorreborre <[email protected]> Signed-off-by: Mrinal Wadhwa <[email protected]>

davide-baldo · 2024-07-26T17:47:51Z

I rebased and applied your feedback.
I had to modify franz kafka a little bit to better expose the fields.

Jeffail · 2024-08-01T11:09:46Z

Merged via #2755

mrinalwadhwa requested a review from Jeffail as a code owner May 29, 2024 02:12

mrinalwadhwa marked this pull request as draft May 29, 2024 02:12

mrinalwadhwa commented May 29, 2024

View reviewed changes

mrinalwadhwa force-pushed the mrinal/ockam_plugins branch 3 times, most recently from f5fabc0 to 969e043 Compare May 30, 2024 14:28

mrinalwadhwa marked this pull request as ready for review May 30, 2024 15:11

mrinalwadhwa force-pushed the mrinal/ockam_plugins branch from 81cb85b to ee41dd8 Compare May 30, 2024 20:07

mrinalwadhwa force-pushed the mrinal/ockam_plugins branch from ee41dd8 to b26b136 Compare June 13, 2024 04:53

Jeffail requested a review from asimms41 July 23, 2024 07:23

Jeffail reviewed Jul 25, 2024

View reviewed changes

mrinalwadhwa and others added 4 commits July 26, 2024 13:10

Add disable_content_encryption parameter to ockam kafka

6424894

Better parameter organization for ockam kafka

a99123e

davide-baldo force-pushed the mrinal/ockam_plugins branch from a9eda6f to a99123e Compare July 26, 2024 17:45

Jeffail mentioned this pull request Aug 1, 2024

Add an Ockam Kafka Input and an Ockam Kafka Output (Lints & Docs fixes) #2755

Merged

Jeffail closed this Aug 1, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add an Ockam Kafka Input and an Ockam Kafka Output #2616

Add an Ockam Kafka Input and an Ockam Kafka Output #2616

mrinalwadhwa commented May 29, 2024 •

edited

Loading

mrinalwadhwa May 29, 2024

mrinalwadhwa May 29, 2024

CLAassistant commented May 30, 2024 •

edited

Loading

Jeffail left a comment

Jeffail Jul 25, 2024

Jeffail Jul 25, 2024

Jeffail Jul 25, 2024

davide-baldo Jul 26, 2024

Jeffail Jul 25, 2024

davide-baldo commented Jul 26, 2024

Jeffail commented Aug 1, 2024

Add an Ockam Kafka Input and an Ockam Kafka Output #2616

Add an Ockam Kafka Input and an Ockam Kafka Output #2616

Conversation

mrinalwadhwa commented May 29, 2024 • edited Loading

Examples

On one machine

Consumer

Producer

Run

Between two machines

Consumer

Producer

Run

Across several machines

Consumer

Producer

Run

Across several machines - using Redpanda Serverless

Consumer

Producers

Run

mrinalwadhwa May 29, 2024

Choose a reason for hiding this comment

mrinalwadhwa May 29, 2024

Choose a reason for hiding this comment

CLAassistant commented May 30, 2024 • edited Loading

Jeffail left a comment

Choose a reason for hiding this comment

Jeffail Jul 25, 2024

Choose a reason for hiding this comment

Jeffail Jul 25, 2024

Choose a reason for hiding this comment

Jeffail Jul 25, 2024

Choose a reason for hiding this comment

davide-baldo Jul 26, 2024

Choose a reason for hiding this comment

Jeffail Jul 25, 2024

Choose a reason for hiding this comment

davide-baldo commented Jul 26, 2024

Jeffail commented Aug 1, 2024

mrinalwadhwa commented May 29, 2024 •

edited

Loading

CLAassistant commented May 30, 2024 •

edited

Loading