# assessing-vector-and-embedding-weaknesses

Test RAG vector stores (Pinecone, Qdrant, Weaviate, Chroma, pgvector, FAISS) for embedding inversion, cross-tenant data leakage, and data poisoning per OWASP LLM08:2025. Use when performing an authorized security assessment of a RAG pipeline's retrieval layer or auditing multi-tenant vector-store isolation.

- **Kind:** skill
- **Source:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **Page:** https://forefy.com/skills/5005304f-1843-4c8e-8de9-d340b6914bc5
- **API (JSON + files):** https://forefy.com/api/asr/5005304f-1843-4c8e-8de9-d340b6914bc5

---

## LICENSE

```

```

## SKILL.md

---
name: assessing-vector-and-embedding-weaknesses
description: Test RAG vector stores (Pinecone, Qdrant, Weaviate, Chroma, pgvector,
  FAISS) for embedding inversion, cross-tenant data leakage, and data poisoning per
  OWASP LLM08:2025. Use when performing an authorized security assessment of a RAG
  pipeline's retrieval layer or auditing multi-tenant vector-store isolation.
domain: cybersecurity
subdomain: ai-security
tags:
- ai-security
- vector-database
- embedding-inversion
- rag-security
- owasp-llm08
- multi-tenant-isolation
- data-poisoning
- retrieval-augmented-generation
version: '1.0'
author: mahipal
license: Apache-2.0
nist_ai_rmf:
- MEASURE-2.7
atlas_techniques:
- AML.T0024
---
# Assessing Vector and Embedding Weaknesses

> **Authorized use only:** These tests interact with vector stores and embedding models in RAG systems you own or are authorized to assess. Embedding inversion and cross-tenant probing against systems you do not control may expose third-party data and is prohibited without authorization.

## Overview

Retrieval-Augmented Generation (RAG) systems convert documents into embedding vectors stored in a vector database (Pinecone, Qdrant, Weaviate, Chroma, pgvector, FAISS) and retrieve the nearest vectors to ground LLM responses. OWASP **LLM08:2025 Vector and Embedding Weaknesses** covers the security risks unique to this layer:

- **Embedding inversion** — embeddings are not one-way. A trained inversion model (or a black-box reconstruction attack) can recover substantial portions of the original text from its vector, leaking source documents (maps to MITRE ATLAS **AML.T0024.001 Invert ML Model**).
- **Membership inference** — querying whether a specific record contributed to the corpus (AML.T0024.000).
- **Cross-tenant / multi-tenant leakage** — when one namespace/collection is shared or filter isolation is missing, a tenant retrieves another tenant's chunks.
- **Knowledge-base poisoning** — an attacker who can write to the corpus inserts crafted chunks that dominate retrieval (high cosine similarity to expected queries) and carry indirect prompt-injection payloads.
- **Retrieval manipulation** — adversarial documents tuned to be retrieved for many unrelated queries ("retrieval hijacking").

The parent technique is **AML.T0024 — Exfiltration via ML Inference API**: an attacker uses legitimate inference/query access to exfiltrate data (source text via inversion, membership, or model extraction). This skill provides a repeatable assessment of all five weakness classes.

## When to Use

- During a security assessment of any RAG / vector-search application (OWASP LLM08 coverage).
- When a vector store is multi-tenant and you must prove namespace/metadata isolation.
- When the corpus accepts user-supplied or third-party documents (poisoning surface).
- When the embedding endpoint is externally reachable (inversion/membership surface).
- When validating retrieval-filtering controls before go-live.

## Prerequisites

- Authorization and scope covering the target embedding endpoint and vector store.
- Python 3.10+.
- Read (and, for poisoning tests, write) access to a test collection — never the production corpus.

```bash
# Vector DB clients + embeddings + similarity tooling
python -m pip install numpy scikit-learn sentence-transformers
python -m pip install qdrant-client chromadb pinecone-client weaviate-client
# (optional) text-embedding inversion research baseline
python -m pip install vec2text
```

## Objectives

- Measure embedding-inversion exposure on the target embedding model.
- Run a membership-inference probe against the corpus.
- Test multi-tenant isolation (namespace, metadata filter, RBAC) for cross-tenant leakage.
- Inject benign poisoned chunks into a *test* collection and measure retrieval dominance.
- Detect indirect prompt-injection content surviving in retrieved chunks.
- Recommend controls: tenant-scoped filters, content validation, embedding-access limits.

## MITRE ATT&CK Mapping

