Back to Guides

    Security Guide

    Authentication, authorization, encryption, multi-tenant isolation, and agent security.

    Table of Contents

    1. 1.Overview
    2. 2.Authentication (who are you?)
    3. 3.Authorization (what can you do?)
    4. 4.Encryption in Transit (TLS/mTLS)
    5. 5.Encryption at Rest
    6. 6.Multi-Tenant Isolation
    7. 7.Agent Security
    8. 8.Production Configuration

    1. Overview

    ┌─────────────────────────────────────────────────────────────────┐
│                      SECURITY LAYERS                            │
│                                                                 │
│  ┌─────────────┐  ┌───────────────┐  ┌──────────────────────┐  │
│  │ TRANSIT     │  │ AUTHN         │  │ AUTHZ                │  │
│  │ TLS/mTLS    │  │ SASL PLAIN    │  │ ACL per topic        │  │
│  │ Cert pinning│  │ JWT tokens    │  │ RBAC per agent/tool  │  │
│  │             │  │ API keys      │  │ Namespace isolation  │  │
│  └─────────────┘  └───────────────┘  └──────────────────────┘  │
│                                                                 │
│  ┌─────────────┐  ┌───────────────┐  ┌──────────────────────┐  │
│  │ AT REST     │  │ MULTI-TENANT  │  │ AGENTS               │  │
│  │ AES-256-GCM │  │ Namespace     │  │ RBAC tools           │  │
│  │ Master key  │  │ Quotas/plans  │  │ ReasoningAuthorizer  │  │
│  │ Auto-wipe   │  │ FREE/PRO/ENT  │  │ Sandboxing           │  │
│  └─────────────┘  └───────────────┘  └──────────────────────┘  │
└─────────────────────────────────────────────────────────────────┘

    2. Authentication (who are you?)

    Kafka Wire Protocol (SASL PLAIN)

    Producers/consumers authenticate via SASL PLAIN (standard Kafka):

    Properties
    1# streamflow.properties, server side
2kafka.protocol.sasl.enabled=true
3kafka.protocol.sasl.credentials=admin:secret123,app-user:app-pass,readonly:read-only-pass
    Java
    1// Client side, SDK
2Properties props = new Properties();
3props.put("security.protocol", "SASL_PLAINTEXT");
4props.put("sasl.mechanism", "PLAIN");
5props.put("sasl.jaas.config",
6    "org.apache.kafka.common.security.plain.PlainLoginModule required " +
7    "username=\"app-user\" password=\"app-pass\";");
8
9StreamFlowProducer producer = client.producer(props);
    python
    1# Python (confluent-kafka)
2producer = Producer({
3    'bootstrap.servers': 'streamflow:9092',
4    'security.protocol': 'SASL_PLAINTEXT',
5    'sasl.mechanism': 'PLAIN',
6    'sasl.username': 'app-user',
7    'sasl.password': 'app-pass'
8})

    REST API (JWT + API Keys)

    bash
    1# Get a JWT token
2curl -X POST http://streamflow:8080/api/auth/login \
3  -H "Content-Type: application/json" \
4  -d '{"email": "admin@example.com", "password": "secret"}'
5# Response: {"token": "eyJhbGciOiJIUzI1NiJ9..."}
6
7# Use the token
8curl http://streamflow:8080/api/v1/agents \
9  -H "Authorization: Bearer eyJhbGciOiJIUzI1NiJ9..."
10
11# Or via API Key (for services)
12curl http://streamflow:8080/api/v1/agents \
13  -H "Authorization: Bearer sk-asst-xxxxxxxxxxxxx"

    JWT Format

    JSON
    1{
2  "sub": "user-123",
3  "email": "admin@example.com",
4  "name": "Admin User",
5  "tenant": "acme-corp",
6  "role": "ADMIN",
7  "iat": 1711929600,
8  "exp": 1712016000
9}

    gRPC (mTLS or Bearer Token)

    Identity is resolved in this order:

    • mTLS client certificate → CN (Common Name) extracted
    • Authorization: Bearer <token> header → JWT validated
    Properties
    1# streamflow.properties
2grpc.tls.enabled=true
3grpc.tls.mutual=true
4grpc.tls.cert.path=/opt/streamflow/certs/server.crt
5grpc.tls.key.path=/opt/streamflow/certs/server.key
6grpc.tls.trust.cert.path=/opt/streamflow/certs/ca.crt

    3. Authorization (what can you do?)

    Topic ACL (deny-by-default)

    Every operation on a topic is evaluated against ACL rules. By default: everything is denied.

    Properties
    1# streamflow.properties
2topic.acl.enabled=true
    Java
    1// Define ACL rules
