# Security Best Practices for CocoIndex Connectors: A Complete Guide

> Secure your CocoIndex connectors with this complete guide. Learn about ContextKey, TLS, and least-privilege IAM for robust data security in your pipelines.

- Repository: [CocoIndex/cocoindex](https://github.com/cocoindex-io/cocoindex)
- Tags: best-practices
- Published: 2026-05-05

---

**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`](https://github.com/cocoindex-io/cocoindex/blob/main/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.

```python
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.

```python
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.

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`.

```python
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`](https://github.com/cocoindex-io/cocoindex/blob/main/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:

```bash
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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/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`](https://github.com/cocoindex-io/cocoindex/blob/main/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.