| ID | Tactic | Official Technique Name | Role in this skill |
|----|--------|-------------------------|--------------------|
| AML.T0024 | ATLAS: Exfiltration | Exfiltration via ML Inference API | Using query/embedding access to exfiltrate source data |
| AML.T0024.000 | ATLAS: Exfiltration | Infer Training Data Membership | Membership-inference probe against the corpus |
| AML.T0024.001 | ATLAS: Exfiltration | Invert ML Model | Embedding-inversion reconstruction of source text |
| AML.T0020 | ATLAS: Resource Development | Poison Training Data | Knowledge-base poisoning of the corpus |
| AML.T0051.001 | ATLAS: Initial Access | LLM Prompt Injection: Indirect | Injection payloads embedded in retrieved chunks |

## Workflow

### Step 1: Inventory the RAG pipeline

Document the embedding model + dimensions, the vector store and its tenancy model, the chunking strategy, retrieval `top_k` and similarity metric (cosine/dot/L2), and any metadata filters applied at query time.

```python
# Example: inspect a Qdrant collection
from qdrant_client import QdrantClient
client = QdrantClient(url="http://localhost:6333")
info = client.get_collection("docs")
print(info.config.params.vectors)   # size + distance metric
print(client.count("docs"))         # corpus size
```

### Step 2: Test embedding-inversion exposure

Embeddings of similar text are close; an attacker with the embedding endpoint can iteratively reconstruct text whose embedding matches a target vector. Measure how much a nearest-neighbour-in-embedding-space recovers, using cosine similarity between candidate reconstructions and the target.

```python
import numpy as np
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

model = SentenceTransformer("all-MiniLM-L6-v2")
secret = "Patient John Doe, MRN 553120, diagnosed with hypertension."
target_vec = model.encode([secret])

# Attacker has only target_vec and the embedding endpoint. Hill-climb candidate text.
candidates = [
    "Patient name and medical record number with a diagnosis.",
    "John Doe medical record hypertension diagnosis",
    "Patient John Doe MRN diagnosed hypertension",
]
cand_vecs = model.encode(candidates)
sims = cosine_similarity(target_vec, cand_vecs)[0]
for c, s in sorted(zip(candidates, sims), key=lambda x: -x[1]):
    print(f"{s:.3f}  {c}")
# High similarity for a near-verbatim guess => inversion risk is real for this model.
```

For a research-grade reconstruction baseline, `vec2text` can be used against compatible embedding models to demonstrate full-text recovery.

### Step 3: Membership inference

Determine whether a specific document is in the corpus by measuring the top-1 retrieval similarity for an exact-quote query: in-corpus items return a markedly higher max similarity than out-of-corpus controls.

```python
def membership_score(client, collection, embed, text):
    vec = embed([text])[0].tolist()
    hits = client.search(collection_name=collection, query_vector=vec, limit=1)
    return hits[0].score if hits else 0.0

in_corpus = membership_score(client, "docs", model.encode, "<exact quote from a known chunk>")
control  = membership_score(client, "docs", model.encode, "An unrelated random sentence.")
print(f"in-corpus={in_corpus:.3f}  control={control:.3f}  delta={in_corpus-control:.3f}")
# A large positive delta indicates the item is in the corpus (membership leak).
```

### Step 4: Test multi-tenant isolation

Confirm that tenant B cannot retrieve tenant A's chunks. Issue tenant-B-authenticated queries that *should* be filtered, and verify no tenant-A `tenant_id` appears in results.

```python
# Query as tenant B; expect ONLY tenant_id == "B" results.
from qdrant_client.models import Filter, FieldCondition, MatchValue

vec = model.encode(["confidential salary information"])[0].tolist()
hits = client.search(
    collection_name="docs",
    query_vector=vec,
    limit=10,
    query_filter=Filter(must=[FieldCondition(key="tenant_id", match=MatchValue(value="B"))]),
)
leaked = [h for h in hits if h.payload.get("tenant_id") != "B"]
print("CROSS-TENANT LEAK" if leaked else "isolation OK", "->", len(leaked), "foreign rows")

# Critical test: repeat WITHOUT the filter to confirm the server, not the client,
# enforces isolation. If unfiltered queries return tenant A data, isolation is client-side only.
hits_nofilter = client.search(collection_name="docs", query_vector=vec, limit=10)
print("server-side isolation FAILS" if any(h.payload.get("tenant_id") != "B" for h in hits_nofilter) else "OK")
```

### Step 5: Knowledge-base poisoning (test collection only)

Insert a benign poisoned chunk crafted to be retrieved for many unrelated queries, then measure how often it appears in `top_k`.