2AclRule allowProducer = AclRule.builder()
3    .subject("client:order-service")
4    .resource(AclResource.TOPICS)
5    .resourcePattern("orders")         // exact topic
6    .verb(AclVerb.PUBLISH)
7    .effect(AclEffect.ALLOW)
8    .priority(100)
9    .build();
10
11AclRule allowConsumer = AclRule.builder()
12    .subject("role:analytics")
13    .resource(AclResource.TOPICS)
14    .resourcePattern("orders.*")       // wildcard
15    .verb(AclVerb.SUBSCRIBE)
16    .effect(AclEffect.ALLOW)
17    .priority(100)
18    .build();
19
20AclRule denyPII = AclRule.builder()
21    .subject("role:external")
22    .resource(AclResource.TOPICS)
23    .resourcePattern("users-pii")
24    .verb(AclVerb.SUBSCRIBE)
25    .effect(AclEffect.DENY)            // DENY > ALLOW (same priority)
26    .priority(200)
27    .build();

    Resolution Order

    1. Collect all matching rules (subject + resource + verb)
2. Sort by priority (highest first)
3. First DENY → denied
4. First ALLOW → allowed
5. No match → denied (deny-by-default)

    RBAC for Agents and MCP Tools

    Agents have limited permissions on which tools they can call:

    Properties
    1# streamflow.properties
2mcp.rbac.enabled=true
    Java
    1// Grant agent "fraud-detector" access to 2 tools
2rbacManager.grantPermission("fraud-detector",
3    new Permission("tool:database-query",
4        Set.of("execute"), Duration.ofHours(24)));
5
6rbacManager.grantPermission("fraud-detector",
7    new Permission("tool:send-email",
8        Set.of("execute"), Duration.ofHours(24)));
9
10// Agent "intern-bot" does NOT have access to database-query
11// → If it tries, ReasoningAuthorizationException is thrown

    Role Hierarchy

    ADMIN → can do everything (tools, topics, agents)
  └── OPERATOR → manage agents, read all topics
       └── DEVELOPER → create agents, publish/subscribe on own topics
            └── VIEWER → read only

    4. Encryption in Transit (TLS/mTLS)

    TLS for Kafka Wire Protocol

    Properties
    1# streamflow.properties
2kafka.protocol.tls.enabled=true
3kafka.protocol.tls.keystore.path=/opt/streamflow/certs/kafka.jks
4kafka.protocol.tls.keystore.password=${KEYSTORE_PASSWORD}
    Java
    1// Client SDK with TLS
2Properties props = new Properties();
3props.put("security.protocol", "SSL");
4props.put("ssl.truststore.location", "/path/to/truststore.jks");
5props.put("ssl.truststore.password", "truststore-pass");
6
7StreamFlowClient client = StreamFlowClient.remote("streamflow:9092");
8StreamFlowProducer producer = client.producer(props);

    TLS for gRPC (inter-node)

    Properties
    1# streamflow.properties
2grpc.tls.enabled=true
3grpc.tls.cert.path=/opt/streamflow/certs/server.crt
4grpc.tls.key.path=/opt/streamflow/certs/server.key
5
6# mTLS (mutual), nodes authenticate each other
7grpc.tls.mutual=true
8grpc.tls.trust.cert.path=/opt/streamflow/certs/ca.crt

    Certificate Pinning (MITM protection)

    Properties
    1# streamflow.properties, pin the cluster certificate
2grpc.tls.pinned.certificates=a1b2c3d4e5f6...  # SHA-256 of SPKI

    The pin is verified at TLS handshake, zero cost per message.

    5. Encryption at Rest

    Raft Snapshots

    Cluster state snapshots are encrypted automatically:

    bash
    1# Environment variable for the master key (32 bytes, Base64)
2export PULSE_MASTER_KEY=$(openssl rand -base64 32)
    • Algorithm: AES-256-GCM (authenticated encryption)
    • Random IV: 12 bytes per operation
    • Auth tag: integrated (protects against tampering)
    • Auto-wipe: plaintext automatically cleared from memory after 100ms

    Segment Data (Event Log)

    Properties
    1# streamflow.properties
2log.encryption.enabled=true
3log.encryption.key=${LOG_ENCRYPTION_KEY}

    6. Multi-Tenant Isolation

    Tenant Model

    ┌─────────────────────────────────────────────────────┐
