Retour aux Guides

    Guide de Sécurité

    Authentification, autorisation, chiffrement, isolation multi-tenant et sécurité des agents.

    Table des Matières

    1. 1.Vue d'ensemble
    2. 2.Authentification (qui es-tu ?)
    3. 3.Autorisation (que peux-tu faire ?)
    4. 4.Chiffrement en Transit (TLS/mTLS)
    5. 5.Chiffrement au Repos
    6. 6.Isolation Multi-Tenant
    7. 7.Sécurité des Agents
    8. 8.Configuration de Production

    1. Vue d'ensemble

    ┌─────────────────────────────────────────────────────────────────┐
│                      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. Authentification (qui es-tu ?)

    Kafka Wire Protocol (SASL PLAIN)

    Les producers/consumers s'authentifient 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"

    Format JWT

    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)

    L'identité est résolue dans cet ordre :

    • Certificat client mTLS → CN (Common Name) extrait
    • Header Authorization: Bearer <token> → JWT validé
    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. Autorisation (que peux-tu faire ?)

    ACL par Topic (deny-by-default)

    Chaque opération sur un topic est évaluée contre des règles ACL. Par défaut : tout est refusé.

    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();

    Ordre de Résolution

    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 pour Agents et Tools MCP

    Les agents ont des permissions limitées sur les tools qu'ils peuvent appeler :

    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

    Hiérarchie des Rôles

    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. Chiffrement en 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 (protection contre MITM)

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

    Le pin est vérifié au handshake TLS, zéro coût par message.

    5. Chiffrement au Repos

    Snapshots Raft

    Les snapshots de l'état du cluster sont chiffrés automatiquement :

    bash
    1# Environment variable for the master key (32 bytes, Base64)
2export PULSE_MASTER_KEY=$(openssl rand -base64 32)
    • Algorithme : AES-256-GCM (chiffrement authentifié)
    • IV aléatoire : 12 bytes par opération
    • Tag d'authentification : intégré (protège contre la falsification)
    • Effacement auto : le plaintext est automatiquement effacé de la mémoire après 100ms

    Données des Segments (Event Log)

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

    6. Isolation Multi-Tenant

    Modèle de Tenants

    ┌─────────────────────────────────────────────────────┐
│ 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  │             │
│ └──────────────────────────────────────┘             │
└─────────────────────────────────────────────────────┘

    Isolation par Namespace

    • Chaque tenant a son propre namespace
    • Les topics sont préfixés : <namespace>.<topic>
    • Les agents ne voient que les topics de leur namespace
    • Les ACL sont évaluées dans le contexte du namespace

    7. Sécurité des Agents

    Sandboxing

    Les agents ont des limites strictes :

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

    Commandes Bloquées (patterns regex)

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

    ReasoningAuthorizer

    Avant chaque appel d'outil, l'agent passe par l'autoriseur :

    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. Configuration de Production

    Checklist de Sécurité

    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)"

    Générer les Certificats TLS

    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

    Résumé des Ports et Protocoles

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

    Ce qui est Auditable

    ActionLogDétails
    AuthentificationWARN if failedIP source, username
    ACL refuséWARNprincipal, topic, action
    RBAC refuséWARNagent, tool, permission
    Agent tool callINFO if audit modeagent, tool, paramètres
    API key grant/revokeINFOkey hash, admin