```python
from qdrant_client.models import PointStruct

# Benign marker payload (no real injection) to measure retrieval dominance.
poison = "POISON-CANARY. " + " ".join(
    ["password reset billing refund account login support error help"] * 8
)
client.upsert("docs_test", points=[
    PointStruct(id=999999, vector=model.encode([poison])[0].tolist(),
                payload={"tenant_id": "B", "source": "poison-test"})
])

queries = ["how do I get a refund", "reset my password", "what is the weather"]
for q in queries:
    hits = client.search("docs_test", model.encode([q])[0].tolist(), limit=5)
    dominated = any(h.payload.get("source") == "poison-test" for h in hits)
    print(f"{'POISON in top5' if dominated else 'clean'}: {q}")
```

### Step 6: Detect indirect prompt injection in retrieved chunks

Scan retrieved chunk text for injection markers before it is concatenated into the prompt.

```python
import re
INJECTION_PATTERNS = [
    r"ignore (all|previous|the above) instructions",
    r"system prompt", r"you are now", r"disregard", r"</?(system|instructions)>",
]
def chunk_is_injection(text):
    low = text.lower()
    return [p for p in INJECTION_PATTERNS if re.search(p, low)]

for hit in client.search("docs", model.encode(["help"])[0].tolist(), limit=10):
    flags = chunk_is_injection(hit.payload.get("text", ""))
    if flags:
        print("INDIRECT INJECTION in chunk", hit.id, flags)
```

### Step 7: Report and remediate

- **Inversion/membership:** rate-limit and authenticate the embedding endpoint; avoid returning raw similarity scores; restrict who can query embeddings.
- **Cross-tenant:** enforce tenant filters server-side (separate collections/namespaces per tenant where feasible); never rely on client-supplied filters.
- **Poisoning:** validate and provenance-tag every ingested chunk; scan inputs for injection; cap any single source's share of retrieval.
- **Indirect injection:** sanitize retrieved chunks and apply output guardrails (see `defending-llms-with-guardrails`).

## Tools and Resources

| Tool | Purpose | Primary Source |
|------|---------|----------------|
| OWASP LLM08 | Vector and Embedding Weaknesses guidance | https://genai.owasp.org/llmrisk/llm082025-vector-and-embedding-weaknesses/ |
| sentence-transformers | Embedding generation for testing | https://www.sbert.net/ |
| Qdrant client | Vector store + filtered search | https://qdrant.tech/documentation/ |
| Chroma / Weaviate / Pinecone | Alternative vector stores | https://docs.trychroma.com/ |
| vec2text | Embedding-inversion research baseline | https://github.com/jxmorris12/vec2text |
| MITRE ATLAS | AML.T0024 Exfiltration via ML Inference API | https://atlas.mitre.org/ |

## Validation Criteria

- [ ] RAG pipeline inventoried (embedding model, store, tenancy, metric, top_k, filters).
- [ ] Embedding-inversion exposure measured and rated.
- [ ] Membership-inference delta computed for in-corpus vs control items.
- [ ] Multi-tenant isolation tested both with and without client filters (server-side enforcement confirmed).
- [ ] Poisoning dominance measured in a test collection only.
- [ ] Retrieved chunks scanned for indirect-injection content.
- [ ] Findings reported with remediation for each weakness class.
- [ ] No production corpus modified during the assessment.

## references

```

```

## references/api-reference.md

# API and Command Reference

## sentence-transformers (embedding generation)
| Call | Purpose |
|------|---------|
| `SentenceTransformer("all-MiniLM-L6-v2")` | Load an embedding model (384-dim) |
| `model.encode([texts])` | Return numpy array of embeddings |
| `model.encode(text, normalize_embeddings=True)` | L2-normalized vectors (for cosine) |

## scikit-learn similarity
| Call | Purpose |
|------|---------|
| `cosine_similarity(a, b)` | Pairwise cosine similarity matrix |

## Qdrant client (qdrant-client)
| Call | Purpose |
|------|---------|
| `QdrantClient(url="http://localhost:6333")` | Connect |
| `client.get_collection(name)` | Inspect vector size + distance metric |
| `client.count(name)` | Corpus size |
| `client.search(collection_name, query_vector, limit, query_filter)` | k-NN search with optional filter |
| `client.upsert(name, points=[PointStruct(id, vector, payload)])` | Insert/update points |
| `Filter(must=[FieldCondition(key, match=MatchValue(value))])` | Metadata filter (tenant isolation) |

## Chroma (chromadb)
| Call | Purpose |
|------|---------|
| `chromadb.Client()` / `PersistentClient(path)` | Connect |
| `collection.query(query_embeddings=[...], n_results=k, where={...})` | k-NN with metadata filter |
| `collection.add(ids, embeddings, metadatas, documents)` | Insert |

## Pinecone (pinecone-client)
| Call | Purpose |
|------|---------|
| `Pinecone(api_key=...)` | Connect |
| `index.query(vector=..., top_k=k, namespace="tenant", filter={...})` | k-NN; namespace = tenant boundary |
| `index.upsert(vectors=[(id, vec, meta)], namespace=...)` | Insert |

## Assessment metrics
| Metric | Meaning |
|--------|---------|
| Inversion cosine | Similarity between reconstructed candidate and target vector; high = recoverable. |
| Membership delta | top-1 score(in-corpus query) − top-1 score(control query); large positive = membership leak. |
| Poison dominance | Fraction of unrelated queries returning the poison chunk in top_k. |
| Cross-tenant count | Number of foreign-tenant rows returned to a tenant query (should be 0). |

## vec2text (research baseline)
| Call | Purpose |
|------|---------|
| `vec2text.load_pretrained_corrector("gtr-base")` | Load inversion corrector for compatible embedder |
| `vec2text.invert_embeddings(embeddings, corrector)` | Reconstruct text from embeddings |

## references/standards.md

# Standards and Framework Mapping

## NIST AI Risk Management Framework (AI RMF 1.0 / GenAI Profile NIST AI 600-1)

| ID | Name | Rationale |
|----|------|-----------|
| MEASURE-2.7 | AI system security and resilience are evaluated and documented | Assessing inversion, membership, isolation, and poisoning weaknesses measures the security/resilience of the RAG vector layer. |

## MITRE ATLAS

| ID | Name | Rationale |
|----|------|-----------|
| AML.T0024 | Exfiltration via ML Inference API | Parent technique: query/embedding access is abused to exfiltrate source data. |
| AML.T0024.000 | Infer Training Data Membership | Membership-inference probe determines whether a record is in the corpus. |
| AML.T0024.001 | Invert ML Model | Embedding inversion reconstructs source text from vectors. |
| AML.T0020 | Poison Training Data | Knowledge-base poisoning inserts adversarial chunks into the corpus. |
| AML.T0051.001 | LLM Prompt Injection: Indirect | Injection payloads surviving in retrieved chunks. |

## OWASP Top 10 for LLM Applications (2025)

| ID | Name | Rationale |
|----|------|-----------|
| LLM08 | Vector and Embedding Weaknesses | The core risk class under test (inversion, leakage, poisoning). |
| LLM02 | Sensitive Information Disclosure | Inversion/membership leakage discloses sensitive source data. |
| LLM01 | Prompt Injection | Indirect injection delivered through poisoned retrieval. |

## Weakness class to control mapping

| Weakness | Control |
|----------|---------|
| Embedding inversion | Authenticate + rate-limit embedding endpoint; avoid exposing raw scores. |
| Membership inference | Restrict similarity-score exposure; add query auditing. |
| Cross-tenant leakage | Server-side tenant filters or per-tenant collections/namespaces. |
| Knowledge-base poisoning | Provenance tagging, content validation, per-source retrieval caps. |
| Indirect injection in chunks | Sanitize retrieved text; apply output guardrails. |

## scripts

```

```

## scripts/agent.py

```python
#!/usr/bin/env python3
"""Vector/embedding weakness assessor.

Runs four checks against a RAG pipeline you are authorized to test:
  inversion  - measures how close a guessed reconstruction sits to a target vector
  membership - computes in-corpus vs control top-1 similarity delta (Qdrant)
  isolation  - verifies server-side tenant filtering (Qdrant)
  injection  - scans retrieved chunk text for indirect prompt-injection markers

Examples
--------
    python agent.py inversion --secret "MRN 553120 hypertension" \
        --guess "patient MRN hypertension diagnosis"
    python agent.py membership --url http://localhost:6333 --collection docs \
        --quote "<exact chunk quote>" --control "unrelated sentence"
    python agent.py isolation --url http://localhost:6333 --collection docs \
        --tenant-field tenant_id --tenant B --query "salary information"
"""
import argparse
import re
import sys

INJECTION_PATTERNS = [
    r"ignore (all|previous|the above) instructions",
    r"system prompt", r"you are now", r"disregard previous",
    r"</?(system|instructions)>",
]


def get_embedder(model_name):
    try:
        from sentence_transformers import SentenceTransformer
    except ImportError:
        sys.exit("[!] sentence-transformers not installed. pip install sentence-transformers")
    return SentenceTransformer(model_name)


def get_qdrant(url):
    try:
        from qdrant_client import QdrantClient
    except ImportError:
        sys.exit("[!] qdrant-client not installed. pip install qdrant-client")
    return QdrantClient(url=url)


def cmd_inversion(args):
    from sklearn.metrics.pairwise import cosine_similarity
    model = get_embedder(args.model)
    tv = model.encode([args.secret])
    gv = model.encode([args.guess])
    sim = float(cosine_similarity(tv, gv)[0][0])
    print(f"inversion cosine(secret, guess) = {sim:.4f}")
    verdict = "HIGH inversion risk" if sim >= 0.85 else "moderate" if sim >= 0.6 else "low"
    print(f"verdict: {verdict}")
    return 1 if sim >= 0.85 else 0


def cmd_membership(args):
    model = get_embedder(args.model)
    client = get_qdrant(args.url)

    def top1(text):
        vec = model.encode([text])[0].tolist()
        hits = client.search(collection_name=args.collection, query_vector=vec, limit=1)
        return hits[0].score if hits else 0.0

    in_corpus = top1(args.quote)
    control = top1(args.control)
    delta = in_corpus - control
    print(f"in-corpus top1 = {in_corpus:.4f}")
    print(f"control  top1 = {control:.4f}")
    print(f"delta         = {delta:.4f}")
    leak = delta >= 0.2
    print("verdict:", "MEMBERSHIP LEAK" if leak else "no clear membership signal")
    return 1 if leak else 0


def cmd_isolation(args):
    from qdrant_client.models import Filter, FieldCondition, MatchValue
    model = get_embedder(args.model)
    client = get_qdrant(args.url)
    vec = model.encode([args.query])[0].tolist()

    filtered = client.search(
        collection_name=args.collection, query_vector=vec, limit=10,
        query_filter=Filter(must=[FieldCondition(
            key=args.tenant_field, match=MatchValue(value=args.tenant))]),
    )
    leak_filtered = [h for h in filtered
                     if h.payload.get(args.tenant_field) != args.tenant]

    unfiltered = client.search(collection_name=args.collection, query_vector=vec, limit=10)
    leak_unfiltered = [h for h in unfiltered
                       if h.payload.get(args.tenant_field) != args.tenant]

    print(f"filtered query foreign rows   : {len(leak_filtered)}")
    print(f"unfiltered query foreign rows : {len(leak_unfiltered)}")
    if leak_filtered:
        print("CRITICAL: filtered query returned other tenants' data")
    elif leak_unfiltered:
        print("WARNING: isolation appears client-side only; server returns cross-tenant data")
    else:
        print("isolation OK at server level for this query")
    return 1 if (leak_filtered or leak_unfiltered) else 0


def cmd_injection(args):
    model = get_embedder(args.model)
    client = get_qdrant(args.url)
    vec = model.encode([args.query])[0].tolist()
    hits = client.search(collection_name=args.collection, query_vector=vec, limit=args.limit)
    found = 0
    for h in hits:
        text = (h.payload or {}).get("text", "")
        flags = [p for p in INJECTION_PATTERNS if re.search(p, text.lower())]
        if flags:
            found += 1
            print(f"INJECTION in chunk {h.id}: {flags}")
    print(f"{found} of {len(hits)} retrieved chunks flagged for indirect injection")
    return 1 if found else 0


def main():
    p = argparse.ArgumentParser(description="Vector/embedding weakness assessor")
    p.add_argument("--model", default="all-MiniLM-L6-v2", help="embedding model")
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("inversion"); s.add_argument("--secret", required=True); s.add_argument("--guess", required=True)
    s = sub.add_parser("membership")
    s.add_argument("--url", required=True); s.add_argument("--collection", required=True)
    s.add_argument("--quote", required=True); s.add_argument("--control", required=True)
    s = sub.add_parser("isolation")
    s.add_argument("--url", required=True); s.add_argument("--collection", required=True)
    s.add_argument("--tenant-field", default="tenant_id"); s.add_argument("--tenant", required=True)
    s.add_argument("--query", required=True)
    s = sub.add_parser("injection")
    s.add_argument("--url", required=True); s.add_argument("--collection", required=True)
    s.add_argument("--query", default="help"); s.add_argument("--limit", type=int, default=10)

    args = p.parse_args()
    dispatch = {
        "inversion": cmd_inversion, "membership": cmd_membership,
        "isolation": cmd_isolation, "injection": cmd_injection,
    }
    try:
        sys.exit(dispatch[args.cmd](args))
    except Exception as exc:
        sys.exit(f"[!] {args.cmd} failed: {exc}")


if __name__ == "__main__":
    main()
```