│ FREE (Shared Cluster)                                │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐             │
│ │ ns:acme  │ │ ns:beta  │ │ ns:gamma │  isolated    │
│ │ quota:1K │ │ quota:1K │ │ quota:1K │  namespaces  │
│ └──────────┘ └──────────┘ └──────────┘             │
├─────────────────────────────────────────────────────┤
│ PRO (Shared + Dedicated)                             │
│ ┌──────────────────────┐                             │
│ │ ns:enterprise-x      │  higher quotas              │
│ │ quota:100K, 50 agents│  dedicated resources        │
│ └──────────────────────┘                             │
├─────────────────────────────────────────────────────┤
│ ENTERPRISE (Dedicated Cluster)                       │
│ ┌──────────────────────────────────────┐             │
│ │ Dedicated cluster, SLA 99.99%       │             │
│ │ Custom quotas, compliance features  │             │
│ └──────────────────────────────────────┘             │
└─────────────────────────────────────────────────────┘

    Namespace Isolation

    • Each tenant has its own namespace
    • Topics are prefixed: <namespace>.<topic>
    • Agents only see topics from their namespace
    • ACLs are evaluated within the namespace context

    7. Agent Security

    Sandboxing

    Agents have strict limits:

    Properties
    1# streamflow.properties
2agent.max.actions.per.minute=100
3agent.max.tokens.per.day=1000000
4agent.strict.mode=true           # confirmation required for actions

    Blocked Commands (regex patterns)

    1rm\s+-rf           # recursive deletion
2DROP\s+TABLE       # SQL destruction
3FORMAT\s+          # disk formatting
4shutdown           # system shutdown
5fork.*bomb         # fork bomb

    ReasoningAuthorizer

    Before each tool call, the agent goes through the authorizer:

    Agent "fraud-detector" wants to call tool "database-query"
    ↓
ReasoningAuthorizer.authorize(agentId, toolId, action)
    ↓
RBAC check: does fraud-detector have Permission("tool:database-query", "execute")?
    ↓
YES → execute the tool
NO  → ReasoningAuthorizationException → agent receives "Permission denied"

    8. Production Configuration

    Security Checklist

    Properties
    1# ══════════════════════════════════════════════════
2# streamflow.properties, PRODUCTION SECURITY
3# ══════════════════════════════════════════════════
4
5# ── 1. TLS everywhere ──
6grpc.tls.enabled=true
7grpc.tls.mutual=true
8grpc.tls.cert.path=/opt/streamflow/certs/server.crt
9grpc.tls.key.path=/opt/streamflow/certs/server.key
10grpc.tls.trust.cert.path=/opt/streamflow/certs/ca.crt
11controller.network.tls.enabled=true
12controller.network.tls.mutual=true
13
14# ── 2. SASL for Kafka clients ──
15kafka.protocol.sasl.enabled=true
16kafka.protocol.sasl.credentials=${KAFKA_SASL_CREDENTIALS}
17
18# ── 3. Topic ACL ──
19topic.acl.enabled=true
20
21# ── 4. RBAC for agents ──
22mcp.rbac.enabled=true
23
24# ── 5. Encryption at rest ──
25log.encryption.enabled=true
    bash
    1# Sensitive environment variables
2export PULSE_MASTER_KEY=$(openssl rand -base64 32)
3export LOG_ENCRYPTION_KEY=$(openssl rand -base64 32)
4export KAFKA_SASL_CREDENTIALS="admin:$(openssl rand -hex 16),app:$(openssl rand -hex 16)"

    Generate TLS Certificates

    bash
    1# 1. Create the CA
2openssl genrsa -out ca.key 4096
3openssl req -new -x509 -days 3650 -key ca.key -out ca.crt \
4  -subj "/CN=StreamFlow CA"
5
6# 2. Create server certificate (for each node)
7openssl genrsa -out server.key 2048
8openssl req -new -key server.key -out server.csr \
9  -subj "/CN=streamflow-node1"
10openssl x509 -req -days 365 -in server.csr -CA ca.crt -CAkey ca.key \
11  -CAcreateserial -out server.crt
12
13# 3. Create client certificate (for mTLS)
14openssl genrsa -out client.key 2048
15openssl req -new -key client.key -out client.csr \
16  -subj "/CN=my-application"
17openssl x509 -req -days 365 -in client.csr -CA ca.crt -CAkey ca.key \
18  -CAcreateserial -out client.crt

    Ports and Protocols Summary

    PortProtocolSecurityUsage
    9092Kafka wireSASL + TLSProducers/Consumers
    8080HTTP RESTJWT/API Key + TLSAdmin API, Agents
    9169gRPCmTLSInter-node Raft replication
    5679TCPmTLSCluster membership (Atomix)

    What is Auditable

    ActionLogDetails
    AuthenticationWARN if failedSource IP, username
    ACL deniedWARNprincipal, topic, action
    RBAC deniedWARNagent, tool, permission
    Agent tool callINFO if audit modeagent, tool, parameters
    API key grant/revokeINFOkey hash, admin
    StreamFlow© 2026 StreamFlow, Built for real-time AI at scale.