How to Secure AI Ingestion Pipelines: Connector to Corpus

TL;DR — Securing AI ingestion pipelines requires five controls: (1) isolate OAuth credentials with envelope encryption and per-tenant keys, (2) redact PII before chunking and embedding so sensitive data never reaches the corpus, (3) encrypt data in transit (TLS 1.3) and at rest (AES-256), (4) enforce document-level ACLs that propagate from source systems into the vector store, and (5) maintain immutable audit trails for every document ingested. Most AI data breaches occur at the ingestion layer, not the model layer — attackers target connector credentials and intermediate processing stages where data exists in unencrypted form.

AI platforms ingest data from dozens of sources: Slack messages, Gmail threads, Google Drive documents, GitHub repositories, Jira tickets. Each source connects through an OAuth-based connector that pulls data into a processing pipeline. That pipeline chunks, embeds, and stores documents in a searchable corpus.

The ingestion layer is where most AI security failures happen. Not at the model. Not at the API. At the pipeline that moves data from source to corpus.

A CISA advisory on AI data security (May 2025) identified the data supply chain as a primary attack surface, recommending cryptographic integrity checks, data provenance tracking, and secure enclaves for processing. The FBI's own guidance states that "training data is the primary attack surface that most AI security strategies leave unguarded."

This post covers the full ingestion security stack: from connector credential management through PII redaction, encryption, ACL propagation, and audit trails.

What Are the Threats to AI Ingestion Pipelines?

AI ingestion pipelines face five categories of threat:

Credential theft is the most common. Connectors need OAuth tokens to access source systems. If those tokens are stored in environment variables, configuration files, or database columns in plaintext, any attacker with read access to the infrastructure can extract them and impersonate the AI platform.

PII exposure is the most damaging. If personally identifiable information (emails, phone numbers, SSNs) passes through the pipeline without redaction, it ends up in the vector database — searchable by anyone who can query the AI platform. This is a direct GDPR Article 5 violation and a HIPAA violation if the data contains protected health information.

How to Secure Connector Credentials?

Connector credentials are the keys to your data kingdom. Every source system (Slack, Gmail, Drive) grants OAuth tokens with specific scopes. Securing these tokens is the first and most critical control.

Credential Storage Architecture

The following diagram shows a secure credential storage architecture using envelope encryption:

Never store OAuth tokens in:
- Environment variables (visible in process listings, container inspection)
- Plaintext database columns (exposed in database dumps, backup leaks)
- Configuration files (committed to version control, readable by all developers)
- Application logs (accidental token leakage)

Always store OAuth tokens in:
- An encrypted secret manager (HashiCorp Vault, AWS Secrets Manager, or a custom Fernet-based store)
- Encrypted at the field level with per-tenant keys using envelope encryption
- Accessible only to the connector service with scoped IAM roles

from cryptography.fernet import Fernet
import base64
import hashlib

class CredentialStore:
    """Per-tenant encrypted credential storage using envelope encryption."""

    def __init__(self, kms_key: bytes):
        self.kms_key = kms_key  # Master key from KMS

    def _derive_tenant_key(self, tenant_id: str) -> bytes:
        """Derive a per-tenant Data Encryption Key (DEK) from the master key."""
        derived = hashlib.pbkdf2_hmac(
            'sha256',
            self.kms_key,
            tenant_id.encode(),
            iterations=100_000
        )
        return base64.urlsafe_b64encode(derived[:32])

    def store_token(self, tenant_id: str, source: str, token: str):
        """Encrypt and store an OAuth token for a specific tenant."""
        dek = self._derive_tenant_key(tenant_id)
        f = Fernet(dek)
        encrypted = f.encrypt(token.encode())
        # Store encrypted token in database with tenant_id + source as key
        # The DEK never persists — it's derived at runtime from the master key
        return encrypted

    def retrieve_token(self, tenant_id: str, source: str) -> str:
        """Decrypt and return an OAuth token for runtime use."""
        dek = self._derive_tenant_key(tenant_id)
        f = Fernet(dek)
        # Fetch encrypted token from database
        encrypted = self._fetch_encrypted(tenant_id, source)
        return f.decrypt(encrypted).decode()

