Security Best Practices for CocoIndex Connectors: A Complete Guide

CocoIndex connectors use ContextKey dependency injection and environment variable isolation to ensure credentials never appear in pipeline code, while enforcing TLS encryption and least-privilege IAM roles for all external systems.

CocoIndex is an open-source Python framework for building data pipelines that interact with external storage systems like PostgreSQL, Kafka, and S3. Because these connectors handle sensitive authentication material and network connections, implementing robust security best practices for cocoindex connectors is critical to prevent credential exposure and unauthorized data access. The library's architecture is designed to keep secrets out of component functions through dependency injection patterns, but proper configuration by the developer remains essential.

Isolate Credentials Using ContextKey

All CocoIndex connectors expect a ContextKey that provides a pre-initialized client or connection pool rather than raw credentials. This key is resolved once per process during application startup and injected into the component's lifespan, ensuring that connection strings and passwords never appear in pipeline logic or stack traces.

In postgres/_target.py, the declare_table_target() function accepts a ContextKey[asyncpg.Pool] and stores only the key reference, never accessing the DSN directly. The same pattern applies to Kafka (ContextKey[AIOConsumer]), S3 (ContextKey[AioSession]), OCI Object Storage, and Neo4j connectors.

import asyncpg
import cocoindex as coco
from cocoindex.connectors.postgres import declare_table_target
import os

# Create a ContextKey that provides an asyncpg.Pool

POSTGRES_POOL = coco.ContextKey[asyncpg.Pool]("postgres_pool")

# Populate the context from environment variables in a lifespan builder

builder = coco.LifespanBuilder()
builder.provide(
    POSTGRES_POOL, 
    asyncpg.create_pool(dsn=os.getenv("POSTGRES_DSN"), ssl="verify-full")
)

# Use the key when creating a target—no credentials in pipeline code

target = declare_table_target(
    db=POSTGRES_POOL,
    table_name="events",
    table_schema=schema,
)

By centralizing secret material in the lifespan builder, you guarantee that credentials are loaded once at startup, cannot be accidentally logged or pickled, and can be easily substituted with mock pools for testing.

Load Secrets from Environment Variables

CocoIndex does not read environment variables directly. Instead, read secrets in your startup code and pass the resulting objects to the ContextProvider. This isolates secret handling from the library and allows you to use tools like python-dotenv or CI secret injection without exposing variables to the framework.

from dotenv import load_dotenv
import os

load_dotenv()  # Reads .env into os.environ safely

builder.provide(
    POSTGRES_POOL,
    asyncpg.create_pool(dsn=os.getenv("POSTGRES_DSN"))
)

Never hardcode credentials in Python files or commit them to version control. Always validate that .env files are listed in .gitignore before deploying.

Enforce Least-Privilege Permissions

When provisioning external resources, grant only the actions required by the specific connector. Overly permissive roles increase the blast radius of a compromised credential.

Connector	Minimal Required Permissions
PostgreSQL	SELECT, INSERT, UPDATE, DELETE on target tables. Avoid SUPERUSER or schema-wide CREATE unless using managed_by=SYSTEM.
Kafka	READ on topics plus OFFSET_COMMIT for the consumer group. Do not grant WRITE unless producing.
S3 / OCI	GetObject, ListBucket, and ListObjectVersions if using live-bucket watching. Avoid PutObject unless writing.
Neo4j	READ/WRITE on specific node and relationship types only.

For relational targets, use managed_by=coco.target.ManagedBy.USER to ensure tables are pre-created with audited permissions rather than defaulting to system-generated privileges.

declare_table_target(
    db=POSTGRES_POOL,
    table_name="events",
    table_schema=schema,
    managed_by=coco.target.ManagedBy.USER,  # Table must exist with restricted grants

)

Enable TLS for All Network Connections

All connectors support encrypted transports. Configure TLS in the client objects you create before passing them to ContextKey, keeping certificates out of connector logic.

• PostgreSQL: Pass sslmode="verify-full" (or "require" for self-signed certs) when creating the pool.
• Kafka: Configure AIOConsumer with security.protocol="SSL" and supply CA/client certificates via ssl.ca.location, ssl.certificate.location, and ssl.key.location.
• S3/OCI: Ensure endpoint URLs use https://; the SDKs default to HTTPS but verify the configuration explicitly.

Because TLS configuration lives in the client initialization code, the connector never handles raw certificate data.

Handle Kafka Offsets Safely

CocoIndex's Kafka source (topic_as_map and topic_as_stream) commits offsets only after downstream components signal readiness, guaranteeing exactly-once processing for stateful pipelines. The connector explicitly disables broker auto-commit to prevent duplicate processing during crashes.

In kafka/_source.py, the _PartitionState.track() and _try_drain_and_commit() methods manage this logic. If you attempt to enable enable.auto.commit=True in the consumer configuration, the connector raises an ImportError.

KAFKA_CONSUMER = coco.ContextKey["confluent_kafka.aio.AIOConsumer"]("kafka_consumer")

consumer = await coco.connectors.kafka.AIOConsumer({
    "bootstrap.servers": os.getenv("KAFKA_BOOTSTRAP"),
    "group.id": "my_group",
    "enable.auto.commit": False,  # Required: must be False

    "security.protocol": "SSL",
    "ssl.ca.location": "/etc/ssl/certs/ca.pem",
})

Prevent Resource Leaks and Blocking I/O

All target and source implementations are async-first, using asyncpg, confluent_kafka.aio, and async SDKs. When writing custom encoders or callbacks, ensure functions are pure CPU work or return quickly. Offload heavy computation to a thread pool via coco.run_in_executor to avoid blocking the event loop and delaying critical offset commits.

When components are destroyed (e.g., tables dropped or topics unmounted), connector attachment handlers run teardown logic. Follow the pattern shown in _SqlCommandHandler.reconcile in postgres/_target.py to ensure resources are removed atomically and prevent orphaned objects that could leak data.

Maintain Code Hygiene and Dependencies

CocoIndex enforces strict import-prefix conventions (e.g., import asyncpg as _asyncpg) to keep the public API surface small. When extending connectors, follow this pattern to prevent accidental exposure of low-level client objects.

Security vulnerabilities often reside in third-party libraries. Regularly update dependencies using the repository's uv workflow:

uv lock --upgrade
uv sync

Run uv run mypy and uv run ruff check . after upgrades to ensure type safety and catch potential security anti-patterns.

Summary

• Use ContextKey to inject connection pools and clients, keeping credentials out of pipeline functions and logs.
• Load secrets via environment variables or secret managers in the lifespan builder, never hardcoding them in source files.
• Grant least-privilege permissions to database users and IAM roles, limiting access to specific tables, topics, or buckets.
• Enable TLS/SSL for all network connections, configuring certificates during client initialization.
• Disable Kafka auto-commit to ensure exactly-once processing and prevent offset drift during failures.
• Avoid blocking I/O in async callbacks to prevent event loop starvation and delayed commits.
• Implement proper cleanup in attachment handlers to remove resources atomically when components teardown.
• Pre-validate schemas using managed_by=USER to prevent automatic creation of over-permissive tables.
• Follow import conventions and run static analysis to minimize API surface area.
• Update dependencies regularly using uv lock --upgrade to patch transitive vulnerabilities.

Frequently Asked Questions

How does ContextKey prevent credential leaks in CocoIndex pipelines?

ContextKey uses dependency injection to provide initialized clients (like asyncpg.Pool or AIOConsumer) to connectors without passing raw connection strings or passwords through pipeline code. Because the connector stores only the key reference—as seen in declare_table_target() in postgres/_target.py—credentials are resolved once at startup and never appear in component function bodies, stack traces, or logs.

What SSL configuration is recommended for PostgreSQL connectors?

Pass sslmode="verify-full" when creating the asyncpg.Pool in your lifespan builder to enforce server certificate validation. For self-signed certificates in development environments, use sslmode="require". This configuration is applied during client initialization before the pool is injected via ContextKey, ensuring the connector itself never handles certificate paths or SSL modes directly.

Why does CocoIndex disable Kafka auto-commit by default?

The Kafka connector in kafka/_source.py manages offsets manually through _PartitionState.track() and commits only after downstream components confirm successful processing. This prevents duplicate message processing during consumer crashes. If enable.auto.commit is set to True, the connector raises an ImportError because automatic commits would violate the exactly-once guarantees required for stateful pipeline consistency.

How should I handle resource cleanup when dropping tables or topics?

Implement cleanup logic in attachment handlers following the pattern in _SqlCommandHandler.reconcile within postgres/_target.py. When a target state is deleted, these handlers execute appropriate teardown SQL or final offset commits atomically. This ensures that dropping a table or unmounting a topic removes all associated triggers, indexes, or temporary files without leaving orphaned resources that could retain sensitive data.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}