Detect data exfiltration via RAG

Introduction

When attackers pull sensitive documents out of a RAG system one legitimate-looking query at a time, naive per-second rate limits never trip and the audit log looks like ordinary research traffic — by the time anyone notices, the corpus has already leaked. Unlike a direct database breach, exfiltration via RAG happens through the production query API: an adversary uses the same retrieval interface as your real users, just systematically and patiently. By the end of this lesson you'll be able to instrument three independent behavioral signals — query volume, query coverage, and response sensitivity — combine them into a composite risk score, and use that score to throttle or alert on suspected exfiltration before a full collection leaks.

Key Terminology

• Exfiltration via RAG: an attack in which an adversary uses a legitimate retrieval interface to systematically extract sensitive documents over many queries, rather than compromising the underlying vector store directly.
• Sliding window: a moving time interval (e.g. the last 3,600 seconds) used to count recent queries per user without resetting at fixed clock boundaries; implemented here with Redis sorted sets keyed by Unix timestamp.
• Query coverage: the fraction of unique documents in a collection that a single user has retrieved within a rolling window; high coverage (e.g. >80% in a week) is a strong indicator of systematic extraction rather than targeted research.
• Response sensitivity ratio: the share of documents returned to a user that are classified confidential or restricted; a sudden spike above a baseline (e.g. >30%) indicates anomalous access to sensitive content.
• Composite risk score: a weighted average of the volume, coverage, and sensitivity signals normalized to [0.0, 1.0], used as the single decision input for throttling and alerting.

Concepts

Detecting RAG exfiltration is a behavioral problem, not an authentication problem — every query under analysis is from an authorized user. The detection system therefore reasons about patterns over time, using three complementary signals so an attacker who evades one still trips another.

The volume signal counts a user's queries in a sliding window. It catches the naive case of an attacker hammering the API, but it misses slow, evenly paced extraction. The coverage signal closes that gap: it tracks the set of distinct document IDs each user has retrieved per collection and flags a user once their unique-document set crosses a fraction of the collection — slow pacing does not help, because every retrieved document still adds to the set. The sensitivity signal complements both by watching the classification level of returned documents, so an attacker who stays under volume and coverage thresholds but disproportionately pulls confidential content is still surfaced.

Each signal is normalized into a score on [0.0, 1.0] and combined into a composite risk score with configurable weights. A single decision threshold on the composite score drives the response: below it, the query is served normally; above it, the system throttles the user (rate-limit delays or truncated result sets) and emits an alert for human review. Keeping state in Redis (sorted sets for windows, sets for coverage, hashes for sensitivity counters) lets every signal expire automatically with a TTL, so the detector is bounded in memory and naturally forgets stale behavior.

Code Walkthrough

Query Volume and Coverage Tracking

The first two signals are quantitative behavioral counters per user. Volume tracks how many queries a user has made inside a sliding time window — it catches naive attackers who hammer the API, but a slow attacker can evade it by pacing requests. Coverage closes that gap by tracking the set of distinct document IDs each user has retrieved per collection; slow pacing does not help, because every retrieved doc still adds to the set. The combined snippet below uses Redis sorted sets for the volume sliding window and Redis sets for the coverage set, so both are bounded in memory via key TTL and cheap to update on the hot path.

Code snippetpython
1import time
2from dataclasses import dataclass
3from typing import Optional
4
5@dataclass
6class VolumeAlert:
7    """Alert generated when query volume exceeds threshold."""
8    user_id: str
9    query_count: int
10    window_seconds: int
11    threshold: int
12    timestamp: float
13
14@dataclass
15class CoverageAlert:
16    """Alert for suspicious query coverage patterns."""
17    user_id: str
18    collection_id: str
19    documents_retrieved: int
20    total_documents: int
21    coverage_percentage: float
22    threshold: float
23
24class QueryVolumeTracker:
25    """Tracks per-user query volume using Redis sliding windows."""
26
27    def __init__(
28        self,
29        redis_client,
30        window_seconds: int = 3600,
31        threshold: int = 50,
32    ):
33        self.redis = redis_client
34        self.window = window_seconds
35        self.threshold = threshold
36
37    def record_query(self, user_id: str) -> Optional[VolumeAlert]:
38        now = time.time()
39        key = f"query_volume:{user_id}"
40        self.redis.zadd(key, {f"{now}:{user_id}": now})
41        cutoff = now - self.window
42        self.redis.zremrangebyscore(key, "-inf", cutoff)
43        self.redis.expire(key, self.window + 60)
44        count = self.redis.zcard(key)
45        if count > self.threshold:
46            return VolumeAlert(
47                user_id=user_id,
48                query_count=count,
49                window_seconds=self.window,
50                threshold=self.threshold,
51                timestamp=now,
52            )
53        return None
54
55class CoverageMonitor:
56    """Monitors per-user document retrieval coverage."""
57
58    def __init__(
59        self,
60        redis_client,
61        coverage_threshold: float = 0.8,
62        window_seconds: int = 604800,
63    ):
64        self.redis = redis_client
65        self.coverage_threshold = coverage_threshold
66        self.window = window_seconds
67
68    def record_retrieval(
69        self,
70        user_id: str,
71        collection_id: str,
72        document_ids: list[str],
73        total_documents: int,
74    ) -> Optional[CoverageAlert]:
75        key = f"coverage:{user_id}:{collection_id}"
76        for doc_id in document_ids:
77            self.redis.sadd(key, doc_id)
78        self.redis.expire(key, self.window)
79        retrieved_count = self.redis.scard(key)
80        coverage = retrieved_count / max(total_documents, 1)
81        if coverage > self.coverage_threshold:
82            return CoverageAlert(
83                user_id=user_id,
84                collection_id=collection_id,
85                documents_retrieved=retrieved_count,
86                total_documents=total_documents,
87                coverage_percentage=round(coverage, 4),
88                threshold=self.coverage_threshold,
89            )
90        return None

• VolumeAlert / CoverageAlert carry the user, the observed value, and the threshold that was crossed so downstream alerting is self-describing.
• QueryVolumeTracker uses a Redis sorted set whose members are query IDs and scores are Unix timestamps: ZADD to insert, ZREMRANGEBYSCORE to trim everything older than now - window, then ZCARD for the count. The key TTL is set just past the window so idle users' state evicts automatically.
• CoverageMonitor uses a Redis set keyed by (user, collection): every retrieved doc ID is SADD-ed, the set's cardinality divided by total_documents gives the coverage ratio. A user who has seen >80% of documents in a week is exfiltrating regardless of how slowly they paced queries.

Response Sensitivity Monitoring

The third signal tracks the classification level of documents returned to each user. A user who normally retrieves public and internal documents but suddenly starts retrieving many confidential documents is exhibiting anomalous behavior that warrants investigation. The sensitivity monitor maintains a running count of documents retrieved at each classification level and alerts when the confidential-and-above ratio exceeds a configurable threshold.

The SensitivityMonitor class below uses Redis hashes to track per-user document counts by classification level. After each query, it updates the counts and computes the sensitive document ratio. This signal complements the volume and coverage signals: a user might make a normal number of queries with low coverage, but if every query retrieves confidential documents, the sensitivity signal will trigger an alert.

1@dataclass
2class SensitivityAlert:
3    """Alert for unusual response sensitivity patterns."""
4    user_id: str
5    sensitive_count: int
6    total_count: int
7    sensitivity_ratio: float
8    threshold: float
9
10class SensitivityMonitor:
11    """Monitors classification levels of retrieved documents."""
12
13    SENSITIVE_LEVELS = {"confidential", "restricted"}
14
15    def __init__(
16        self,
17        redis_client,
18        sensitivity_threshold: float = 0.3,
19        window_seconds: int = 86400,
20    ):
21        self.redis = redis_client
22        self.sensitivity_threshold = sensitivity_threshold
23        self.window = window_seconds
24
25    def record_response(
26        self,
27        user_id: str,
28        document_levels: list[str],
29    ) -> Optional[SensitivityAlert]:
30        """Record classification levels and check sensitivity.
31
32        Args:
33            user_id: The querying user's ID.
34            document_levels: Classification levels of returned docs.
35
36        Returns:
37            SensitivityAlert if sensitive ratio exceeds threshold.
38        """
39        key = f"sensitivity:{user_id}"
40        for level in document_levels:
41            self.redis.hincrby(key, level, 1)
42        self.redis.expire(key, self.window)
43        counts = self.redis.hgetall(key)
44        total = sum(int(v) for v in counts.values())
45        sensitive = sum(
46            int(counts.get(level, 0))
47            for level in self.SENSITIVE_LEVELS
48        )
49        ratio = sensitive / max(total, 1)
50        if ratio > self.sensitivity_threshold:
51            return SensitivityAlert(
52                user_id=user_id,
53                sensitive_count=sensitive,
54                total_count=total,
55                sensitivity_ratio=round(ratio, 4),
56                threshold=self.sensitivity_threshold,
57            )
58        return None

Composite Risk Score and Throttling

The three signals combine into a composite exfiltration risk score per user. Each signal produces a normalized score between 0.0 and 1.0, and the composite score is a weighted average with configurable weights. When the composite score exceeds the throttling threshold, the system automatically reduces the user's query rate by introducing delays or returning fewer results.

The diagram captures how the three per-signal scores feed a single composite-score gate that decides between throttle-plus-alert and pass-through on each query.

Do's and Don'ts

Do's

1. ✓Do use Redis sorted sets with ZADD + ZREMRANGEBYSCORE + ZCARD for the volume sliding window — QueryVolumeTracker trims entries older than now - window on every write so ZCARD always reflects the true rolling interval; a fixed-bucket counter would reset at the boundary and let an attacker time bursts to stay under the threshold.
2. ✓Do key the CoverageMonitor Redis set as coverage:{user_id}:{collection_id} and let it accumulate across sessions — because SADD is idempotent per document ID, a slow attacker who paces queries over days still drives the coverage ratio monotonically toward the coverage_threshold, making low-and-slow extraction detectable even when volume never spikes.
3. ✓Do combine VolumeAlert, CoverageAlert, and SensitivityAlert into a composite risk score before acting — volume alone is evaded by pacing, coverage alone misses targeted confidential-document pulls, and sensitivity alone misses full-collection sweeps; only the composite catches all three evasion strategies simultaneously.

Don'ts

1. ✗Don't omit the ZREMRANGEBYSCORE trim before reading ZCARD in QueryVolumeTracker — without pruning entries older than now - window, the sorted set accumulates every historical query and ZCARD returns the all-time count, causing any sufficiently active legitimate user to permanently exceed the threshold and generating continuous false-positive alerts.
2. ✗Don't pass a stale or startup-cached total_documents to CoverageMonitor.record_retrieval — coverage is computed as scard(key) / max(total_documents, 1), so a denominator that doesn't grow alongside the collection silently inflates the ratio for legitimate users while letting an attacker who seeds the collection with new documents exfiltrate them below the coverage_threshold = 0.8 floor.
3. ✗Don't collapse SENSITIVE_LEVELS to a single classification tier — SensitivityMonitor counts sensitive documents as the union of {"confidential", "restricted"}; removing either tier means a user who retrieves exclusively documents at the omitted level contributes zero to the sensitive numerator, bypassing the sensitivity_threshold = 0.3 check entirely.