OAuth Scope Minimization

Every OAuth grant should request the minimum scopes needed:

The AI platform should also implement OAuth credential security patterns: never store provider OAuth credentials, use just-in-time token refresh, and revoke tokens when a tenant disconnects.

How to Redact PII Before Storage?

PII redaction must happen immediately after data is pulled from the source system — before chunking, before embedding, before storage. This is covered in detail in our PII redaction guide, but the pipeline-specific implementation matters.

Redaction Pipeline Architecture

import re
from dataclasses import dataclass

@dataclass
class PIIPattern:
    name: str
    pattern: re.Pattern
    replacement: str

PII_PATTERNS = [
    PIIPattern("email", re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'), '[EMAIL]'),
    PIIPattern("phone", re.compile(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b'), '[PHONE]'),
    PIIPattern("ssn", re.compile(r'\b\d{3}-\d{2}-\d{4}\b'), '[SSN]'),
    PIIPattern("credit_card", re.compile(r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b'), '[CC]'),
]

def redact_pii(text: str, tenant_id: str) -> str:
    """Redact PII from text before chunking and embedding."""
    for pattern in PII_PATTERNS:
        text = pattern.pattern.sub(pattern.replacement, text)
    return text

def process_document(doc: dict, tenant_id: str) -> dict:
    """Process a single document through the secure ingestion pipeline."""
    # Step 1: Redact PII
    doc['content'] = redact_pii(doc['content'], tenant_id)

    # Step 2: Compute content hash for deduplication
    import hashlib
    doc['content_hash'] = hashlib.sha256(
        doc['content'].encode()
    ).hexdigest()

    # Step 3: Preserve source ACLs
    doc['acl'] = doc.get('source_permissions', [])

    return doc

Key principle: PII redaction is a pipeline stage, not a feature flag. Every document passes through redaction regardless of tenant configuration. This prevents configuration errors from exposing PII.

What to Redact

The redaction list should be configurable per tenant. A healthcare tenant may need to redact medical record numbers. A financial tenant may need to redact account numbers. The base patterns cover universal PII; tenant-specific patterns extend the list.

How to Encrypt Data in Transit and at Rest?

In Transit

All data movement between pipeline stages must use TLS 1.3. No exceptions.

At Rest

Every storage layer in the pipeline must encrypt data at rest:

For multi-tenant AI platforms, field-level encryption with per-tenant keys ensures that one tenant's data is unreadable by another, even if they share the same database.

How to Propagate Document-Level ACLs?

Source systems have permissions. A Google Drive document might be visible to only the engineering team. A Slack message might be in a private channel. These permissions must propagate through the ingestion pipeline into the corpus.

ACL Propagation Flow

The retrieval system must filter results by the requesting user's permissions. This is covered in detail in our guide to document-level ACLs for AI retrieval.

def retrieve_with_acls(
    query_embedding: list[float],
    user_groups: list[str],
    vector_store,
    top_k: int = 10
) -> list[dict]:
    """Retrieve documents filtered by user's ACL memberships."""
    results = vector_store.search(
        query_embedding,
        top_k=top_k * 3,  # Over-fetch to compensate for ACL filtering
        filter={"acl": {"$in": user_groups}}
    )
    return results[:top_k]

Critical rule: ACLs are enforced at the storage layer, not the application layer. If the vector store does not support ACL filtering natively, use a pre-filter step that removes inaccessible documents before retrieval. Never rely on post-retrieval filtering — it leaks metadata through the model's context window.

How to Maintain Audit Trails?

Every document that enters the pipeline must have a complete audit trail:

ingestion_audit_entry:
  document_id: "doc_abc123"
  tenant_id: "tenant_001"
  source: "google_drive"
  source_document_id: "1a2b3c4d5e6f"
  ingested_at: "2026-07-14T10:30:00Z"
  processed_by: "ingestion-worker-3"
  content_hash: "sha256:9f86d081..."
  pii_redacted: true
  pii_patterns_found: ["email:2", "phone:1"]
  acl_inherited: ["engineering-team", "leadership"]
  chunk_count: 15
  embedding_model: "text-embedding-3-small"
  status: "success"

The audit trail must be:
- Immutable — append-only, no updates or deletes
- Queryable — filterable by tenant, source, date, status
- Retained — per regulatory requirements (GDPR: 2 years, HIPAA: 6 years)
- Exportable — for compliance audits and incident investigations

How to Prevent Data Poisoning?

Data poisoning attacks inject malicious documents through compromised connectors or supply chain attacks. The pipeline must validate document integrity before processing.

Content Hash Verification

import hashlib

def verify_document_integrity(doc: dict, expected_hash: str) -> bool:
    """Verify document hasn't been tampered with since ingestion."""
    actual_hash = hashlib.sha256(doc['content'].encode()).hexdigest()
    return actual_hash == expected_hash

def deduplicate_by_hash(doc: dict, existing_hashes: set[str]) -> bool:
    """Skip re-processing unchanged documents."""
    if doc['content_hash'] in existing_hashes:
        return False  # Already ingested, skip
    existing_hashes.add(doc['content_hash'])
    return True

Content hashing serves double duty: it prevents re-processing unchanged documents (near-free re-ingestion) and detects tampering. If a document's hash changes between syncs without a corresponding source update, the pipeline should flag it for investigation.

Source Verification

Every connector should verify the source system's identity using:
- OAuth token validation (check token scopes and expiry)
- Source system API certificate pinning
- Rate limiting to detect anomalous ingestion volumes
- Document source attribution (who created the document, when, from where)

Pipeline Security Checklist

• [ ] OAuth tokens stored in encrypted secret manager (not env vars)
• [ ] Per-tenant encryption keys using envelope encryption
• [ ] OAuth scopes minimized to read-only where possible
• [ ] TLS 1.3 for all data movement between pipeline stages
• [ ] AES-256 encryption at rest for all storage layers
• [ ] PII redaction before chunking and embedding
• [ ] Content hashing for deduplication and integrity verification
• [ ] Document-level ACLs propagated from source to corpus
• [ ] ACL filtering at the storage layer (not application layer)
• [ ] Immutable audit trail for every ingested document
• [ ] Rate limiting on connectors to detect anomalous ingestion
• [ ] Token rotation on a regular schedule
• [ ] Token revocation when tenants disconnect
• [ ] Source system identity verification
• [ ] Data poisoning detection via hash verification

FAQ

What is an AI ingestion pipeline?

An AI ingestion pipeline is the set of components that move data from source systems (Slack, Gmail, Drive, GitHub) through connectors, processing, PII redaction, chunking, embedding, and into a searchable corpus (vector database, knowledge graph). Securing this pipeline means controlling access at every stage, encrypting data in transit and at rest, redacting PII before storage, and maintaining audit trails for every document processed.

Where do most AI data breaches occur?

Most AI data breaches occur at the ingestion layer, not the model layer. Attackers target OAuth credentials stored in plaintext, connector configurations with excessive scopes, and intermediate processing stages where data exists in unencrypted form. The model itself is rarely the attack surface — the pipeline that feeds it is.

Should PII be redacted before or after embedding?

Before. PII redaction must happen before chunking and embedding so that sensitive data never reaches the vector database or knowledge graph. Redacting after embedding means PII is already stored in searchable form, creating a data breach risk. Redaction should occur immediately after data is pulled from the source system, before any processing.

How do you secure OAuth credentials for AI connectors?

Store OAuth tokens in an encrypted secret manager (not in environment variables or database columns in plaintext). Use envelope encryption with per-tenant keys. Rotate tokens on a schedule. Scope OAuth grants to the minimum permissions needed. Never log tokens in application logs or audit trails. Use separate credentials per tenant in multi-tenant systems.

What is content hashing in AI ingestion?

Content hashing computes a SHA-256 hash of each document's content before ingestion. When a document is re-ingested (e.g., after an edit), the hash is compared. If the hash matches, the document is skipped — no re-chunking, re-embedding, or re-indexing needed. This makes re-ingestion near-free for unchanged documents and provides an integrity check for detecting tampering.

Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →