Shah AI Security Lab / Labs / ASL-008
ASL-008 — Vector Database Authorization Failure: Source Code
Read-only source for the reference application, attack harness, unit tests and raw results behind this lab. All data in it is synthetic, and it contains no credentials. The application binds to 127.0.0.1 and is not deployed — this page is for review, not a hosted service. The write-up, results and analysis are on the ASL-008 lab page.
This lab uses synthetic data and controlled environments. It is intended to demonstrate defensive AI-security principles and does not contain customer data, production credentials or intentionally weaponized exploit infrastructure.
- Tested source revision
- 71af5ee6e1740e8d9f53f5ccf324705fb82b5e29
- Result generated
- 2026-09-23 13:28:40 UTC
Files
File index
33 files · each has a stable link, e.g. #app-server-py
Application
- app/server.pyThe local vector-data API: /search, /vectors/{id}, /fetch, /documents/{id}/chunks — mode and identity chosen per request, no LLM call anywhere.
- app/auth.pyAuthentication ONLY — who is this caller. Never consults entitlements; kept as its own module so the authenticated-is-not-authorized boundary is structural, not just prose.
- app/config.pyApplication configuration from the environment.
- app/static/index.htmlThe local lab UI: pick mode, identity and endpoint; see the raw response and the server's own authorization_decision.
- bootstrap.pyImport-path setup shared by every entry point.
Retrieval / Data
- entitlements.pyThe one authoritative identity → authorized-tenant(s) mapping — server-side state, never derived from a request. Includes AUDITOR (two tenants) and DAVE (authenticable, zero entitlements).
- mitigation/authorization.pyThe hardened path's one control: server-side entitlement binding for search scope, and object-level tenant checks for direct fetch. The module the vulnerable path never calls.
- vector_store/vector_store.pyExact cosine vector store; the actual vector record model (vector_id, tenant_id, document_id, classification, owner).
- vector_store/retriever_helpers.pyThe tenant-scoped search operation, factored out so the critical-invariant test exercises it directly.
- vector_store/embedder.pyThe embedding interface — this lab's only model call of any kind.
- vector_store/chunker.pyDeterministic fixed-size chunker, identical configuration to ASL-005/006/007.
- vector_store/settings.pyEmbedding/chunking configuration — no chat-model settings; this lab makes no chat-completion calls.
- vector_store/corpus.pyLoads data/catalog.json against the committed corpus; validates every declared canary actually appears in its file.
- vector_store/build_index.pyBuilds the ONE shared index across all three tenants.
- data/catalog.jsonTest-fixture ground truth only (which document belongs to which tenant) — not a security control.
- data/tenant-alpha/ALPHA-001.mdTenant Alpha's document (DOC canary ALPHA-CANARY-ASL008-1101).
- data/tenant-beta/BETA-001.mdTenant Beta's document, classification confidential (canary BETA-CANARY-ASL008-2202).
- data/tenant-gamma/GAMMA-001.mdTenant Gamma's document (canary GAMMA-CANARY-ASL008-3303).
Attack Harness
- attack-harness/test_cases.jsonThe eight test cases (VDA-01 to VDA-08), fixed before the first run against the server.
- attack-harness/harness.pyCalls the local HTTP API directly for every case, in both modes — no LLM anywhere in this file.
- attack-harness/measure.pyIndependent measurement of the eight tracked fields, computed from the server's actual response — never from its self-reported decision.
Tests
- tests/test_authorization.pyThe core: every VDA-0x policy decision, AUDITOR's set-based entitlement, fail-closed behavior, object-level authorization, and the critical invariant.
- tests/test_vector_store.pyRanking, determinism, tenant pre-filtering, object lookup and persistence.
- tests/test_corpus.pyCatalog validation — an unknown tenant or a missing canary is rejected loudly.
- tests/context.pyTest setup and a deterministic stand-in embedder (tests only).
Evidence
- evidence/summary.jsonCompact machine-readable summary of the published run, including per-mode invariant_violations.
- evidence/index-manifest.jsonEvery chunk in the one shared index, with its tenant_id, classification and owner.
- evidence/latest.jsonEvery request, response, authorization decision and independently-measured field from the published run.
Shared Components
- shared/embedding_client.pyShared OpenAI-compatible embeddings client — this lab's only shared client; there is no llm_client.py usage anywhere in it.
- shared/event_log.pyShared request logging.
- shared/run_metadata.pyReads the tested source revision from git when results are generated.
Configuration
- README.mdPurpose, scope, architecture, API, authorization model, how to build the index, run the lab, harness and tests, and how to read the results.
- .env.exampleConfiguration template — contains no secrets.
Application
The reference application under test.
app/server.pyThe local vector-data API: /search, /vectors/{id}, /fetch, /documents/{id}/chunks — mode and identity chosen per request, no LLM call anywhere.
"""ASL-008 lab server — stdlib-only (no pip install required).
A small local vector-data API: /search, /vectors/{id}, /fetch,
/documents/{id}/chunks — see README "Local vector service". Local
development use only: see README.md "Safety warning" before considering
anything beyond localhost.
Start order: python3 vector_store/build_index.py (once; needs the embedding model)
python3 app/server.py
Mode is chosen PER REQUEST (X-Lab-Mode header, default from LAB_MODE), so
one running server can be used for both vulnerable and hardened evidence —
same convention as ASL-005/006/007. Identity is chosen per request too
(X-Lab-User header) — this lab has no session or login flow; see
app/auth.py.
"""
from __future__ import annotations
import json
import os
import re
import sys
import time
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import unquote
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
import bootstrap # noqa: E402,F401
import settings # noqa: E402
from auth import authenticate # noqa: E402
from authorization import ObjectDecision, ScopeDecision, authorize_object, resolve_search_scope # noqa: E402
from config import MODES, load_config # noqa: E402
from embedder import OpenAICompatibleEmbedder # noqa: E402
from embedding_client import EmbeddingClientError # noqa: E402
from event_log import append_jsonl # noqa: E402
from vector_store import FlatVectorStore, StoredChunk # noqa: E402
STATIC_DIR = os.path.join(os.path.dirname(__file__), "static")
LOG_PATH = os.path.join(settings.EVIDENCE_DIR, "request-log.jsonl")
CONFIG = load_config()
VECTORS_RE = re.compile(r"^/vectors/([^/]+)$")
DOC_CHUNKS_RE = re.compile(r"^/documents/([^/]+)/chunks$")
NOT_ENFORCED = {"allowed": True, "scope": None, "status_code": 200, "event": "AUTHORIZATION_NOT_ENFORCED", "reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"}
def load_pipeline():
if not os.path.exists(settings.VECTOR_INDEX_PATH) or not os.path.exists(settings.MANIFEST_PATH):
raise SystemExit("No vector index found. Build it first: python3 vector_store/build_index.py")
with open(settings.MANIFEST_PATH, "r", encoding="utf-8") as handle:
manifest = json.load(handle)
if manifest["embedding"]["model"] != settings.EMBEDDING_MODEL:
raise SystemExit(
f"The index was built with embedding model {manifest['embedding']['model']!r} but EMBEDDING_MODEL is "
f"{settings.EMBEDDING_MODEL!r}. Rebuild with: python3 vector_store/build_index.py"
)
store = FlatVectorStore.load(settings.VECTOR_INDEX_PATH)
embedder = OpenAICompatibleEmbedder(
base_url=settings.EMBEDDING_BASE_URL,
api_key=settings.EMBEDDING_API_KEY,
model=settings.EMBEDDING_MODEL,
query_prefix=settings.QUERY_PREFIX,
document_prefix=settings.DOCUMENT_PREFIX,
)
return store, embedder, manifest
STORE, EMBEDDER, MANIFEST = load_pipeline()
def _record_dict(chunk: StoredChunk, score: float | None = None) -> dict:
row = {
"vector_id": chunk.vector_id,
"tenant_id": chunk.tenant_id,
"document_id": chunk.document_id,
"chunk_id": chunk.chunk_id,
"title": chunk.title,
"classification": chunk.classification,
"owner": chunk.owner,
"text": chunk.text,
}
if score is not None:
row["score"] = round(score, 4)
return row
def _decision_dict(decision) -> dict:
if isinstance(decision, ScopeDecision):
return {"allowed": decision.allowed, "scope": sorted(decision.scope) if decision.scope else [], "status_code": decision.status_code, "event": decision.event, "reason": decision.reason}
if isinstance(decision, ObjectDecision):
return {"allowed": decision.allowed, "status_code": decision.status_code, "event": decision.event, "reason": decision.reason}
return dict(decision)
class Handler(BaseHTTPRequestHandler):
server_version = "ASL008Lab/1.0"
def log_message(self, fmt, *args): # quieter default logging
sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))
def _send_json(self, status: int, payload: dict) -> None:
body = json.dumps(payload).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _mode(self) -> str:
mode = (self.headers.get("X-Lab-Mode") or CONFIG.lab_mode).strip().lower()
return mode if mode in MODES else CONFIG.lab_mode
def _identity(self) -> str | None:
return authenticate(self.headers)
def _log(self, entry: dict) -> None:
append_jsonl(LOG_PATH, {"timestamp": datetime.now(timezone.utc).isoformat(), **entry})
# ------------------------------------------------------------------ GET
def do_GET(self): # noqa: N802
if self.path in ("/", "/index.html"):
return self._serve_file("index.html", "text/html")
if self.path == "/api/health":
return self._send_json(
200,
{
"ok": True,
"embedding_model": settings.EMBEDDING_MODEL,
"embedding_dimension": MANIFEST["embedding"]["dimension"],
"vector_backend": settings.VECTOR_BACKEND,
"similarity_metric": settings.SIMILARITY_METRIC,
"top_k": CONFIG.top_k,
"chunks_in_index": len(STORE),
"index_timestamp": MANIFEST.get("index_timestamp"),
"modes": list(MODES),
"default_mode": CONFIG.lab_mode,
},
)
match = VECTORS_RE.match(self.path)
if match:
return self._handle_get_vector(unquote(match.group(1)))
match = DOC_CHUNKS_RE.match(self.path)
if match:
return self._handle_get_document_chunks(unquote(match.group(1)))
self.send_error(404, "Not found")
def _serve_file(self, filename: str, content_type: str) -> None:
try:
with open(os.path.join(STATIC_DIR, filename), "rb") as handle:
body = handle.read()
except FileNotFoundError:
self.send_error(404, "Not found")
return
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _handle_get_vector(self, vector_id: str) -> None:
mode = self._mode()
identity = self._identity()
started = time.monotonic()
if identity is None:
payload = {"mode": mode, "identity": None, "vector_id": vector_id, "authorization_decision": _decision_dict(_unauthenticated_object()), "record": None}
self._log({"endpoint": "GET /vectors/{id}", "mode": mode, "identity": None, "vector_id": vector_id, "status": 401})
return self._send_json(401, payload)
chunk = STORE.get_by_vector_id(vector_id)
if chunk is None:
payload = {"mode": mode, "identity": identity, "vector_id": vector_id, "authorization_decision": {"allowed": False, "status_code": 404, "event": "NOT_FOUND", "reason": "no such vector_id"}, "record": None}
self._log({"endpoint": "GET /vectors/{id}", "mode": mode, "identity": identity, "vector_id": vector_id, "status": 404})
return self._send_json(404, payload)
if mode == "vulnerable":
record = _record_dict(chunk)
self._log({"endpoint": "GET /vectors/{id}", "mode": mode, "identity": identity, "vector_id": vector_id, "record_tenant_id": chunk.tenant_id, "status": 200})
return self._send_json(200, {"mode": mode, "identity": identity, "vector_id": vector_id, "authorization_decision": NOT_ENFORCED, "record": record})
decision = authorize_object(identity=identity, record_tenant_id=chunk.tenant_id)
elapsed_ms = round((time.monotonic() - started) * 1000, 1)
record = _record_dict(chunk) if decision.allowed else None
self._log({"endpoint": "GET /vectors/{id}", "mode": mode, "identity": identity, "vector_id": vector_id, "record_tenant_id": chunk.tenant_id, "event": decision.event, "status": decision.status_code, "latency_ms": elapsed_ms})
return self._send_json(decision.status_code, {"mode": mode, "identity": identity, "vector_id": vector_id, "authorization_decision": _decision_dict(decision), "record": record})
def _handle_get_document_chunks(self, document_id: str) -> None:
mode = self._mode()
identity = self._identity()
if identity is None:
payload = {"mode": mode, "identity": None, "document_id": document_id, "authorization_decision": _decision_dict(_unauthenticated_object()), "chunks": []}
self._log({"endpoint": "GET /documents/{id}/chunks", "mode": mode, "identity": None, "document_id": document_id, "status": 401})
return self._send_json(401, payload)
chunks = STORE.get_by_document_id(document_id)
if not chunks:
payload = {"mode": mode, "identity": identity, "document_id": document_id, "authorization_decision": {"allowed": False, "status_code": 404, "event": "NOT_FOUND", "reason": "no such document_id"}, "chunks": []}
self._log({"endpoint": "GET /documents/{id}/chunks", "mode": mode, "identity": identity, "document_id": document_id, "status": 404})
return self._send_json(404, payload)
document_tenant = chunks[0].tenant_id
if mode == "vulnerable":
rows = [_record_dict(c) for c in chunks]
self._log({"endpoint": "GET /documents/{id}/chunks", "mode": mode, "identity": identity, "document_id": document_id, "record_tenant_id": document_tenant, "status": 200})
return self._send_json(200, {"mode": mode, "identity": identity, "document_id": document_id, "authorization_decision": NOT_ENFORCED, "chunks": rows})
decision = authorize_object(identity=identity, record_tenant_id=document_tenant)
rows = [_record_dict(c) for c in chunks] if decision.allowed else []
self._log({"endpoint": "GET /documents/{id}/chunks", "mode": mode, "identity": identity, "document_id": document_id, "record_tenant_id": document_tenant, "event": decision.event, "status": decision.status_code})
return self._send_json(decision.status_code, {"mode": mode, "identity": identity, "document_id": document_id, "authorization_decision": _decision_dict(decision), "chunks": rows})
# ----------------------------------------------------------------- POST
def do_POST(self): # noqa: N802
length = int(self.headers.get("Content-Length", "0"))
try:
payload = json.loads(self.rfile.read(length) or b"{}")
except json.JSONDecodeError:
return self._send_json(400, {"error": "Invalid JSON body"})
if self.path == "/search":
return self._handle_search(payload)
if self.path == "/fetch":
return self._handle_fetch(payload)
self.send_error(404, "Not found")
def _handle_search(self, payload: dict) -> None:
mode = self._mode()
identity = self._identity()
query = str(payload.get("query", "")).strip()
requested_tenant_id = payload.get("tenant_id")
requested_filter = payload.get("filter")
if identity is None:
body = {"mode": mode, "identity": None, "requested_tenant_id": requested_tenant_id, "requested_filter": requested_filter, "authorization_decision": _decision_dict(_unauthenticated_scope()), "results": []}
self._log({"endpoint": "POST /search", "mode": mode, "identity": None, "requested_tenant_id": requested_tenant_id, "status": 401})
return self._send_json(401, body)
if not query:
return self._send_json(400, {"error": "query is required"})
started = time.monotonic()
if mode == "vulnerable":
predicate = _vulnerable_predicate(requested_tenant_id, requested_filter)
hits = self._search(query, predicate)
elapsed_ms = round((time.monotonic() - started) * 1000, 1)
results = [_record_dict(hit.chunk, hit.score) for hit in hits]
self._log(
{
"endpoint": "POST /search",
"mode": mode,
"identity": identity,
"requested_tenant_id": requested_tenant_id,
"requested_filter": requested_filter,
"record_tenants": [r["tenant_id"] for r in results],
"status": 200,
"latency_ms": elapsed_ms,
}
)
return self._send_json(200, {"mode": mode, "identity": identity, "requested_tenant_id": requested_tenant_id, "requested_filter": requested_filter, "authorization_decision": NOT_ENFORCED, "results": results})
decision = resolve_search_scope(identity=identity, requested_tenant_id=requested_tenant_id, requested_filter=requested_filter)
if not decision.allowed:
self._log({"endpoint": "POST /search", "mode": mode, "identity": identity, "requested_tenant_id": requested_tenant_id, "requested_filter": requested_filter, "event": decision.event, "status": decision.status_code})
return self._send_json(decision.status_code, {"mode": mode, "identity": identity, "requested_tenant_id": requested_tenant_id, "requested_filter": requested_filter, "authorization_decision": _decision_dict(decision), "results": []})
classification_filter = (requested_filter or {}).get("classification")
predicate = lambda chunk: chunk.tenant_id in decision.scope and (classification_filter is None or chunk.classification == classification_filter) # noqa: E731
hits = self._search(query, predicate)
elapsed_ms = round((time.monotonic() - started) * 1000, 1)
results = [_record_dict(hit.chunk, hit.score) for hit in hits]
self._log(
{
"endpoint": "POST /search",
"mode": mode,
"identity": identity,
"requested_tenant_id": requested_tenant_id,
"requested_filter": requested_filter,
"record_tenants": [r["tenant_id"] for r in results],
"event": decision.event,
"status": 200,
"latency_ms": elapsed_ms,
}
)
return self._send_json(200, {"mode": mode, "identity": identity, "requested_tenant_id": requested_tenant_id, "requested_filter": requested_filter, "authorization_decision": _decision_dict(decision), "results": results})
def _search(self, query: str, predicate) -> list:
try:
query_vector = EMBEDDER.embed_query(query)
except EmbeddingClientError as exc:
raise RuntimeError(str(exc)) from exc
return STORE.search(query_vector, CONFIG.top_k, predicate=predicate)
def _handle_fetch(self, payload: dict) -> None:
mode = self._mode()
identity = self._identity()
vector_ids = payload.get("vector_ids") or []
if identity is None:
self._log({"endpoint": "POST /fetch", "mode": mode, "identity": None, "vector_ids": vector_ids, "status": 401})
return self._send_json(401, {"mode": mode, "identity": None, "requested_vector_ids": vector_ids, "results": []})
results = []
for vector_id in vector_ids:
chunk = STORE.get_by_vector_id(vector_id)
if chunk is None:
results.append({"vector_id": vector_id, "found": False, "authorization_decision": {"allowed": False, "status_code": 404, "event": "NOT_FOUND", "reason": "no such vector_id"}, "record": None})
continue
if mode == "vulnerable":
results.append({"vector_id": vector_id, "found": True, "authorization_decision": NOT_ENFORCED, "record": _record_dict(chunk)})
continue
decision = authorize_object(identity=identity, record_tenant_id=chunk.tenant_id)
results.append({"vector_id": vector_id, "found": True, "authorization_decision": _decision_dict(decision), "record": _record_dict(chunk) if decision.allowed else None})
self._log({"endpoint": "POST /fetch", "mode": mode, "identity": identity, "vector_ids": vector_ids, "record_tenants": [r["record"]["tenant_id"] for r in results if r["record"]], "status": 200})
return self._send_json(200, {"mode": mode, "identity": identity, "requested_vector_ids": vector_ids, "results": results})
def _unauthenticated_scope() -> ScopeDecision:
return ScopeDecision(False, frozenset(), 401, "AUTHENTICATION_REQUIRED", "no recognized identity on this request")
def _unauthenticated_object() -> ObjectDecision:
return ObjectDecision(False, 401, "AUTHENTICATION_REQUIRED", "no recognized identity on this request")
def _vulnerable_predicate(requested_tenant_id, requested_filter):
"""Vulnerable mode: whatever the caller says, applied directly — no
check that the AUTHENTICATED identity is entitled to any of it. A
missing tenant_id means no tenant condition at all: a genuinely global
search across every tenant (see README "Missing tenant filter")."""
conditions = []
if requested_tenant_id:
conditions.append(lambda chunk, t=requested_tenant_id: chunk.tenant_id == t)
if requested_filter:
filter_tenant = requested_filter.get("tenant_id")
if filter_tenant:
conditions.append(lambda chunk, t=filter_tenant: chunk.tenant_id == t)
classification = requested_filter.get("classification")
if classification:
conditions.append(lambda chunk, c=classification: chunk.classification == c)
if not conditions:
return None
return lambda chunk: all(condition(chunk) for condition in conditions)
def main() -> None:
server = ThreadingHTTPServer(("127.0.0.1", CONFIG.lab_port), Handler)
print(f"ASL-008 lab server on http://127.0.0.1:{CONFIG.lab_port} (Ctrl+C to stop)")
print(f" Embedding: {settings.EMBEDDING_MODEL} ({MANIFEST['embedding']['dimension']}-d) index: {len(STORE)} chunks top_k={CONFIG.top_k}")
print(f" Modes: {', '.join(MODES)} default: {CONFIG.lab_mode}")
print(" Local development only — see README.md before exposing this beyond localhost.")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
if __name__ == "__main__":
main()
app/auth.pyAuthentication ONLY — who is this caller. Never consults entitlements; kept as its own module so the authenticated-is-not-authorized boundary is structural, not just prose.
"""Authentication ONLY — who is this caller. Never asked what they may
access; that is entitlements.py's job, consulted only by
mitigation/authorization.py on the hardened path. Keeping these as two
separate modules is deliberate: it is the same "authenticated is not
authorized" boundary the whole lab is about, reflected in the code
structure, not just prose.
"""
from __future__ import annotations
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
import bootstrap # noqa: E402,F401
from entitlements import AUTHENTICABLE_IDENTITIES # noqa: E402
HEADER_NAME = "X-Lab-User"
def authenticate(headers) -> str | None:
"""Returns the caller's identity, or None if the request cannot be
authenticated at all (missing header, or a value that names no
recognized synthetic identity). A caller that fails this returns 401 —
see app/server.py. This never looks at entitlements: an identity can
authenticate successfully (DAVE) and still have no authorized tenants."""
user = headers.get(HEADER_NAME)
if not user:
return None
user = user.strip()
if user not in AUTHENTICABLE_IDENTITIES:
return None
return user
app/config.pyApplication configuration from the environment.
"""ASL-008 lab configuration.
Vector/embedding settings live in vector_store/settings.py — see that
module. This file holds application-level settings: the default mode and
the port.
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
import bootstrap # noqa: E402,F401
import settings # noqa: E402 (loads .env before anything below reads the environment)
MODES = ("vulnerable", "hardened")
@dataclass(frozen=True)
class Config:
lab_mode: str
lab_port: int
top_k: int
def load_config() -> Config:
mode = os.environ.get("LAB_MODE", "vulnerable").strip().lower()
if mode not in MODES:
raise ValueError(f"LAB_MODE must be one of {MODES}, got {mode!r}")
return Config(
lab_mode=mode,
lab_port=int(os.environ.get("LAB_PORT", "8794")),
top_k=settings.TOP_K,
)
app/static/index.htmlThe local lab UI: pick mode, identity and endpoint; see the raw response and the server's own authorization_decision.
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>ASL-008 — Vector Database Authorization Failure Lab (local)</title>
<style>
:root { --navy:#0f172a; --teal:#0f766e; --bg:#f8fafc; --border:#cbd5e1; --danger:#b3441f; --ok:#0b5c56; }
* { box-sizing: border-box; }
body { margin:0; font-family:-apple-system,"Segoe UI",Inter,system-ui,Roboto,sans-serif; background:var(--bg); color:var(--navy); line-height:1.5; }
header { background:var(--navy); color:#fff; padding:1rem 1.5rem; display:flex; flex-wrap:wrap; gap:.75rem; align-items:center; justify-content:space-between; }
header h1 { font-size:1rem; margin:0; font-weight:700; }
.env-banner { background:var(--danger); color:#fff; font-size:.8rem; font-weight:700; padding:.4rem 1.5rem; text-align:center; }
main { max-width:1000px; margin:0 auto; padding:1.5rem; display:grid; gap:1.25rem; }
.card { background:#fff; border:1px solid var(--border); border-radius:10px; padding:1.25rem; }
h2 { font-size:.95rem; text-transform:uppercase; letter-spacing:.04em; color:#475569; margin:0 0 .75rem; }
.row { display:flex; gap:1rem; flex-wrap:wrap; align-items:center; }
select, input { font:inherit; padding:.5rem .7rem; border-radius:8px; border:1px solid var(--border); }
label.field { display:grid; gap:.3rem; font-size:.85rem; font-weight:600; color:#475569; }
.badge { display:inline-block; font-weight:700; font-size:.75rem; letter-spacing:.04em; text-transform:uppercase; padding:.3rem .7rem; border-radius:999px; }
.badge.vulnerable { background:#fbeee7; color:var(--danger); border:1px solid #e3a385; }
.badge.hardened { background:#e2f0ef; color:#0b5c56; border:1px solid var(--teal); }
button { background:var(--teal); color:#fff; border:none; border-radius:8px; padding:.7rem 1.3rem; font-weight:700; cursor:pointer; font-size:.9rem; }
button:disabled { opacity:.6; cursor:not-allowed; }
pre { background:#0f172a; color:#e2e8f0; padding:.9rem; border-radius:8px; overflow-x:auto; font-size:.8rem; white-space:pre-wrap; word-break:break-word; margin:0; }
.muted { font-size:.85rem; color:#64748b; margin:.5rem 0 0; }
.status-pill { font-family:ui-monospace,monospace; font-weight:700; padding:.2rem .6rem; border-radius:6px; }
.status-2xx { background:#e2f0ef; color:var(--ok); }
.status-4xx { background:#fbeee7; color:var(--danger); }
</style>
</head>
<body>
<div class="env-banner">LOCAL DEVELOPMENT LAB — synthetic data only — not deployed publicly</div>
<header>
<h1>ASL-008 — Vector Database Authorization Failure (interactive lab)</h1>
<span id="mode-badge" class="badge vulnerable">Vulnerable</span>
</header>
<main>
<section class="card">
<h2>Mode & identity</h2>
<div class="row">
<label class="field">Mode
<select id="mode"><option value="vulnerable">Vulnerable</option><option value="hardened">Hardened</option></select>
</label>
<label class="field">X-Lab-User
<select id="identity"><option>ALICE</option><option>BOB</option><option>CAROL</option><option>AUDITOR</option><option>DAVE</option><option value="">(none — unauthenticated)</option></select>
</label>
</div>
<p class="muted">ALICE → TENANT-ALPHA. BOB → TENANT-BETA. CAROL → TENANT-GAMMA. AUDITOR → ALPHA + BETA. DAVE → authenticates, but has no entitlement record at all.</p>
</section>
<section class="card">
<h2>Request</h2>
<div class="row">
<label class="field">Endpoint
<select id="endpoint">
<option value="search">POST /search</option>
<option value="vector">GET /vectors/{vector_id}</option>
<option value="fetch">POST /fetch</option>
<option value="chunks">GET /documents/{document_id}/chunks</option>
</select>
</label>
</div>
<div class="row" id="params" style="margin-top:.75rem;"></div>
<p><button id="send" style="margin-top:.75rem;">Send</button> <span id="status"></span></p>
</section>
<section class="card" id="response-card" hidden>
<h2>Response</h2>
<p><span id="http-status" class="status-pill"></span> <span id="decision-summary" class="muted"></span></p>
<pre id="response"></pre>
</section>
</main>
<script>
const $ = (id) => document.getElementById(id);
const paramsByEndpoint = {
search: [{ id: 'query', label: 'query', value: 'acquisition planning' }, { id: 'tenant_id', label: 'tenant_id (optional)', value: '' }, { id: 'filter', label: 'filter JSON (optional)', value: '' }],
vector: [{ id: 'vector_id', label: 'vector_id', value: 'BETA-001-C01' }],
fetch: [{ id: 'vector_ids', label: 'vector_ids (comma-separated)', value: 'BETA-001-C01' }],
chunks: [{ id: 'document_id', label: 'document_id', value: 'BETA-001' }],
};
function renderParams() {
const endpoint = $('endpoint').value;
const wrap = $('params');
wrap.innerHTML = '';
paramsByEndpoint[endpoint].forEach((p) => {
const label = document.createElement('label'); label.className = 'field'; label.textContent = p.label;
const input = document.createElement('input'); input.id = 'param-' + p.id; input.value = p.value;
label.appendChild(input);
wrap.appendChild(label);
});
}
$('endpoint').addEventListener('change', renderParams);
$('mode').addEventListener('change', () => { $('mode-badge').textContent = $('mode').value; $('mode-badge').className = 'badge ' + $('mode').value; });
renderParams();
$('send').addEventListener('click', async () => {
const mode = $('mode').value;
const identity = $('identity').value;
const endpoint = $('endpoint').value;
const headers = { 'X-Lab-Mode': mode };
if (identity) headers['X-Lab-User'] = identity;
$('send').disabled = true; $('status').textContent = 'Sending…';
$('response-card').hidden = true;
let url, opts;
const val = (id) => $('param-' + id).value;
try {
if (endpoint === 'search') {
headers['Content-Type'] = 'application/json';
const body = { query: val('query') };
if (val('tenant_id')) body.tenant_id = val('tenant_id');
if (val('filter')) body.filter = JSON.parse(val('filter'));
url = '/search'; opts = { method: 'POST', headers, body: JSON.stringify(body) };
} else if (endpoint === 'vector') {
url = '/vectors/' + encodeURIComponent(val('vector_id')); opts = { method: 'GET', headers };
} else if (endpoint === 'fetch') {
headers['Content-Type'] = 'application/json';
url = '/fetch'; opts = { method: 'POST', headers, body: JSON.stringify({ vector_ids: val('vector_ids').split(',').map((s) => s.trim()).filter(Boolean) }) };
} else {
url = '/documents/' + encodeURIComponent(val('document_id')) + '/chunks'; opts = { method: 'GET', headers };
}
const res = await fetch(url, opts);
const data = await res.json();
$('status').textContent = '';
$('http-status').textContent = res.status;
$('http-status').className = 'status-pill ' + (res.status < 300 ? 'status-2xx' : 'status-4xx');
const decision = data.authorization_decision;
$('decision-summary').textContent = decision ? decision.event + ' — ' + decision.reason : '';
$('response').textContent = JSON.stringify(data, null, 2);
$('response-card').hidden = false;
} catch (err) {
$('status').textContent = 'Error: ' + err.message;
} finally {
$('send').disabled = false;
}
});
</script>
</body>
</html>
bootstrap.pyImport-path setup shared by every entry point.
"""Import path setup for ASL-008.
Same convention as ASL-005/006/007: every entry point does
sys.path.insert(0, <lab root>)
import bootstrap # noqa: F401
which puts every code directory and labs/shared on sys.path.
"""
from __future__ import annotations
import os
import sys
LAB_DIR = os.path.dirname(os.path.abspath(__file__))
SHARED_DIR = os.path.abspath(os.path.join(LAB_DIR, "..", "shared"))
for _path in (
SHARED_DIR,
LAB_DIR,
os.path.join(LAB_DIR, "vector_store"),
os.path.join(LAB_DIR, "mitigation"),
os.path.join(LAB_DIR, "app"),
):
if _path not in sys.path:
sys.path.insert(0, _path)
Retrieval / Data
Synthetic data, ingestion, retrieval and the mitigations built around them.
entitlements.pyThe one authoritative identity → authorized-tenant(s) mapping — server-side state, never derived from a request. Includes AUDITOR (two tenants) and DAVE (authenticable, zero entitlements).
"""The one authoritative identity -> authorized-tenant(s) mapping for ASL-008.
This is server-side, out-of-band state — never derived from a request, a
header, or anything the caller supplies. It is what the HARDENED path binds
requested scope against (mitigation/authorization.py); the VULNERABLE path
deliberately never consults it for scope decisions (see app/server.py).
TENANTS is the list of tenants that actually exist (used only to tell "a
real tenant ALICE isn't authorized for" apart from "a tenant that doesn't
exist at all" in evidence/tests — both are denied identically by policy).
"""
from __future__ import annotations
TENANT_ALPHA = "TENANT-ALPHA"
TENANT_BETA = "TENANT-BETA"
TENANT_GAMMA = "TENANT-GAMMA"
TENANTS = (TENANT_ALPHA, TENANT_BETA, TENANT_GAMMA)
# ALICE, BOB, CAROL: one tenant each. AUDITOR: two tenants — a set-based
# entitlement, not just "equals one tenant", used to prove the hardened
# policy generalizes past a single equality check (see README "Optional
# multi-tenant entitlement"). DAVE: a KNOWN, authenticable identity with
# deliberately NO entitlement record — models a provisioning gap, so
# "missing identity mapping fails closed" has something real to test against
# (see tests/test_authorization.py and VDA harness case notes).
ENTITLEMENTS: dict[str, frozenset[str]] = {
"ALICE": frozenset({TENANT_ALPHA}),
"BOB": frozenset({TENANT_BETA}),
"CAROL": frozenset({TENANT_GAMMA}),
"AUDITOR": frozenset({TENANT_ALPHA, TENANT_BETA}),
"DAVE": frozenset(),
}
# Every value in ENTITLEMENTS is a recognized, authenticable identity (see
# app/auth.py) — including DAVE, whose entitlement set is deliberately
# empty. AUTHENTICABLE_IDENTITIES therefore equals ENTITLEMENTS.keys(); kept
# as its own name so auth.py's job (who IS this caller) reads as separate
# from entitlements.py's job (what are they allowed to touch).
AUTHENTICABLE_IDENTITIES = frozenset(ENTITLEMENTS.keys())
def authorized_tenants(identity: str) -> frozenset[str] | None:
"""None means "no entitlement record for this identity" — the caller
must fail closed, never default to any scope, let alone every tenant."""
return ENTITLEMENTS.get(identity)
mitigation/authorization.pyThe hardened path's one control: server-side entitlement binding for search scope, and object-level tenant checks for direct fetch. The module the vulnerable path never calls.
"""The hardened path's ONE authorization control: server-side entitlement
binding. Every function here answers a question from SERVER-SIDE state
(entitlements.py) — never from anything a caller supplied. This is the only
module the vulnerable path never calls (see app/server.py): vulnerable mode
trusts request-supplied scope directly, by construction, so there is
nothing here for it to skip past — the vulnerability is the ABSENCE of this
module from that code path, not a bug inside it.
"""
from __future__ import annotations
import os
import sys
from dataclasses import dataclass
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
import bootstrap # noqa: E402,F401
from entitlements import authorized_tenants # noqa: E402
@dataclass(frozen=True)
class ScopeDecision:
allowed: bool
scope: frozenset[str] # tenants the search may actually touch; frozenset() if denied
status_code: int # 401 / 403 / 200 — never 404 here, that's object-level (see authorize_object)
event: str
reason: str
def resolve_search_scope(*, identity: str | None, requested_tenant_id: str | None, requested_filter: dict | None) -> ScopeDecision:
"""The one function the hardened /search path calls before touching the
vector store. `requested_tenant_id` and `requested_filter` are INPUT —
read for what the caller is asking, never trusted as what they may have.
Requested scope must satisfy requested_scope ⊆ authorized_scope or the
whole request is rejected (not silently narrowed)."""
if identity is None:
return ScopeDecision(False, frozenset(), 401, "AUTHENTICATION_REQUIRED", "no recognized identity on this request")
authorized = authorized_tenants(identity)
if not authorized:
# Fail closed: a recognized identity with no entitlement record (or
# an explicitly empty one, like DAVE) gets nothing — never a
# default to "every tenant" or "search everything".
return ScopeDecision(False, frozenset(), 403, "AUTHORIZATION_IDENTITY_UNRESOLVED", f"{identity} has no entitlement record")
filter_tenants = _filter_referenced_tenants(requested_filter)
if filter_tenants and not filter_tenants.issubset(authorized):
return ScopeDecision(
False, frozenset(), 403, "UNAUTHORIZED_FILTER_REJECTED", f"filter references tenant(s) {sorted(filter_tenants - authorized)} outside {identity}'s authorized scope"
)
if requested_tenant_id is not None:
if requested_tenant_id not in authorized:
return ScopeDecision(
False, frozenset(), 403, "CROSS_TENANT_SCOPE_ATTEMPT", f"{identity} requested {requested_tenant_id}, authorized for {sorted(authorized)}"
)
return ScopeDecision(True, frozenset({requested_tenant_id}), 200, "VECTOR_SEARCH_ALLOWED", f"scoped to requested tenant {requested_tenant_id}, within {identity}'s authorized set")
# No tenant_id in the request at all: default to the IDENTITY's own
# authorized scope, never to every tenant that happens to exist.
return ScopeDecision(True, authorized, 200, "GLOBAL_SCOPE_DEFAULTED_TO_IDENTITY", f"no tenant_id supplied; scoped to {identity}'s authorized set {sorted(authorized)}, not every tenant")
def _filter_referenced_tenants(requested_filter: dict | None) -> frozenset[str]:
"""Pulls any tenant_id-shaped values out of a caller-supplied filter, so
VDA-04-style filter overrides (e.g. filter={"tenant_id": "TENANT-BETA"})
are validated the same as an explicit tenant_id parameter. Deliberately
narrow: this lab's filter syntax is a flat dict of equality clauses (see
README "Filter override" — no query language is built for this)."""
if not requested_filter:
return frozenset()
value = requested_filter.get("tenant_id")
if value is None:
return frozenset()
if isinstance(value, str):
# Flagged even if `value` isn't a real tenant name — an
# unauthorized-looking filter is rejected on its own terms, not
# waved through just because it happens to name nothing real.
return frozenset({value})
if isinstance(value, list):
return frozenset(value)
return frozenset()
@dataclass(frozen=True)
class ObjectDecision:
allowed: bool
status_code: int # 401 / 404 / 200 — 404, not 403, conceals cross-tenant existence (see README "HTTP behavior")
event: str
reason: str
def authorize_object(*, identity: str | None, record_tenant_id: str) -> ObjectDecision:
"""Object-level authorization for direct fetch by ID (GET /vectors/{id},
POST /fetch, GET /documents/{id}/chunks) — checked AFTER the object is
found to exist, using ITS actual tenant_id, never anything the caller
claimed. A denial here and a genuine 404 (object doesn't exist at all)
return the identical status code on purpose, so a cross-tenant ID guess
cannot be distinguished from a wrong ID by status code alone."""
if identity is None:
return ObjectDecision(False, 401, "AUTHENTICATION_REQUIRED", "no recognized identity on this request")
authorized = authorized_tenants(identity)
if not authorized or record_tenant_id not in authorized:
return ObjectDecision(False, 404, "DIRECT_OBJECT_ACCESS_DENIED", f"{identity} is not authorized for tenant {record_tenant_id}")
return ObjectDecision(True, 200, "DIRECT_OBJECT_ACCESS_ALLOWED", f"{identity} is authorized for tenant {record_tenant_id}")
vector_store/vector_store.pyExact cosine vector store; the actual vector record model (vector_id, tenant_id, document_id, classification, owner).
"""A small local vector store: exact (brute-force) cosine search, no
dependencies — mechanically identical to ASL-005/006/007's; the metadata
fields are ASL-008's actual vector record model (vector_id, tenant_id,
document_id, classification, owner) — see README "Vector record model".
`search()` and the object-lookup helpers below accept an optional
`predicate`; whether that predicate ever expresses tenant scope is decided
by the CALLER (app/server.py / mitigation/authorization.py), not by this
module — same separation of concerns as every earlier RAG lab's store.
Vulnerable mode calls these with no predicate at all; hardened mode always
passes one derived from server-side entitlements, never from the request.
"""
from __future__ import annotations
import json
import math
from dataclasses import dataclass, asdict
from typing import Callable, Optional
@dataclass(frozen=True)
class StoredChunk:
vector_id: str # == chunk_id; the identifier a caller can name directly (GET /vectors/{vector_id})
tenant_id: str
document_id: str
chunk_id: str
title: str
classification: str
owner: str
text: str
@dataclass(frozen=True)
class SearchHit:
chunk: StoredChunk
score: float
def _normalize(vector: list[float]) -> list[float]:
norm = math.sqrt(sum(value * value for value in vector))
if norm == 0:
raise ValueError("cannot normalize a zero vector")
return [value / norm for value in vector]
class FlatVectorStore:
def __init__(self, dimension: int) -> None:
self.dimension = dimension
self._entries: list[tuple[StoredChunk, list[float]]] = []
def __len__(self) -> int:
return len(self._entries)
@property
def chunks(self) -> list[StoredChunk]:
return [chunk for chunk, _ in self._entries]
def add(self, chunk: StoredChunk, vector: list[float]) -> None:
if len(vector) != self.dimension:
raise ValueError(f"expected a {self.dimension}-dimensional vector, got {len(vector)}")
if any(existing.vector_id == chunk.vector_id for existing, _ in self._entries):
raise ValueError(f"duplicate vector_id {chunk.vector_id}")
self._entries.append((chunk, _normalize(vector)))
def search(
self,
query_vector: list[float],
k: int,
predicate: Optional[Callable[[StoredChunk], bool]] = None,
) -> list[SearchHit]:
if len(query_vector) != self.dimension:
raise ValueError(f"expected a {self.dimension}-dimensional query, got {len(query_vector)}")
query = _normalize(query_vector)
hits: list[SearchHit] = []
for chunk, vector in self._entries:
if predicate is not None and not predicate(chunk):
continue
hits.append(SearchHit(chunk=chunk, score=sum(q * v for q, v in zip(query, vector))))
hits.sort(key=lambda hit: (-hit.score, hit.chunk.vector_id))
return hits[:k]
def get_by_vector_id(self, vector_id: str) -> Optional[StoredChunk]:
for chunk, _ in self._entries:
if chunk.vector_id == vector_id:
return chunk
return None
def get_by_document_id(self, document_id: str) -> list[StoredChunk]:
return [chunk for chunk, _ in self._entries if chunk.document_id == document_id]
def save(self, path: str) -> None:
payload = {
"format": "flat-vector-store-v1",
"dimension": self.dimension,
"entries": [{"chunk": asdict(chunk), "vector": vector} for chunk, vector in self._entries],
}
with open(path, "w", encoding="utf-8") as handle:
json.dump(payload, handle)
@classmethod
def load(cls, path: str) -> "FlatVectorStore":
with open(path, "r", encoding="utf-8") as handle:
payload = json.load(handle)
if payload.get("format") != "flat-vector-store-v1":
raise ValueError(f"unsupported vector index format in {path}: {payload.get('format')!r}")
store = cls(payload["dimension"])
for entry in payload["entries"]:
store.add(StoredChunk(**entry["chunk"]), entry["vector"])
return store
vector_store/retriever_helpers.pyThe tenant-scoped search operation, factored out so the critical-invariant test exercises it directly.
"""A tiny shared helper: embed a query and search a store, restricted to a
tenant scope — the core of what app/server.py's hardened /search path does
(that path also applies an optional classification filter on top; this
helper covers the tenant-scope enforcement specifically, which is what the
invariant test in tests/test_authorization.py needs to exercise against the
real store rather than a reimplementation of it).
"""
from __future__ import annotations
def search_tenant_scoped(store, query: str, *, scope: frozenset[str], embedder, k: int):
query_vector = embedder.embed_query(query)
return store.search(query_vector, k, predicate=lambda chunk: chunk.tenant_id in scope)
vector_store/embedder.pyThe embedding interface — this lab's only model call of any kind.
"""The embedding interface. Calls labs/shared/embedding_client.py. Batches a
handful of chunks per request with a longer per-request timeout — same
robustness fix as ASL-006/007's copy, against a backend observed to take
~10s per item under load. This lab's corpus is small (three tenants, one
document each), so this is mostly headroom rather than a necessity."""
from __future__ import annotations
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "shared")))
from embedding_client import embed # noqa: E402
BATCH_SIZE = 5
REQUEST_TIMEOUT = 400.0
class OpenAICompatibleEmbedder:
def __init__(self, *, base_url: str, api_key: str, model: str, query_prefix: str = "", document_prefix: str = "") -> None:
self.base_url = base_url
self.api_key = api_key
self.model = model
self.query_prefix = query_prefix
self.document_prefix = document_prefix
def _embed(self, texts: list[str]) -> list[list[float]]:
vectors: list[list[float]] = []
for start in range(0, len(texts), BATCH_SIZE):
batch = texts[start : start + BATCH_SIZE]
vectors.extend(embed(base_url=self.base_url, api_key=self.api_key, model=self.model, inputs=batch, timeout=REQUEST_TIMEOUT).vectors)
return vectors
def embed_documents(self, texts: list[str]) -> list[list[float]]:
return self._embed([self.document_prefix + text for text in texts])
def embed_query(self, text: str) -> list[float]:
return self._embed([self.query_prefix + text])[0]
vector_store/chunker.pyDeterministic fixed-size chunker, identical configuration to ASL-005/006/007.
"""Deterministic fixed-size chunker — identical to ASL-005/006/007's."""
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class Chunk:
chunk_id: str
document_id: str
index: int
text: str
char_start: int
char_end: int
def chunk_text(text: str, chunk_size: int, overlap: int) -> list[tuple[int, int]]:
if chunk_size <= 0:
raise ValueError("chunk_size must be positive")
if not 0 <= overlap < chunk_size:
raise ValueError("overlap must be >= 0 and smaller than chunk_size")
spans: list[tuple[int, int]] = []
length = len(text)
start = 0
while start < length:
end = min(start + chunk_size, length)
if end < length:
cut = max(text.rfind(" ", start + chunk_size // 2, end), text.rfind("\n", start + chunk_size // 2, end))
if cut > start:
end = cut
window = text[start:end]
stripped = window.strip()
if stripped:
lead = len(window) - len(window.lstrip())
spans.append((start + lead, start + lead + len(stripped)))
if end >= length:
break
next_start = max(end - overlap, start + 1)
while next_start < end and not text[next_start - 1].isspace():
next_start += 1
start = next_start
return spans
def chunk_document(document_id: str, text: str, chunk_size: int, overlap: int) -> list[Chunk]:
text = text.replace("\r\n", "\n")
return [
Chunk(
chunk_id=f"{document_id}-C{number:02d}",
document_id=document_id,
index=number,
text=text[start:end],
char_start=start,
char_end=end,
)
for number, (start, end) in enumerate(chunk_text(text, chunk_size, overlap), start=1)
]
vector_store/settings.pyEmbedding/chunking configuration — no chat-model settings; this lab makes no chat-completion calls.
"""Pipeline configuration for ASL-008.
Embedding model and chunking kept identical to ASL-005/006/007 for
comparability. There is no ingestion trust decision and no instruction
detector in this lab — the experimental variable is authorization at the
vector-data interface, not what gets indexed or how retrieved content is
framed to a model. Search relevance quality is secondary here; what matters
is whether tenant scope is enforced server-side (see README).
"""
from __future__ import annotations
import os
LAB_ID = "ASL-008"
LAB_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
DATA_DIR = os.path.join(LAB_DIR, "data")
CATALOG_PATH = os.path.join(DATA_DIR, "catalog.json")
EVIDENCE_DIR = os.path.join(LAB_DIR, "evidence")
# Derived, machine-specific, rebuilt with vector_store/build_index.py — not published.
INDEX_DIR = os.path.join(LAB_DIR, "index")
VECTOR_INDEX_PATH = os.path.join(INDEX_DIR, "vector-index.json")
MANIFEST_PATH = os.path.join(EVIDENCE_DIR, "index-manifest.json")
def _load_dotenv(path: str) -> None:
if not os.path.exists(path):
return
with open(path, "r", encoding="utf-8") as handle:
for line in handle:
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, _, value = line.partition("=")
key = key.strip()
if key and key not in os.environ:
os.environ[key] = value.strip()
_load_dotenv(os.path.join(LAB_DIR, ".env"))
# --- embedding (identical to ASL-005/006/007) --------------------------------
EMBEDDING_MODEL = os.environ.get("EMBEDDING_MODEL", "nomic-embed-text")
EMBEDDING_BASE_URL = os.environ.get("EMBEDDING_BASE_URL") or os.environ.get("LLM_BASE_URL", "http://localhost:11434/v1")
EMBEDDING_API_KEY = os.environ.get("EMBEDDING_API_KEY") or os.environ.get("LLM_API_KEY", "")
QUERY_PREFIX = "search_query: "
DOCUMENT_PREFIX = "search_document: "
# --- chunking (identical to ASL-005/006/007) ----------------------------------
CHUNK_STRATEGY = "fixed-size character windows, snapped back to whitespace"
CHUNK_SIZE_CHARS = 600
CHUNK_OVERLAP_CHARS = 100
# --- search (identical similarity mechanics to ASL-005/006/007) --------------
TOP_K = 6
SIMILARITY_METRIC = "cosine (vectors L2-normalized; score = dot product)"
VECTOR_BACKEND = "FlatVectorStore (in-process exact brute-force search, JSON persistence)"
def chunk_configuration() -> dict:
return {
"strategy": CHUNK_STRATEGY,
"chunk_size_chars": CHUNK_SIZE_CHARS,
"chunk_overlap_chars": CHUNK_OVERLAP_CHARS,
}
vector_store/corpus.pyLoads data/catalog.json against the committed corpus; validates every declared canary actually appears in its file.
"""Load the synthetic multi-tenant corpus and its metadata catalog.
data/catalog.json is the single source of document metadata (document_id,
tenant_id, title, classification, owner, canary); the Markdown files hold
only content. Loading validates the catalog against the files, and that
every declared canary actually appears in its file, so an inconsistent
corpus fails loudly instead of producing a quietly wrong index.
"""
from __future__ import annotations
import json
import os
import sys
from dataclasses import dataclass
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from entitlements import TENANTS # noqa: E402
from settings import CATALOG_PATH, DATA_DIR # noqa: E402
REQUIRED_FIELDS = ("document_id", "path", "tenant_id", "title", "classification", "owner")
@dataclass(frozen=True)
class Document:
document_id: str
path: str # relative to data/
tenant_id: str
title: str
classification: str
owner: str
canary: str | None
text: str
def load_corpus(catalog_path: str = CATALOG_PATH, data_dir: str = DATA_DIR) -> list[Document]:
with open(catalog_path, "r", encoding="utf-8") as handle:
catalog = json.load(handle)
documents: list[Document] = []
seen_ids: set[str] = set()
for entry in catalog["documents"]:
missing = [field for field in REQUIRED_FIELDS if field not in entry]
if missing:
raise ValueError(f"catalog entry {entry.get('document_id', '?')} is missing {missing}")
if entry["document_id"] in seen_ids:
raise ValueError(f"duplicate document_id {entry['document_id']}")
seen_ids.add(entry["document_id"])
if entry["tenant_id"] not in TENANTS:
raise ValueError(f"{entry['document_id']}: tenant_id {entry['tenant_id']!r} is not a known tenant ({TENANTS})")
with open(os.path.join(data_dir, entry["path"]), "r", encoding="utf-8") as handle:
text = handle.read().replace("\r\n", "\n")
canary = entry.get("canary")
if canary and canary not in text:
raise ValueError(f"{entry['document_id']}: canary {canary!r} does not appear in {entry['path']}")
documents.append(
Document(
document_id=entry["document_id"],
path=entry["path"],
tenant_id=entry["tenant_id"],
title=entry["title"],
classification=entry["classification"],
owner=entry["owner"],
canary=canary,
text=text,
)
)
return documents
vector_store/build_index.pyBuilds the ONE shared index across all three tenants.
"""Build the ASL-008 vector index and its manifest.
python3 vector_store/build_index.py # embed the corpus, write the index + manifest
python3 vector_store/build_index.py --check # verify the committed manifest against the
# corpus and settings; needs no embedding model
ONE shared index across all three tenants (see README "Index architecture"
— this lab deliberately demonstrates that a shared index is not
automatically insecure if authorization is enforced correctly elsewhere).
The vector index (index/vector-index.json) is derived and machine-specific,
so it is not published; the manifest (evidence/index-manifest.json) is:
every document and chunk with its tenant_id, classification and owner, the
embedding model, the chunking configuration and a hash of each chunk's text.
"""
from __future__ import annotations
import hashlib
import json
import os
import sys
from datetime import datetime, timezone
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
import bootstrap # noqa: E402,F401
import settings # noqa: E402
from chunker import chunk_document # noqa: E402
from corpus import Document, load_corpus # noqa: E402
from embedder import OpenAICompatibleEmbedder # noqa: E402
from vector_store import FlatVectorStore, StoredChunk # noqa: E402
MANIFEST_VERSION = 1
def sha256(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def chunk_corpus(documents: list[Document]) -> list[tuple[Document, object]]:
pairs = []
for document in documents:
for chunk in chunk_document(document.document_id, document.text, settings.CHUNK_SIZE_CHARS, settings.CHUNK_OVERLAP_CHARS):
pairs.append((document, chunk))
return pairs
def build_manifest(documents: list[Document], *, embedding_dimension: int | None, index_timestamp: str | None) -> dict:
pairs = chunk_corpus(documents)
chunk_ids_by_doc: dict[str, list[str]] = {}
for document, chunk in pairs:
chunk_ids_by_doc.setdefault(document.document_id, []).append(chunk.chunk_id)
return {
"manifest_version": MANIFEST_VERSION,
"lab_id": settings.LAB_ID,
"index_timestamp": index_timestamp,
"embedding": {
"model": settings.EMBEDDING_MODEL,
"dimension": embedding_dimension,
"query_prefix": settings.QUERY_PREFIX,
"document_prefix": settings.DOCUMENT_PREFIX,
},
"vector_backend": settings.VECTOR_BACKEND,
"similarity_metric": settings.SIMILARITY_METRIC,
"chunking": settings.chunk_configuration(),
"documents": [
{
"document_id": d.document_id,
"path": d.path,
"tenant_id": d.tenant_id,
"title": d.title,
"classification": d.classification,
"owner": d.owner,
"text_sha256": sha256(d.text),
"chunk_ids": chunk_ids_by_doc[d.document_id],
}
for d in documents
],
"chunks": [
{
"vector_id": c.chunk_id,
"chunk_id": c.chunk_id,
"document_id": d.document_id,
"tenant_id": d.tenant_id,
"title": d.title,
"classification": d.classification,
"owner": d.owner,
"char_start": c.char_start,
"char_end": c.char_end,
"text_sha256": sha256(c.text),
"text": c.text,
}
for d, c in pairs
],
}
def verify_manifest(manifest: dict, documents: list[Document]) -> list[str]:
expected = build_manifest(documents, embedding_dimension=None, index_timestamp=None)
problems: list[str] = []
for key in ("manifest_version", "lab_id", "vector_backend", "similarity_metric", "chunking", "documents", "chunks"):
if manifest.get(key) != expected[key]:
problems.append(f"{key} differs from what the current corpus/settings produce")
got, want = manifest.get("embedding", {}), expected["embedding"]
for key in ("model", "query_prefix", "document_prefix"):
if got.get(key) != want[key]:
problems.append(f"embedding.{key}: manifest has {got.get(key)!r}, settings have {want[key]!r}")
return problems
def main() -> int:
documents = load_corpus()
if "--check" in sys.argv:
with open(settings.MANIFEST_PATH, "r", encoding="utf-8") as handle:
manifest = json.load(handle)
problems = verify_manifest(manifest, documents)
if problems:
print("MANIFEST DOES NOT MATCH THE CORPUS/SETTINGS:")
for problem in problems:
print(" -", problem)
return 1
print(f"OK: manifest matches the corpus and settings ({len(manifest['documents'])} documents, {len(manifest['chunks'])} chunks).")
return 0
pairs = chunk_corpus(documents)
embedder = OpenAICompatibleEmbedder(
base_url=settings.EMBEDDING_BASE_URL,
api_key=settings.EMBEDDING_API_KEY,
model=settings.EMBEDDING_MODEL,
query_prefix=settings.QUERY_PREFIX,
document_prefix=settings.DOCUMENT_PREFIX,
)
print(f"Embedding {len(pairs)} chunks from {len(documents)} documents ({len(set(d.tenant_id for d in documents))} tenants) with {settings.EMBEDDING_MODEL} ...")
vectors = embedder.embed_documents([chunk.text for _, chunk in pairs])
dimension = len(vectors[0])
store = FlatVectorStore(dimension)
for (document, chunk), vector in zip(pairs, vectors):
store.add(
StoredChunk(
vector_id=chunk.chunk_id,
tenant_id=document.tenant_id,
document_id=document.document_id,
chunk_id=chunk.chunk_id,
title=document.title,
classification=document.classification,
owner=document.owner,
text=chunk.text,
),
vector,
)
os.makedirs(settings.INDEX_DIR, exist_ok=True)
os.makedirs(settings.EVIDENCE_DIR, exist_ok=True)
store.save(settings.VECTOR_INDEX_PATH)
manifest = build_manifest(documents, embedding_dimension=dimension, index_timestamp=datetime.now(timezone.utc).isoformat())
with open(settings.MANIFEST_PATH, "w", encoding="utf-8") as handle:
json.dump(manifest, handle, indent=2)
handle.write("\n")
print(f"Wrote {settings.VECTOR_INDEX_PATH} ({len(store)} vectors, dimension {dimension})")
print(f"Wrote {settings.MANIFEST_PATH}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
data/catalog.jsonTest-fixture ground truth only (which document belongs to which tenant) — not a security control.
{
"corpus_version": 1,
"synthetic_notice": "Every tenant, document, name and canary in this corpus is synthetic and invented for the Shah AI Security Lab (ASL-008). Nothing describes a real organisation, person or system. This file is a TEST-FIXTURE ground-truth description (which document belongs to which tenant, which canary exists) — it is NOT a security control; the vector service never reads this file to make an authorization decision. entitlements.py is the security control.",
"documents": [
{
"document_id": "ALPHA-001",
"path": "tenant-alpha/ALPHA-001.md",
"tenant_id": "TENANT-ALPHA",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"canary": "ALPHA-CANARY-ASL008-1101"
},
{
"document_id": "BETA-001",
"path": "tenant-beta/BETA-001.md",
"tenant_id": "TENANT-BETA",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"canary": "BETA-CANARY-ASL008-2202"
},
{
"document_id": "GAMMA-001",
"path": "tenant-gamma/GAMMA-001.md",
"tenant_id": "TENANT-GAMMA",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"canary": "GAMMA-CANARY-ASL008-3303"
}
]
}
data/tenant-alpha/ALPHA-001.mdTenant Alpha's document (DOC canary ALPHA-CANARY-ASL008-1101).
# Alpha Quarterly Planning Summary
> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Alpha, and everything
> below are fictional, invented for the Shah AI Security Lab (ASL-008).
Marker: ALPHA-CANARY-ASL008-1101
Tenant Alpha's quarterly planning cycle sets budget and headcount targets
for each business unit ahead of the next fiscal quarter. This cycle's
priorities are cloud infrastructure consolidation, a refresh of the
customer support tooling, and a 6% increase in the engineering budget to
support the new observability initiative.
Department leads submit their draft plans two weeks before the quarterly
review; finance reconciles the submissions against the approved top-line
budget before the plans are finalized. Any request above the standard
discretionary threshold requires sign-off from the Tenant Alpha finance
lead before it can be included in the finalized plan.
data/tenant-beta/BETA-001.mdTenant Beta's document, classification confidential (canary BETA-CANARY-ASL008-2202).
# Beta Acquisition Planning Brief
> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything
> below are fictional, invented for the Shah AI Security Lab (ASL-008).
Marker: BETA-CANARY-ASL008-2202
Tenant Beta is evaluating the acquisition of a small fictional logistics
analytics vendor, codenamed Project Harborlight, as part of this quarter's
acquisition planning. Due diligence covers the target's technology stack,
customer contract terms, and outstanding liabilities.
The current acquisition planning timeline targets a signed letter of intent
within six weeks, contingent on the due diligence findings and confirmation
that the target's data handling practices meet Tenant Beta's compliance
requirements. Details of Project Harborlight are restricted to the Tenant
Beta deal team until a public announcement, if any, is made.
data/tenant-gamma/GAMMA-001.mdTenant Gamma's document (canary GAMMA-CANARY-ASL008-3303).
# Gamma Personnel Planning Notes
> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Gamma, and everything
> below are fictional, invented for the Shah AI Security Lab (ASL-008).
Marker: GAMMA-CANARY-ASL008-3303
Tenant Gamma's personnel planning notes for the coming quarter cover
headcount requests for the fictional Gamma support organization, a planned
reorganization of the regional support teams, and updated on-call staffing
targets.
Individual compensation and performance details are intentionally excluded
from this planning note; it covers only headcount counts, role
requisitions, and staffing ratios. Personnel planning decisions are
reviewed by the Tenant Gamma people-operations lead before headcount
requests are opened.
Attack Harness
The black-box harness and the test cases it runs.
attack-harness/test_cases.jsonThe eight test cases (VDA-01 to VDA-08), fixed before the first run against the server.
[
{
"id": "VDA-01",
"category": "cross_tenant_search",
"round": 1,
"is_attack": true,
"identity": "ALICE",
"endpoint": "search",
"request": { "query": "acquisition planning", "tenant_id": "TENANT-BETA" }
},
{
"id": "VDA-02",
"category": "cross_tenant_direct_vector_fetch",
"round": 1,
"is_attack": true,
"identity": "ALICE",
"endpoint": "vector",
"request": { "vector_id": "BETA-001-C01" }
},
{
"id": "VDA-03",
"category": "cross_tenant_document_chunks",
"round": 1,
"is_attack": true,
"identity": "ALICE",
"endpoint": "chunks",
"request": { "document_id": "BETA-001" }
},
{
"id": "VDA-04",
"category": "filter_override",
"round": 1,
"is_attack": true,
"identity": "ALICE",
"endpoint": "search",
"request": { "query": "planning", "filter": { "classification": "confidential", "tenant_id": "TENANT-BETA" } }
},
{
"id": "VDA-05",
"category": "missing_tenant_filter",
"round": 1,
"is_attack": true,
"identity": "ALICE",
"endpoint": "search",
"request": { "query": "planning" }
},
{
"id": "VDA-06",
"category": "unauthorized_real_tenant",
"round": 1,
"is_attack": true,
"identity": "ALICE",
"endpoint": "search",
"request": { "query": "personnel planning", "tenant_id": "TENANT-GAMMA" }
},
{
"id": "VDA-07",
"category": "authorized_search_control",
"round": 1,
"is_attack": false,
"identity": "ALICE",
"endpoint": "search",
"request": { "query": "quarterly planning budget", "tenant_id": "TENANT-ALPHA" },
"expected_tenant_id": "TENANT-ALPHA",
"expected_document_id": "ALPHA-001"
},
{
"id": "VDA-08",
"category": "second_tenant_control",
"round": 1,
"is_attack": false,
"identity": "BOB",
"endpoint": "search",
"request": { "query": "acquisition planning", "tenant_id": "TENANT-BETA" },
"expected_tenant_id": "TENANT-BETA",
"expected_document_id": "BETA-001"
}
]
attack-harness/harness.pyCalls the local HTTP API directly for every case, in both modes — no LLM anywhere in this file.
"""ASL-008 attack harness.
Runs all eight cases (VDA-01 to VDA-08) against both modes (vulnerable,
hardened) by calling the running local vector API directly — no LLM
involved (see README "Why there is no LLM in the primary evidence").
Usage:
python3 vector_store/build_index.py # build the index first
python3 app/server.py & # in another terminal
python3 attack-harness/harness.py
"""
from __future__ import annotations
import json
import os
import sys
import urllib.error
import urllib.request
from datetime import datetime, timezone
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
import bootstrap # noqa: E402,F401
import measure # noqa: E402
import settings # noqa: E402
from entitlements import authorized_tenants # noqa: E402
from run_metadata import source_revision # noqa: E402
HERE = os.path.dirname(os.path.abspath(__file__))
TEST_CASES_PATH = os.path.join(HERE, "test_cases.json")
EVIDENCE_DIR = settings.EVIDENCE_DIR
MODES = ("vulnerable", "hardened")
def call(base_url: str, *, mode: str, identity: str | None, endpoint: str, request: dict) -> tuple[int, dict]:
headers = {"X-Lab-Mode": mode}
if identity:
headers["X-Lab-User"] = identity
if endpoint == "search":
url, method, data = base_url + "/search", "POST", json.dumps(request).encode("utf-8")
headers["Content-Type"] = "application/json"
elif endpoint == "vector":
url, method, data = base_url + "/vectors/" + urllib.request.quote(request["vector_id"]), "GET", None
elif endpoint == "fetch":
url, method, data = base_url + "/fetch", "POST", json.dumps(request).encode("utf-8")
headers["Content-Type"] = "application/json"
elif endpoint == "chunks":
url, method, data = base_url + "/documents/" + urllib.request.quote(request["document_id"]) + "/chunks", "GET", None
else:
raise ValueError(f"unknown endpoint {endpoint!r}")
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=120) as response:
return response.status, json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8")
try:
return exc.code, json.loads(body)
except json.JSONDecodeError:
return exc.code, {"error": body}
def run_case(base_url: str, mode: str, case: dict) -> dict:
timestamp = datetime.now(timezone.utc).isoformat()
status_code, response = call(base_url, mode=mode, identity=case["identity"], endpoint=case["endpoint"], request=case["request"])
evaluated = measure.evaluate(identity=case["identity"], endpoint=case["endpoint"], request=case["request"], status_code=status_code, response=response)
expected = {}
if not case["is_attack"]:
expected = measure.check_authorized_control(identity=case["identity"], expected_tenant_id=case["expected_tenant_id"], expected_document_id=case["expected_document_id"], response=response)
return {
"test_id": case["id"],
"category": case["category"],
"round": case.get("round", 1),
"is_attack": case["is_attack"],
"mode": mode,
"authenticated_user": case["identity"],
"authorized_tenants": sorted(authorized_tenants(case["identity"]) or []),
"endpoint": case["endpoint"],
"request": case["request"],
"timestamp": timestamp,
"http_status": status_code,
"authorization_decision": response.get("authorization_decision"),
**evaluated,
**expected,
"raw_response": response,
}
def summarize(mode_results: list[dict]) -> dict:
attacks = [r for r in mode_results if r["is_attack"]]
controls = [r for r in mode_results if not r["is_attack"]]
search_attacks = [r for r in attacks if r["endpoint"] == "search"]
object_attacks = [r for r in attacks if r["endpoint"] != "search"]
return {
"adversarial_cases": len(attacks),
"authorized_controls": len(controls),
"cross_tenant_searches_accepted": sum(1 for r in search_attacks if r["unauthorized_search_accepted"]),
"cross_tenant_searches_total": len(search_attacks),
"direct_unauthorized_object_fetches": sum(1 for r in object_attacks if r["direct_object_access_bypass"]),
"direct_object_fetches_total": len(object_attacks),
"unauthorized_records_returned": sum(len(r["unauthorized_record_ids"]) for r in attacks),
"unauthorized_canaries_exposed": sorted({m for r in attacks for m in r["protected_markers"]}),
"global_searches_occurred": sum(1 for r in attacks if r.get("global_search_occurred")),
"policy_denied_count": sum(1 for r in attacks if r["policy_denied"]),
"authorized_controls_passed": sum(1 for r in controls if r.get("authorized_functionality_preserved")),
"invariant_violations": [
{"test_id": r["test_id"], "unauthorized_record_ids": r["unauthorized_record_ids"]}
for r in mode_results
if r["unauthorized_record_ids"]
],
}
def run(base_url: str) -> dict:
revision = source_revision([settings.LAB_DIR, os.path.join(settings.LAB_DIR, "..", "shared")], exclude_paths=[EVIDENCE_DIR, settings.INDEX_DIR])
with open(TEST_CASES_PATH, "r", encoding="utf-8") as handle:
test_cases = json.load(handle)
with open(settings.MANIFEST_PATH, "r", encoding="utf-8") as handle:
index_manifest = json.load(handle)
results_by_mode: dict[str, list[dict]] = {}
for mode in MODES:
results_by_mode[mode] = [run_case(base_url, mode, case) for case in test_cases]
return {
"lab_id": settings.LAB_ID,
"run_at": datetime.now(timezone.utc).isoformat(),
"source_revision": revision,
"base_url": base_url,
"index_config": {k: v for k, v in index_manifest.items() if k != "chunks"},
"summary_by_mode": {mode: summarize(results_by_mode[mode]) for mode in MODES},
"results_by_mode": results_by_mode,
}
def main() -> None:
port = int(os.environ.get("LAB_PORT", "8794"))
base_url = f"http://127.0.0.1:{port}"
print(f"Running ASL-008 attack harness against {base_url}")
try:
report = run(base_url)
except FileNotFoundError as exc:
print(f"FAILED: {exc}")
print("Build the index first: python3 vector_store/build_index.py")
raise SystemExit(1)
except Exception as exc:
print(f"FAILED: {exc}")
print("Is the lab server running? python3 app/server.py")
raise SystemExit(1)
os.makedirs(EVIDENCE_DIR, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
with open(os.path.join(EVIDENCE_DIR, "latest.json"), "w", encoding="utf-8") as handle:
json.dump(report, handle, indent=2)
with open(os.path.join(EVIDENCE_DIR, f"run-{stamp}.json"), "w", encoding="utf-8") as handle:
json.dump(report, handle, indent=2)
summary = {k: v for k, v in report.items() if k != "results_by_mode"}
with open(os.path.join(EVIDENCE_DIR, "summary.json"), "w", encoding="utf-8") as handle:
json.dump(summary, handle, indent=2)
print(json.dumps(report["summary_by_mode"], indent=2))
print(f"\nsource revision: {report['source_revision'].get('git_commit')} dirty={report['source_revision'].get('git_dirty')}")
print(f"Wrote {os.path.join(EVIDENCE_DIR, 'latest.json')}")
if __name__ == "__main__":
main()
attack-harness/measure.pyIndependent measurement of the eight tracked fields, computed from the server's actual response — never from its self-reported decision.
"""Independent, deterministic measurement for ASL-008.
Everything here is computed from the server's own observable HTTP
response — the actual records returned, their actual tenant_id, and their
actual text — checked against entitlements.py (the ONE authoritative
policy) directly. It is never taken from the server's own self-reported
`authorization_decision` field, which is logged alongside for evidence but
is not what determines these results — a server that returned unauthorized
records while CLAIMING to have denied them would still be caught here.
Eight fields, deliberately kept separate (never collapsed into one "attack
success rate"):
UNAUTHORIZED_SEARCH_ACCEPTED did a search request that named an
out-of-scope tenant/filter get a 200
instead of being rejected?
UNAUTHORIZED_RECORD_RETURNED did ANY returned record belong to a
tenant outside the identity's
authorized set?
UNAUTHORIZED_TEXT_EXPOSED was such a record's text field present?
UNAUTHORIZED_METADATA_EXPOSED was such a record's classification/
owner/document_id present?
DIRECT_OBJECT_ACCESS_BYPASS for a direct-fetch case (vector/fetch/
chunks), was the object returned despite
belonging to an unauthorized tenant?
GLOBAL_SEARCH_OCCURRED for a request with no tenant_id, did the
result set span more than one tenant?
POLICY_DENIED was the request rejected (401/403/404)?
AUTHORIZED_FUNCTIONALITY_PRESERVED for a benign control, did it succeed
and return only correctly-scoped records?
"""
from __future__ import annotations
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from entitlements import authorized_tenants # noqa: E402
def _records_of(response: dict, endpoint: str) -> list[dict]:
if endpoint == "search":
return response.get("results", [])
if endpoint == "vector":
record = response.get("record")
return [record] if record else []
if endpoint == "fetch":
return [row["record"] for row in response.get("results", []) if row.get("record")]
if endpoint == "chunks":
return response.get("chunks", [])
raise ValueError(f"unknown endpoint {endpoint!r}")
def evaluate(*, identity: str, endpoint: str, request: dict, status_code: int, response: dict) -> dict:
authorized = authorized_tenants(identity) or frozenset()
records = _records_of(response, endpoint)
record_tenants = [r["tenant_id"] for r in records]
unauthorized_records = [r for r in records if r["tenant_id"] not in authorized]
policy_denied = status_code >= 400
unauthorized_search_accepted = None
global_search_occurred = None
direct_object_access_bypass = None
if endpoint == "search":
requested_tenant = request.get("tenant_id")
requested_out_of_scope = bool(requested_tenant) and requested_tenant not in authorized
filter_tenant = (request.get("filter") or {}).get("tenant_id")
filter_out_of_scope = bool(filter_tenant) and filter_tenant not in authorized
unauthorized_search_accepted = (requested_out_of_scope or filter_out_of_scope) and status_code == 200
if not requested_tenant and not filter_tenant:
# Truly unscoped — no tenant_id AND no filter.tenant_id at all
# (VDA-05). A request that names a tenant only via `filter`
# (VDA-04) is scoped, just through a different field, so it is
# not "global" — its risk is measured by
# unauthorized_search_accepted instead.
global_search_occurred = len(set(record_tenants)) > 1 or bool(unauthorized_records)
else:
direct_object_access_bypass = bool(unauthorized_records) and status_code == 200
return {
"unauthorized_search_accepted": unauthorized_search_accepted,
"unauthorized_record_returned": bool(unauthorized_records),
"unauthorized_text_exposed": any(bool(r.get("text")) for r in unauthorized_records),
"unauthorized_metadata_exposed": any(r.get("classification") or r.get("owner") or r.get("document_id") for r in unauthorized_records),
"direct_object_access_bypass": direct_object_access_bypass,
"global_search_occurred": global_search_occurred,
"policy_denied": policy_denied,
"record_tenants": record_tenants,
"record_ids_returned": [r["vector_id"] for r in records],
"unauthorized_record_ids": [r["vector_id"] for r in unauthorized_records],
"protected_markers": [marker for marker in _markers_in(records) if _marker_tenant(marker) not in authorized],
}
# Canaries are literal substrings of the corpus text (see data/catalog.json);
# looked up directly here rather than imported from the corpus loader, so
# this measurement stays independent of the pipeline it is checking.
_CANARIES = {
"ALPHA-CANARY-ASL008-1101": "TENANT-ALPHA",
"BETA-CANARY-ASL008-2202": "TENANT-BETA",
"GAMMA-CANARY-ASL008-3303": "TENANT-GAMMA",
}
def _markers_in(records: list[dict]) -> list[str]:
found = []
for record in records:
text = record.get("text", "")
for marker in _CANARIES:
if marker in text:
found.append(marker)
return found
def _marker_tenant(marker: str) -> str:
return _CANARIES[marker]
def check_authorized_control(*, identity: str, expected_tenant_id: str, expected_document_id: str, response: dict) -> dict:
records = response.get("results", [])
record_tenants = {r["tenant_id"] for r in records}
expected_doc_present = any(r["document_id"] == expected_document_id for r in records)
only_authorized_tenant = record_tenants <= {expected_tenant_id}
return {
"expected_document_present": expected_doc_present,
"only_authorized_tenant_returned": only_authorized_tenant,
"authorized_functionality_preserved": expected_doc_present and only_authorized_tenant,
}
Tests
Unit and integration tests that need no model.
tests/test_authorization.pyThe core: every VDA-0x policy decision, AUDITOR's set-based entitlement, fail-closed behavior, object-level authorization, and the critical invariant.
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
import context # noqa: E402,F401
from authorization import authorize_object, resolve_search_scope # noqa: E402
from retriever_helpers import search_tenant_scoped # noqa: E402
class SearchScopeBasics(unittest.TestCase):
def test_alice_is_authorized_for_alpha(self):
decision = resolve_search_scope(identity="ALICE", requested_tenant_id="TENANT-ALPHA", requested_filter=None)
self.assertTrue(decision.allowed)
self.assertEqual(decision.scope, frozenset({"TENANT-ALPHA"}))
self.assertEqual(decision.status_code, 200)
def test_alice_is_denied_beta(self):
decision = resolve_search_scope(identity="ALICE", requested_tenant_id="TENANT-BETA", requested_filter=None)
self.assertFalse(decision.allowed)
self.assertEqual(decision.status_code, 403)
self.assertEqual(decision.event, "CROSS_TENANT_SCOPE_ATTEMPT")
def test_bob_is_authorized_for_beta(self):
decision = resolve_search_scope(identity="BOB", requested_tenant_id="TENANT-BETA", requested_filter=None)
self.assertTrue(decision.allowed)
self.assertEqual(decision.scope, frozenset({"TENANT-BETA"}))
def test_missing_tenant_defaults_to_identity_scope_not_global(self):
decision = resolve_search_scope(identity="ALICE", requested_tenant_id=None, requested_filter=None)
self.assertTrue(decision.allowed)
self.assertEqual(decision.scope, frozenset({"TENANT-ALPHA"}))
self.assertEqual(decision.event, "GLOBAL_SCOPE_DEFAULTED_TO_IDENTITY")
def test_a_real_but_unauthorized_tenant_is_denied(self):
decision = resolve_search_scope(identity="ALICE", requested_tenant_id="TENANT-GAMMA", requested_filter=None)
self.assertFalse(decision.allowed)
self.assertEqual(decision.status_code, 403)
def test_an_invented_tenant_is_denied_identically(self):
decision = resolve_search_scope(identity="ALICE", requested_tenant_id="TENANT-DOES-NOT-EXIST", requested_filter=None)
self.assertFalse(decision.allowed)
self.assertEqual(decision.status_code, 403)
def test_unauthenticated_is_401_not_403(self):
decision = resolve_search_scope(identity=None, requested_tenant_id="TENANT-ALPHA", requested_filter=None)
self.assertFalse(decision.allowed)
self.assertEqual(decision.status_code, 401)
def test_a_filter_naming_an_unauthorized_tenant_is_rejected(self):
decision = resolve_search_scope(identity="ALICE", requested_tenant_id=None, requested_filter={"tenant_id": "TENANT-BETA"})
self.assertFalse(decision.allowed)
self.assertEqual(decision.event, "UNAUTHORIZED_FILTER_REJECTED")
class FailClosedOnMissingEntitlement(unittest.TestCase):
"""DAVE authenticates successfully (a real, recognized identity) but has
NO entitlement record — models a provisioning gap. The hardened path
must deny, never default to every tenant that exists."""
def test_a_known_identity_with_no_entitlement_record_is_denied(self):
decision = resolve_search_scope(identity="DAVE", requested_tenant_id=None, requested_filter=None)
self.assertFalse(decision.allowed)
self.assertEqual(decision.status_code, 403)
self.assertEqual(decision.event, "AUTHORIZATION_IDENTITY_UNRESOLVED")
self.assertEqual(decision.scope, frozenset())
def test_a_known_identity_with_no_entitlement_cannot_request_a_specific_tenant_either(self):
decision = resolve_search_scope(identity="DAVE", requested_tenant_id="TENANT-ALPHA", requested_filter=None)
self.assertFalse(decision.allowed)
class SetBasedEntitlement(unittest.TestCase):
"""AUDITOR is authorized for TWO tenants — proves the policy is genuine
set membership, not a hardcoded "only Alpha" equality check."""
def test_auditor_can_access_alpha(self):
self.assertTrue(resolve_search_scope(identity="AUDITOR", requested_tenant_id="TENANT-ALPHA", requested_filter=None).allowed)
def test_auditor_can_access_beta(self):
self.assertTrue(resolve_search_scope(identity="AUDITOR", requested_tenant_id="TENANT-BETA", requested_filter=None).allowed)
def test_auditor_cannot_access_gamma(self):
self.assertFalse(resolve_search_scope(identity="AUDITOR", requested_tenant_id="TENANT-GAMMA", requested_filter=None).allowed)
def test_auditors_default_scope_is_both_authorized_tenants(self):
decision = resolve_search_scope(identity="AUDITOR", requested_tenant_id=None, requested_filter=None)
self.assertEqual(decision.scope, frozenset({"TENANT-ALPHA", "TENANT-BETA"}))
class ObjectLevelAuthorization(unittest.TestCase):
def test_direct_beta_vector_is_denied_to_alice(self):
decision = authorize_object(identity="ALICE", record_tenant_id="TENANT-BETA")
self.assertFalse(decision.allowed)
self.assertEqual(decision.status_code, 404)
def test_direct_alpha_vector_is_allowed_to_alice(self):
decision = authorize_object(identity="ALICE", record_tenant_id="TENANT-ALPHA")
self.assertTrue(decision.allowed)
self.assertEqual(decision.status_code, 200)
def test_unauthenticated_object_access_is_401(self):
decision = authorize_object(identity=None, record_tenant_id="TENANT-ALPHA")
self.assertEqual(decision.status_code, 401)
def test_daves_missing_entitlement_denies_object_access_too(self):
decision = authorize_object(identity="DAVE", record_tenant_id="TENANT-ALPHA")
self.assertFalse(decision.allowed)
class CriticalInvariant(unittest.TestCase):
"""For every identity and every real tenant, a hardened-mode search's
returned records must belong ONLY to that identity's authorized
tenants — checked directly against the real corpus and store, not a
fixture."""
def test_every_hardened_search_result_belongs_to_an_authorized_tenant(self):
store = context.build_store()
cases = [
("ALICE", "TENANT-ALPHA"),
("BOB", "TENANT-BETA"),
("CAROL", "TENANT-GAMMA"),
("AUDITOR", "TENANT-ALPHA"),
("AUDITOR", "TENANT-BETA"),
]
for identity, tenant in cases:
with self.subTest(identity=identity, tenant=tenant):
decision = resolve_search_scope(identity=identity, requested_tenant_id=tenant, requested_filter=None)
self.assertTrue(decision.allowed)
hits = search_tenant_scoped(store, "planning", scope=decision.scope, embedder=context.FakeEmbedder(), k=10)
for hit in hits:
self.assertIn(hit.chunk.tenant_id, decision.scope, f"{identity} received a {hit.chunk.tenant_id} record outside {decision.scope}")
if __name__ == "__main__":
unittest.main()
tests/test_vector_store.pyRanking, determinism, tenant pre-filtering, object lookup and persistence.
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(__file__))
import context # noqa: E402,F401
from vector_store import FlatVectorStore, StoredChunk # noqa: E402
def chunk(vector_id="V1", tenant_id="TENANT-ALPHA", document_id="D1", text="hello"):
return StoredChunk(vector_id=vector_id, tenant_id=tenant_id, document_id=document_id, chunk_id=vector_id, title="T", classification="internal", owner="owner@example.com", text=text)
class Validation(unittest.TestCase):
def test_dimension_mismatch_is_rejected(self):
store = FlatVectorStore(3)
with self.assertRaises(ValueError):
store.add(chunk(), [1.0, 0.0])
def test_duplicate_vector_id_is_rejected(self):
store = FlatVectorStore(2)
store.add(chunk(vector_id="V1"), [1.0, 0.0])
with self.assertRaises(ValueError):
store.add(chunk(vector_id="V1"), [0.0, 1.0])
def test_zero_vector_is_rejected(self):
store = FlatVectorStore(2)
with self.assertRaises(ValueError):
store.add(chunk(), [0.0, 0.0])
class Ranking(unittest.TestCase):
def test_orders_by_cosine_similarity(self):
store = FlatVectorStore(2)
store.add(chunk(vector_id="close"), [1.0, 0.01])
store.add(chunk(vector_id="far"), [0.01, 1.0])
hits = store.search([1.0, 0.0], k=2)
self.assertEqual([h.chunk.vector_id for h in hits], ["close", "far"])
def test_ties_break_on_vector_id_so_results_are_deterministic(self):
store = FlatVectorStore(2)
store.add(chunk(vector_id="b"), [1.0, 0.0])
store.add(chunk(vector_id="a"), [1.0, 0.0])
hits = store.search([1.0, 0.0], k=2)
self.assertEqual([h.chunk.vector_id for h in hits], ["a", "b"])
class TenantPreFilter(unittest.TestCase):
def test_predicate_removes_out_of_scope_tenants(self):
store = FlatVectorStore(2)
store.add(chunk(vector_id="a", tenant_id="TENANT-ALPHA"), [1.0, 0.0])
store.add(chunk(vector_id="b", tenant_id="TENANT-BETA"), [0.9, 0.1])
hits = store.search([1.0, 0.0], k=2, predicate=lambda c: c.tenant_id == "TENANT-ALPHA")
self.assertEqual([h.chunk.vector_id for h in hits], ["a"])
class ObjectLookup(unittest.TestCase):
def test_get_by_vector_id_finds_the_exact_record(self):
store = FlatVectorStore(2)
store.add(chunk(vector_id="a"), [1.0, 0.0])
self.assertEqual(store.get_by_vector_id("a").vector_id, "a")
self.assertIsNone(store.get_by_vector_id("does-not-exist"))
def test_get_by_document_id_finds_every_chunk_of_that_document(self):
store = FlatVectorStore(2)
store.add(chunk(vector_id="a-C01", document_id="a"), [1.0, 0.0])
store.add(chunk(vector_id="a-C02", document_id="a"), [0.0, 1.0])
store.add(chunk(vector_id="b-C01", document_id="b"), [1.0, 1.0])
self.assertEqual({c.vector_id for c in store.get_by_document_id("a")}, {"a-C01", "a-C02"})
self.assertEqual(store.get_by_document_id("does-not-exist"), [])
class Persistence(unittest.TestCase):
def test_save_load_roundtrip_preserves_tenant_metadata(self):
import tempfile
store = FlatVectorStore(2)
store.add(chunk(vector_id="a", tenant_id="TENANT-BETA"), [1.0, 0.0])
with tempfile.TemporaryDirectory() as tmp:
path = os.path.join(tmp, "index.json")
store.save(path)
loaded = FlatVectorStore.load(path)
self.assertEqual(loaded.get_by_vector_id("a").tenant_id, "TENANT-BETA")
if __name__ == "__main__":
unittest.main()
tests/test_corpus.pyCatalog validation — an unknown tenant or a missing canary is rejected loudly.
import json
import os
import sys
import tempfile
import unittest
sys.path.insert(0, os.path.dirname(__file__))
import context # noqa: E402,F401
from corpus import load_corpus # noqa: E402
class LoadsTheRealCorpus(unittest.TestCase):
def test_three_documents_one_per_tenant(self):
documents = load_corpus()
self.assertEqual(len(documents), 3)
self.assertEqual({d.tenant_id for d in documents}, {"TENANT-ALPHA", "TENANT-BETA", "TENANT-GAMMA"})
def test_every_canary_appears_verbatim_in_its_own_document(self):
for document in load_corpus():
self.assertIn(document.canary, document.text)
class CatalogValidation(unittest.TestCase):
def _catalog_dir(self, documents):
tmp = tempfile.mkdtemp()
os.makedirs(os.path.join(tmp, "data"))
with open(os.path.join(tmp, "data", "a.md"), "w", encoding="utf-8") as handle:
handle.write("hello world")
with open(os.path.join(tmp, "catalog.json"), "w", encoding="utf-8") as handle:
json.dump({"documents": documents}, handle)
return os.path.join(tmp, "catalog.json"), os.path.join(tmp, "data")
def test_a_declared_canary_missing_from_the_file_is_rejected(self):
catalog_path, data_dir = self._catalog_dir(
[{"document_id": "D1", "path": "a.md", "tenant_id": "TENANT-ALPHA", "title": "A", "classification": "internal", "owner": "x", "canary": "NOT-IN-FILE"}]
)
with self.assertRaises(ValueError):
load_corpus(catalog_path, data_dir)
def test_an_unknown_tenant_is_rejected(self):
catalog_path, data_dir = self._catalog_dir(
[{"document_id": "D1", "path": "a.md", "tenant_id": "TENANT-DOES-NOT-EXIST", "title": "A", "classification": "internal", "owner": "x"}]
)
with self.assertRaises(ValueError):
load_corpus(catalog_path, data_dir)
def test_a_missing_required_field_is_rejected(self):
catalog_path, data_dir = self._catalog_dir([{"document_id": "D1", "path": "a.md", "tenant_id": "TENANT-ALPHA"}])
with self.assertRaises(ValueError):
load_corpus(catalog_path, data_dir)
def test_duplicate_document_ids_are_rejected(self):
catalog_path, data_dir = self._catalog_dir(
[
{"document_id": "D1", "path": "a.md", "tenant_id": "TENANT-ALPHA", "title": "A", "classification": "internal", "owner": "x"},
{"document_id": "D1", "path": "a.md", "tenant_id": "TENANT-ALPHA", "title": "A2", "classification": "internal", "owner": "x"},
]
)
with self.assertRaises(ValueError):
load_corpus(catalog_path, data_dir)
if __name__ == "__main__":
unittest.main()
tests/context.pyTest setup and a deterministic stand-in embedder (tests only).
"""Shared test setup: import paths and a deterministic stand-in embedder,
so tests need no model or network.
"""
from __future__ import annotations
import hashlib
import os
import re
import sys
LAB = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, LAB)
import bootstrap # noqa: E402,F401
import settings # noqa: E402
from chunker import chunk_document # noqa: E402
from corpus import load_corpus # noqa: E402
from vector_store import FlatVectorStore, StoredChunk # noqa: E402
class FakeEmbedder:
dimension = 64
def _vector(self, text: str) -> list[float]:
vector = [0.0] * self.dimension
for token in re.findall(r"[a-z0-9.]+", text.lower()):
vector[int(hashlib.sha256(token.encode()).hexdigest(), 16) % self.dimension] += 1.0
if not any(vector):
vector[0] = 1.0
return vector
def embed_documents(self, texts: list[str]) -> list[list[float]]:
return [self._vector(text) for text in texts]
def embed_query(self, text: str) -> list[float]:
return self._vector(text)
def build_store(*, embedder=None) -> FlatVectorStore:
embedder = embedder or FakeEmbedder()
documents = load_corpus()
store = FlatVectorStore(embedder.dimension)
for document in documents:
chunks = chunk_document(document.document_id, document.text, settings.CHUNK_SIZE_CHARS, settings.CHUNK_OVERLAP_CHARS)
for chunk, vector in zip(chunks, embedder.embed_documents([c.text for c in chunks])):
store.add(
StoredChunk(
vector_id=chunk.chunk_id,
tenant_id=document.tenant_id,
document_id=document.document_id,
chunk_id=chunk.chunk_id,
title=document.title,
classification=document.classification,
owner=document.owner,
text=chunk.text,
),
vector,
)
return store
Evidence
Machine-readable output of the published run.
evidence/summary.jsonCompact machine-readable summary of the published run, including per-mode invariant_violations.
{
"lab_id": "ASL-008",
"run_at": "2026-09-23T13:28:40.331678+00:00",
"source_revision": {
"git_commit": "71af5ee6e1740e8d9f53f5ccf324705fb82b5e29",
"git_dirty": false,
"git_dirty_paths": [],
"git_checked_paths": [
"labs/asl-008-vector-database-authorization-failure",
"labs/shared"
],
"git_excluded_paths": [
"labs/asl-008-vector-database-authorization-failure/evidence",
"labs/asl-008-vector-database-authorization-failure/index"
]
},
"base_url": "http://127.0.0.1:8794",
"index_config": {
"manifest_version": 1,
"lab_id": "ASL-008",
"index_timestamp": "2026-09-23T10:29:03.579856+00:00",
"embedding": {
"model": "nomic-embed-text",
"dimension": 768,
"query_prefix": "search_query: ",
"document_prefix": "search_document: "
},
"vector_backend": "FlatVectorStore (in-process exact brute-force search, JSON persistence)",
"similarity_metric": "cosine (vectors L2-normalized; score = dot product)",
"chunking": {
"strategy": "fixed-size character windows, snapped back to whitespace",
"chunk_size_chars": 600,
"chunk_overlap_chars": 100
},
"documents": [
{
"document_id": "ALPHA-001",
"path": "tenant-alpha/ALPHA-001.md",
"tenant_id": "TENANT-ALPHA",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text_sha256": "71410d4010f68c30b003389f49241b4db0a282d6d0471ee84dcd7c9d18f61fc6",
"chunk_ids": [
"ALPHA-001-C01",
"ALPHA-001-C02"
]
},
{
"document_id": "BETA-001",
"path": "tenant-beta/BETA-001.md",
"tenant_id": "TENANT-BETA",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text_sha256": "0b082b5c17b13b6592eb51c296c5e6d0e6786cad8e09ca938333a2fa04e1de05",
"chunk_ids": [
"BETA-001-C01",
"BETA-001-C02"
]
},
{
"document_id": "GAMMA-001",
"path": "tenant-gamma/GAMMA-001.md",
"tenant_id": "TENANT-GAMMA",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text_sha256": "d8e05c9f9db424b6d2721ae4bd43cf982c0139d7999bdad5586714234f151ba7",
"chunk_ids": [
"GAMMA-001-C01",
"GAMMA-001-C02"
]
}
]
},
"summary_by_mode": {
"vulnerable": {
"adversarial_cases": 6,
"authorized_controls": 2,
"cross_tenant_searches_accepted": 3,
"cross_tenant_searches_total": 4,
"direct_unauthorized_object_fetches": 2,
"direct_object_fetches_total": 2,
"unauthorized_records_returned": 13,
"unauthorized_canaries_exposed": [
"BETA-CANARY-ASL008-2202",
"GAMMA-CANARY-ASL008-3303"
],
"global_searches_occurred": 1,
"policy_denied_count": 0,
"authorized_controls_passed": 2,
"invariant_violations": [
{
"test_id": "VDA-01",
"unauthorized_record_ids": [
"BETA-001-C02",
"BETA-001-C01"
]
},
{
"test_id": "VDA-02",
"unauthorized_record_ids": [
"BETA-001-C01"
]
},
{
"test_id": "VDA-03",
"unauthorized_record_ids": [
"BETA-001-C01",
"BETA-001-C02"
]
},
{
"test_id": "VDA-04",
"unauthorized_record_ids": [
"BETA-001-C02",
"BETA-001-C01"
]
},
{
"test_id": "VDA-05",
"unauthorized_record_ids": [
"GAMMA-001-C02",
"BETA-001-C02",
"GAMMA-001-C01",
"BETA-001-C01"
]
},
{
"test_id": "VDA-06",
"unauthorized_record_ids": [
"GAMMA-001-C02",
"GAMMA-001-C01"
]
}
]
},
"hardened": {
"adversarial_cases": 6,
"authorized_controls": 2,
"cross_tenant_searches_accepted": 0,
"cross_tenant_searches_total": 4,
"direct_unauthorized_object_fetches": 0,
"direct_object_fetches_total": 2,
"unauthorized_records_returned": 0,
"unauthorized_canaries_exposed": [],
"global_searches_occurred": 0,
"policy_denied_count": 5,
"authorized_controls_passed": 2,
"invariant_violations": []
}
}
}evidence/index-manifest.jsonEvery chunk in the one shared index, with its tenant_id, classification and owner.
Raw evidence used by the published report.
{
"manifest_version": 1,
"lab_id": "ASL-008",
"index_timestamp": "2026-09-23T10:29:03.579856+00:00",
"embedding": {
"model": "nomic-embed-text",
"dimension": 768,
"query_prefix": "search_query: ",
"document_prefix": "search_document: "
},
"vector_backend": "FlatVectorStore (in-process exact brute-force search, JSON persistence)",
"similarity_metric": "cosine (vectors L2-normalized; score = dot product)",
"chunking": {
"strategy": "fixed-size character windows, snapped back to whitespace",
"chunk_size_chars": 600,
"chunk_overlap_chars": 100
},
"documents": [
{
"document_id": "ALPHA-001",
"path": "tenant-alpha/ALPHA-001.md",
"tenant_id": "TENANT-ALPHA",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text_sha256": "71410d4010f68c30b003389f49241b4db0a282d6d0471ee84dcd7c9d18f61fc6",
"chunk_ids": [
"ALPHA-001-C01",
"ALPHA-001-C02"
]
},
{
"document_id": "BETA-001",
"path": "tenant-beta/BETA-001.md",
"tenant_id": "TENANT-BETA",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text_sha256": "0b082b5c17b13b6592eb51c296c5e6d0e6786cad8e09ca938333a2fa04e1de05",
"chunk_ids": [
"BETA-001-C01",
"BETA-001-C02"
]
},
{
"document_id": "GAMMA-001",
"path": "tenant-gamma/GAMMA-001.md",
"tenant_id": "TENANT-GAMMA",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text_sha256": "d8e05c9f9db424b6d2721ae4bd43cf982c0139d7999bdad5586714234f151ba7",
"chunk_ids": [
"GAMMA-001-C01",
"GAMMA-001-C02"
]
}
],
"chunks": [
{
"vector_id": "ALPHA-001-C01",
"chunk_id": "ALPHA-001-C01",
"document_id": "ALPHA-001",
"tenant_id": "TENANT-ALPHA",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"char_start": 0,
"char_end": 594,
"text_sha256": "d0b3af032275a374a05dd5dffb1f60ba785bef25ac8355d0f24933e797921591",
"text": "# Alpha Quarterly Planning Summary\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Alpha, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: ALPHA-CANARY-ASL008-1101\n\nTenant Alpha's quarterly planning cycle sets budget and headcount targets\nfor each business unit ahead of the next fiscal quarter. This cycle's\npriorities are cloud infrastructure consolidation, a refresh of the\ncustomer support tooling, and a 6% increase in the engineering budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two"
},
{
"vector_id": "ALPHA-001-C02",
"chunk_id": "ALPHA-001-C02",
"document_id": "ALPHA-001",
"tenant_id": "TENANT-ALPHA",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"char_start": 496,
"char_end": 890,
"text_sha256": "995dfcf72351c0ee52d6cd52974bdb04bf3e27ede701a04c6d8ed0c6452928f5",
"text": "budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two weeks before the quarterly\nreview; finance reconciles the submissions against the approved top-line\nbudget before the plans are finalized. Any request above the standard\ndiscretionary threshold requires sign-off from the Tenant Alpha finance\nlead before it can be included in the finalized plan."
},
{
"vector_id": "BETA-001-C01",
"chunk_id": "BETA-001-C01",
"document_id": "BETA-001",
"tenant_id": "TENANT-BETA",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"char_start": 0,
"char_end": 599,
"text_sha256": "d5c0cd890d97e8c859f1117e5b1b1ab630fda51017f162d15cda4ee456db6d50",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent"
},
{
"vector_id": "BETA-001-C02",
"chunk_id": "BETA-001-C02",
"document_id": "BETA-001",
"tenant_id": "TENANT-BETA",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"char_start": 499,
"char_end": 853,
"text_sha256": "a12efb98b7e9d51463c7b360613c01bb6e6b8efe6e4f5154c7deb84103c399c1",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made."
},
{
"vector_id": "GAMMA-001-C01",
"chunk_id": "GAMMA-001-C01",
"document_id": "GAMMA-001",
"tenant_id": "TENANT-GAMMA",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"char_start": 0,
"char_end": 599,
"text_sha256": "a0282ac2320d143f2ce433ffa81bd7604015f123672da48710807f56cbfd3228",
"text": "# Gamma Personnel Planning Notes\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Gamma, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: GAMMA-CANARY-ASL008-3303\n\nTenant Gamma's personnel planning notes for the coming quarter cover\nheadcount requests for the fictional Gamma support organization, a planned\nreorganization of the regional support teams, and updated on-call staffing\ntargets.\n\nIndividual compensation and performance details are intentionally excluded\nfrom this planning note; it covers only headcount counts, role\nrequisitions,"
},
{
"vector_id": "GAMMA-001-C02",
"chunk_id": "GAMMA-001-C02",
"document_id": "GAMMA-001",
"tenant_id": "TENANT-GAMMA",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"char_start": 500,
"char_end": 743,
"text_sha256": "1fcf8907eade37d1691f28c2f2fa4e3e1062a42f54134d2ffd783ed259cdc29f",
"text": "intentionally excluded\nfrom this planning note; it covers only headcount counts, role\nrequisitions, and staffing ratios. Personnel planning decisions are\nreviewed by the Tenant Gamma people-operations lead before headcount\nrequests are opened."
}
]
}
evidence/latest.jsonEvery request, response, authorization decision and independently-measured field from the published run.
Raw evidence used by the published report.
{
"lab_id": "ASL-008",
"run_at": "2026-09-23T13:28:40.331678+00:00",
"source_revision": {
"git_commit": "71af5ee6e1740e8d9f53f5ccf324705fb82b5e29",
"git_dirty": false,
"git_dirty_paths": [],
"git_checked_paths": [
"labs/asl-008-vector-database-authorization-failure",
"labs/shared"
],
"git_excluded_paths": [
"labs/asl-008-vector-database-authorization-failure/evidence",
"labs/asl-008-vector-database-authorization-failure/index"
]
},
"base_url": "http://127.0.0.1:8794",
"index_config": {
"manifest_version": 1,
"lab_id": "ASL-008",
"index_timestamp": "2026-09-23T10:29:03.579856+00:00",
"embedding": {
"model": "nomic-embed-text",
"dimension": 768,
"query_prefix": "search_query: ",
"document_prefix": "search_document: "
},
"vector_backend": "FlatVectorStore (in-process exact brute-force search, JSON persistence)",
"similarity_metric": "cosine (vectors L2-normalized; score = dot product)",
"chunking": {
"strategy": "fixed-size character windows, snapped back to whitespace",
"chunk_size_chars": 600,
"chunk_overlap_chars": 100
},
"documents": [
{
"document_id": "ALPHA-001",
"path": "tenant-alpha/ALPHA-001.md",
"tenant_id": "TENANT-ALPHA",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text_sha256": "71410d4010f68c30b003389f49241b4db0a282d6d0471ee84dcd7c9d18f61fc6",
"chunk_ids": [
"ALPHA-001-C01",
"ALPHA-001-C02"
]
},
{
"document_id": "BETA-001",
"path": "tenant-beta/BETA-001.md",
"tenant_id": "TENANT-BETA",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text_sha256": "0b082b5c17b13b6592eb51c296c5e6d0e6786cad8e09ca938333a2fa04e1de05",
"chunk_ids": [
"BETA-001-C01",
"BETA-001-C02"
]
},
{
"document_id": "GAMMA-001",
"path": "tenant-gamma/GAMMA-001.md",
"tenant_id": "TENANT-GAMMA",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text_sha256": "d8e05c9f9db424b6d2721ae4bd43cf982c0139d7999bdad5586714234f151ba7",
"chunk_ids": [
"GAMMA-001-C01",
"GAMMA-001-C02"
]
}
]
},
"summary_by_mode": {
"vulnerable": {
"adversarial_cases": 6,
"authorized_controls": 2,
"cross_tenant_searches_accepted": 3,
"cross_tenant_searches_total": 4,
"direct_unauthorized_object_fetches": 2,
"direct_object_fetches_total": 2,
"unauthorized_records_returned": 13,
"unauthorized_canaries_exposed": [
"BETA-CANARY-ASL008-2202",
"GAMMA-CANARY-ASL008-3303"
],
"global_searches_occurred": 1,
"policy_denied_count": 0,
"authorized_controls_passed": 2,
"invariant_violations": [
{
"test_id": "VDA-01",
"unauthorized_record_ids": [
"BETA-001-C02",
"BETA-001-C01"
]
},
{
"test_id": "VDA-02",
"unauthorized_record_ids": [
"BETA-001-C01"
]
},
{
"test_id": "VDA-03",
"unauthorized_record_ids": [
"BETA-001-C01",
"BETA-001-C02"
]
},
{
"test_id": "VDA-04",
"unauthorized_record_ids": [
"BETA-001-C02",
"BETA-001-C01"
]
},
{
"test_id": "VDA-05",
"unauthorized_record_ids": [
"GAMMA-001-C02",
"BETA-001-C02",
"GAMMA-001-C01",
"BETA-001-C01"
]
},
{
"test_id": "VDA-06",
"unauthorized_record_ids": [
"GAMMA-001-C02",
"GAMMA-001-C01"
]
}
]
},
"hardened": {
"adversarial_cases": 6,
"authorized_controls": 2,
"cross_tenant_searches_accepted": 0,
"cross_tenant_searches_total": 4,
"direct_unauthorized_object_fetches": 0,
"direct_object_fetches_total": 2,
"unauthorized_records_returned": 0,
"unauthorized_canaries_exposed": [],
"global_searches_occurred": 0,
"policy_denied_count": 5,
"authorized_controls_passed": 2,
"invariant_violations": []
}
},
"results_by_mode": {
"vulnerable": [
{
"test_id": "VDA-01",
"category": "cross_tenant_search",
"round": 1,
"is_attack": true,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "acquisition planning",
"tenant_id": "TENANT-BETA"
},
"timestamp": "2026-09-23T13:28:31.166713+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": true,
"unauthorized_record_returned": true,
"unauthorized_text_exposed": true,
"unauthorized_metadata_exposed": true,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-BETA",
"TENANT-BETA"
],
"record_ids_returned": [
"BETA-001-C02",
"BETA-001-C01"
],
"unauthorized_record_ids": [
"BETA-001-C02",
"BETA-001-C01"
],
"protected_markers": [
"BETA-CANARY-ASL008-2202"
],
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"requested_tenant_id": "TENANT-BETA",
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"results": [
{
"vector_id": "BETA-001-C02",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C02",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made.",
"score": 0.7503
},
{
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent",
"score": 0.6714
}
]
}
},
{
"test_id": "VDA-02",
"category": "cross_tenant_direct_vector_fetch",
"round": 1,
"is_attack": true,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "vector",
"request": {
"vector_id": "BETA-001-C01"
},
"timestamp": "2026-09-23T13:28:31.502000+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": null,
"unauthorized_record_returned": true,
"unauthorized_text_exposed": true,
"unauthorized_metadata_exposed": true,
"direct_object_access_bypass": true,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-BETA"
],
"record_ids_returned": [
"BETA-001-C01"
],
"unauthorized_record_ids": [
"BETA-001-C01"
],
"protected_markers": [
"BETA-CANARY-ASL008-2202"
],
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"vector_id": "BETA-001-C01",
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"record": {
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent"
}
}
},
{
"test_id": "VDA-03",
"category": "cross_tenant_document_chunks",
"round": 1,
"is_attack": true,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "chunks",
"request": {
"document_id": "BETA-001"
},
"timestamp": "2026-09-23T13:28:31.503239+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": null,
"unauthorized_record_returned": true,
"unauthorized_text_exposed": true,
"unauthorized_metadata_exposed": true,
"direct_object_access_bypass": true,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-BETA",
"TENANT-BETA"
],
"record_ids_returned": [
"BETA-001-C01",
"BETA-001-C02"
],
"unauthorized_record_ids": [
"BETA-001-C01",
"BETA-001-C02"
],
"protected_markers": [
"BETA-CANARY-ASL008-2202"
],
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"document_id": "BETA-001",
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"chunks": [
{
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent"
},
{
"vector_id": "BETA-001-C02",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C02",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made."
}
]
}
},
{
"test_id": "VDA-04",
"category": "filter_override",
"round": 1,
"is_attack": true,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "planning",
"filter": {
"classification": "confidential",
"tenant_id": "TENANT-BETA"
}
},
"timestamp": "2026-09-23T13:28:31.504242+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": true,
"unauthorized_record_returned": true,
"unauthorized_text_exposed": true,
"unauthorized_metadata_exposed": true,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-BETA",
"TENANT-BETA"
],
"record_ids_returned": [
"BETA-001-C02",
"BETA-001-C01"
],
"unauthorized_record_ids": [
"BETA-001-C02",
"BETA-001-C01"
],
"protected_markers": [
"BETA-CANARY-ASL008-2202"
],
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"requested_tenant_id": null,
"requested_filter": {
"classification": "confidential",
"tenant_id": "TENANT-BETA"
},
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"results": [
{
"vector_id": "BETA-001-C02",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C02",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made.",
"score": 0.6317
},
{
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent",
"score": 0.6071
}
]
}
},
{
"test_id": "VDA-05",
"category": "missing_tenant_filter",
"round": 1,
"is_attack": true,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "planning"
},
"timestamp": "2026-09-23T13:28:31.577889+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": true,
"unauthorized_text_exposed": true,
"unauthorized_metadata_exposed": true,
"direct_object_access_bypass": null,
"global_search_occurred": true,
"policy_denied": false,
"record_tenants": [
"TENANT-ALPHA",
"TENANT-GAMMA",
"TENANT-ALPHA",
"TENANT-BETA",
"TENANT-GAMMA",
"TENANT-BETA"
],
"record_ids_returned": [
"ALPHA-001-C01",
"GAMMA-001-C02",
"ALPHA-001-C02",
"BETA-001-C02",
"GAMMA-001-C01",
"BETA-001-C01"
],
"unauthorized_record_ids": [
"GAMMA-001-C02",
"BETA-001-C02",
"GAMMA-001-C01",
"BETA-001-C01"
],
"protected_markers": [
"GAMMA-CANARY-ASL008-3303",
"BETA-CANARY-ASL008-2202"
],
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"requested_tenant_id": null,
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"results": [
{
"vector_id": "ALPHA-001-C01",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C01",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "# Alpha Quarterly Planning Summary\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Alpha, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: ALPHA-CANARY-ASL008-1101\n\nTenant Alpha's quarterly planning cycle sets budget and headcount targets\nfor each business unit ahead of the next fiscal quarter. This cycle's\npriorities are cloud infrastructure consolidation, a refresh of the\ncustomer support tooling, and a 6% increase in the engineering budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two",
"score": 0.6882
},
{
"vector_id": "GAMMA-001-C02",
"tenant_id": "TENANT-GAMMA",
"document_id": "GAMMA-001",
"chunk_id": "GAMMA-001-C02",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text": "intentionally excluded\nfrom this planning note; it covers only headcount counts, role\nrequisitions, and staffing ratios. Personnel planning decisions are\nreviewed by the Tenant Gamma people-operations lead before headcount\nrequests are opened.",
"score": 0.6842
},
{
"vector_id": "ALPHA-001-C02",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C02",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two weeks before the quarterly\nreview; finance reconciles the submissions against the approved top-line\nbudget before the plans are finalized. Any request above the standard\ndiscretionary threshold requires sign-off from the Tenant Alpha finance\nlead before it can be included in the finalized plan.",
"score": 0.6376
},
{
"vector_id": "BETA-001-C02",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C02",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made.",
"score": 0.6317
},
{
"vector_id": "GAMMA-001-C01",
"tenant_id": "TENANT-GAMMA",
"document_id": "GAMMA-001",
"chunk_id": "GAMMA-001-C01",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text": "# Gamma Personnel Planning Notes\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Gamma, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: GAMMA-CANARY-ASL008-3303\n\nTenant Gamma's personnel planning notes for the coming quarter cover\nheadcount requests for the fictional Gamma support organization, a planned\nreorganization of the regional support teams, and updated on-call staffing\ntargets.\n\nIndividual compensation and performance details are intentionally excluded\nfrom this planning note; it covers only headcount counts, role\nrequisitions,",
"score": 0.614
},
{
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent",
"score": 0.6071
}
]
}
},
{
"test_id": "VDA-06",
"category": "unauthorized_real_tenant",
"round": 1,
"is_attack": true,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "personnel planning",
"tenant_id": "TENANT-GAMMA"
},
"timestamp": "2026-09-23T13:28:32.940172+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": true,
"unauthorized_record_returned": true,
"unauthorized_text_exposed": true,
"unauthorized_metadata_exposed": true,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-GAMMA",
"TENANT-GAMMA"
],
"record_ids_returned": [
"GAMMA-001-C02",
"GAMMA-001-C01"
],
"unauthorized_record_ids": [
"GAMMA-001-C02",
"GAMMA-001-C01"
],
"protected_markers": [
"GAMMA-CANARY-ASL008-3303"
],
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"requested_tenant_id": "TENANT-GAMMA",
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"results": [
{
"vector_id": "GAMMA-001-C02",
"tenant_id": "TENANT-GAMMA",
"document_id": "GAMMA-001",
"chunk_id": "GAMMA-001-C02",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text": "intentionally excluded\nfrom this planning note; it covers only headcount counts, role\nrequisitions, and staffing ratios. Personnel planning decisions are\nreviewed by the Tenant Gamma people-operations lead before headcount\nrequests are opened.",
"score": 0.6892
},
{
"vector_id": "GAMMA-001-C01",
"tenant_id": "TENANT-GAMMA",
"document_id": "GAMMA-001",
"chunk_id": "GAMMA-001-C01",
"title": "Gamma Personnel Planning Notes",
"classification": "internal",
"owner": "gamma-people-ops-lead@meridian-demo.example",
"text": "# Gamma Personnel Planning Notes\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Gamma, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: GAMMA-CANARY-ASL008-3303\n\nTenant Gamma's personnel planning notes for the coming quarter cover\nheadcount requests for the fictional Gamma support organization, a planned\nreorganization of the regional support teams, and updated on-call staffing\ntargets.\n\nIndividual compensation and performance details are intentionally excluded\nfrom this planning note; it covers only headcount counts, role\nrequisitions,",
"score": 0.6741
}
]
}
},
{
"test_id": "VDA-07",
"category": "authorized_search_control",
"round": 1,
"is_attack": false,
"mode": "vulnerable",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "quarterly planning budget",
"tenant_id": "TENANT-ALPHA"
},
"timestamp": "2026-09-23T13:28:34.865524+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-ALPHA",
"TENANT-ALPHA"
],
"record_ids_returned": [
"ALPHA-001-C01",
"ALPHA-001-C02"
],
"unauthorized_record_ids": [],
"protected_markers": [],
"expected_document_present": true,
"only_authorized_tenant_returned": true,
"authorized_functionality_preserved": true,
"raw_response": {
"mode": "vulnerable",
"identity": "ALICE",
"requested_tenant_id": "TENANT-ALPHA",
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"results": [
{
"vector_id": "ALPHA-001-C01",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C01",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "# Alpha Quarterly Planning Summary\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Alpha, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: ALPHA-CANARY-ASL008-1101\n\nTenant Alpha's quarterly planning cycle sets budget and headcount targets\nfor each business unit ahead of the next fiscal quarter. This cycle's\npriorities are cloud infrastructure consolidation, a refresh of the\ncustomer support tooling, and a 6% increase in the engineering budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two",
"score": 0.7061
},
{
"vector_id": "ALPHA-001-C02",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C02",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two weeks before the quarterly\nreview; finance reconciles the submissions against the approved top-line\nbudget before the plans are finalized. Any request above the standard\ndiscretionary threshold requires sign-off from the Tenant Alpha finance\nlead before it can be included in the finalized plan.",
"score": 0.6572
}
]
}
},
{
"test_id": "VDA-08",
"category": "second_tenant_control",
"round": 1,
"is_attack": false,
"mode": "vulnerable",
"authenticated_user": "BOB",
"authorized_tenants": [
"TENANT-BETA"
],
"endpoint": "search",
"request": {
"query": "acquisition planning",
"tenant_id": "TENANT-BETA"
},
"timestamp": "2026-09-23T13:28:34.946809+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-BETA",
"TENANT-BETA"
],
"record_ids_returned": [
"BETA-001-C02",
"BETA-001-C01"
],
"unauthorized_record_ids": [],
"protected_markers": [],
"expected_document_present": true,
"only_authorized_tenant_returned": true,
"authorized_functionality_preserved": true,
"raw_response": {
"mode": "vulnerable",
"identity": "BOB",
"requested_tenant_id": "TENANT-BETA",
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": null,
"status_code": 200,
"event": "AUTHORIZATION_NOT_ENFORCED",
"reason": "vulnerable mode trusts caller-supplied scope; no server-side check performed"
},
"results": [
{
"vector_id": "BETA-001-C02",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C02",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made.",
"score": 0.7503
},
{
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent",
"score": 0.6714
}
]
}
}
],
"hardened": [
{
"test_id": "VDA-01",
"category": "cross_tenant_search",
"round": 1,
"is_attack": true,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "acquisition planning",
"tenant_id": "TENANT-BETA"
},
"timestamp": "2026-09-23T13:28:36.888448+00:00",
"http_status": 403,
"authorization_decision": {
"allowed": false,
"scope": [],
"status_code": 403,
"event": "CROSS_TENANT_SCOPE_ATTEMPT",
"reason": "ALICE requested TENANT-BETA, authorized for ['TENANT-ALPHA']"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": true,
"record_tenants": [],
"record_ids_returned": [],
"unauthorized_record_ids": [],
"protected_markers": [],
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"requested_tenant_id": "TENANT-BETA",
"requested_filter": null,
"authorization_decision": {
"allowed": false,
"scope": [],
"status_code": 403,
"event": "CROSS_TENANT_SCOPE_ATTEMPT",
"reason": "ALICE requested TENANT-BETA, authorized for ['TENANT-ALPHA']"
},
"results": []
}
},
{
"test_id": "VDA-02",
"category": "cross_tenant_direct_vector_fetch",
"round": 1,
"is_attack": true,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "vector",
"request": {
"vector_id": "BETA-001-C01"
},
"timestamp": "2026-09-23T13:28:36.889431+00:00",
"http_status": 404,
"authorization_decision": {
"allowed": false,
"status_code": 404,
"event": "DIRECT_OBJECT_ACCESS_DENIED",
"reason": "ALICE is not authorized for tenant TENANT-BETA"
},
"unauthorized_search_accepted": null,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": false,
"global_search_occurred": null,
"policy_denied": true,
"record_tenants": [],
"record_ids_returned": [],
"unauthorized_record_ids": [],
"protected_markers": [],
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"vector_id": "BETA-001-C01",
"authorization_decision": {
"allowed": false,
"status_code": 404,
"event": "DIRECT_OBJECT_ACCESS_DENIED",
"reason": "ALICE is not authorized for tenant TENANT-BETA"
},
"record": null
}
},
{
"test_id": "VDA-03",
"category": "cross_tenant_document_chunks",
"round": 1,
"is_attack": true,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "chunks",
"request": {
"document_id": "BETA-001"
},
"timestamp": "2026-09-23T13:28:36.890253+00:00",
"http_status": 404,
"authorization_decision": {
"allowed": false,
"status_code": 404,
"event": "DIRECT_OBJECT_ACCESS_DENIED",
"reason": "ALICE is not authorized for tenant TENANT-BETA"
},
"unauthorized_search_accepted": null,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": false,
"global_search_occurred": null,
"policy_denied": true,
"record_tenants": [],
"record_ids_returned": [],
"unauthorized_record_ids": [],
"protected_markers": [],
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"document_id": "BETA-001",
"authorization_decision": {
"allowed": false,
"status_code": 404,
"event": "DIRECT_OBJECT_ACCESS_DENIED",
"reason": "ALICE is not authorized for tenant TENANT-BETA"
},
"chunks": []
}
},
{
"test_id": "VDA-04",
"category": "filter_override",
"round": 1,
"is_attack": true,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "planning",
"filter": {
"classification": "confidential",
"tenant_id": "TENANT-BETA"
}
},
"timestamp": "2026-09-23T13:28:36.890850+00:00",
"http_status": 403,
"authorization_decision": {
"allowed": false,
"scope": [],
"status_code": 403,
"event": "UNAUTHORIZED_FILTER_REJECTED",
"reason": "filter references tenant(s) ['TENANT-BETA'] outside ALICE's authorized scope"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": true,
"record_tenants": [],
"record_ids_returned": [],
"unauthorized_record_ids": [],
"protected_markers": [],
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"requested_tenant_id": null,
"requested_filter": {
"classification": "confidential",
"tenant_id": "TENANT-BETA"
},
"authorization_decision": {
"allowed": false,
"scope": [],
"status_code": 403,
"event": "UNAUTHORIZED_FILTER_REJECTED",
"reason": "filter references tenant(s) ['TENANT-BETA'] outside ALICE's authorized scope"
},
"results": []
}
},
{
"test_id": "VDA-05",
"category": "missing_tenant_filter",
"round": 1,
"is_attack": true,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "planning"
},
"timestamp": "2026-09-23T13:28:36.891801+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": [
"TENANT-ALPHA"
],
"status_code": 200,
"event": "GLOBAL_SCOPE_DEFAULTED_TO_IDENTITY",
"reason": "no tenant_id supplied; scoped to ALICE's authorized set ['TENANT-ALPHA'], not every tenant"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": false,
"policy_denied": false,
"record_tenants": [
"TENANT-ALPHA",
"TENANT-ALPHA"
],
"record_ids_returned": [
"ALPHA-001-C01",
"ALPHA-001-C02"
],
"unauthorized_record_ids": [],
"protected_markers": [],
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"requested_tenant_id": null,
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": [
"TENANT-ALPHA"
],
"status_code": 200,
"event": "GLOBAL_SCOPE_DEFAULTED_TO_IDENTITY",
"reason": "no tenant_id supplied; scoped to ALICE's authorized set ['TENANT-ALPHA'], not every tenant"
},
"results": [
{
"vector_id": "ALPHA-001-C01",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C01",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "# Alpha Quarterly Planning Summary\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Alpha, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: ALPHA-CANARY-ASL008-1101\n\nTenant Alpha's quarterly planning cycle sets budget and headcount targets\nfor each business unit ahead of the next fiscal quarter. This cycle's\npriorities are cloud infrastructure consolidation, a refresh of the\ncustomer support tooling, and a 6% increase in the engineering budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two",
"score": 0.6882
},
{
"vector_id": "ALPHA-001-C02",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C02",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two weeks before the quarterly\nreview; finance reconciles the submissions against the approved top-line\nbudget before the plans are finalized. Any request above the standard\ndiscretionary threshold requires sign-off from the Tenant Alpha finance\nlead before it can be included in the finalized plan.",
"score": 0.6376
}
]
}
},
{
"test_id": "VDA-06",
"category": "unauthorized_real_tenant",
"round": 1,
"is_attack": true,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "personnel planning",
"tenant_id": "TENANT-GAMMA"
},
"timestamp": "2026-09-23T13:28:38.287043+00:00",
"http_status": 403,
"authorization_decision": {
"allowed": false,
"scope": [],
"status_code": 403,
"event": "CROSS_TENANT_SCOPE_ATTEMPT",
"reason": "ALICE requested TENANT-GAMMA, authorized for ['TENANT-ALPHA']"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": true,
"record_tenants": [],
"record_ids_returned": [],
"unauthorized_record_ids": [],
"protected_markers": [],
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"requested_tenant_id": "TENANT-GAMMA",
"requested_filter": null,
"authorization_decision": {
"allowed": false,
"scope": [],
"status_code": 403,
"event": "CROSS_TENANT_SCOPE_ATTEMPT",
"reason": "ALICE requested TENANT-GAMMA, authorized for ['TENANT-ALPHA']"
},
"results": []
}
},
{
"test_id": "VDA-07",
"category": "authorized_search_control",
"round": 1,
"is_attack": false,
"mode": "hardened",
"authenticated_user": "ALICE",
"authorized_tenants": [
"TENANT-ALPHA"
],
"endpoint": "search",
"request": {
"query": "quarterly planning budget",
"tenant_id": "TENANT-ALPHA"
},
"timestamp": "2026-09-23T13:28:38.288016+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": [
"TENANT-ALPHA"
],
"status_code": 200,
"event": "VECTOR_SEARCH_ALLOWED",
"reason": "scoped to requested tenant TENANT-ALPHA, within ALICE's authorized set"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-ALPHA",
"TENANT-ALPHA"
],
"record_ids_returned": [
"ALPHA-001-C01",
"ALPHA-001-C02"
],
"unauthorized_record_ids": [],
"protected_markers": [],
"expected_document_present": true,
"only_authorized_tenant_returned": true,
"authorized_functionality_preserved": true,
"raw_response": {
"mode": "hardened",
"identity": "ALICE",
"requested_tenant_id": "TENANT-ALPHA",
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": [
"TENANT-ALPHA"
],
"status_code": 200,
"event": "VECTOR_SEARCH_ALLOWED",
"reason": "scoped to requested tenant TENANT-ALPHA, within ALICE's authorized set"
},
"results": [
{
"vector_id": "ALPHA-001-C01",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C01",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "# Alpha Quarterly Planning Summary\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Alpha, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: ALPHA-CANARY-ASL008-1101\n\nTenant Alpha's quarterly planning cycle sets budget and headcount targets\nfor each business unit ahead of the next fiscal quarter. This cycle's\npriorities are cloud infrastructure consolidation, a refresh of the\ncustomer support tooling, and a 6% increase in the engineering budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two",
"score": 0.7061
},
{
"vector_id": "ALPHA-001-C02",
"tenant_id": "TENANT-ALPHA",
"document_id": "ALPHA-001",
"chunk_id": "ALPHA-001-C02",
"title": "Alpha Quarterly Planning Summary",
"classification": "internal",
"owner": "alpha-finance-lead@meridian-demo.example",
"text": "budget to\nsupport the new observability initiative.\n\nDepartment leads submit their draft plans two weeks before the quarterly\nreview; finance reconciles the submissions against the approved top-line\nbudget before the plans are finalized. Any request above the standard\ndiscretionary threshold requires sign-off from the Tenant Alpha finance\nlead before it can be included in the finalized plan.",
"score": 0.6572
}
]
}
},
{
"test_id": "VDA-08",
"category": "second_tenant_control",
"round": 1,
"is_attack": false,
"mode": "hardened",
"authenticated_user": "BOB",
"authorized_tenants": [
"TENANT-BETA"
],
"endpoint": "search",
"request": {
"query": "acquisition planning",
"tenant_id": "TENANT-BETA"
},
"timestamp": "2026-09-23T13:28:38.395501+00:00",
"http_status": 200,
"authorization_decision": {
"allowed": true,
"scope": [
"TENANT-BETA"
],
"status_code": 200,
"event": "VECTOR_SEARCH_ALLOWED",
"reason": "scoped to requested tenant TENANT-BETA, within BOB's authorized set"
},
"unauthorized_search_accepted": false,
"unauthorized_record_returned": false,
"unauthorized_text_exposed": false,
"unauthorized_metadata_exposed": false,
"direct_object_access_bypass": null,
"global_search_occurred": null,
"policy_denied": false,
"record_tenants": [
"TENANT-BETA",
"TENANT-BETA"
],
"record_ids_returned": [
"BETA-001-C02",
"BETA-001-C01"
],
"unauthorized_record_ids": [],
"protected_markers": [],
"expected_document_present": true,
"only_authorized_tenant_returned": true,
"authorized_functionality_preserved": true,
"raw_response": {
"mode": "hardened",
"identity": "BOB",
"requested_tenant_id": "TENANT-BETA",
"requested_filter": null,
"authorization_decision": {
"allowed": true,
"scope": [
"TENANT-BETA"
],
"status_code": 200,
"event": "VECTOR_SEARCH_ALLOWED",
"reason": "scoped to requested tenant TENANT-BETA, within BOB's authorized set"
},
"results": [
{
"vector_id": "BETA-001-C02",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C02",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent on the due diligence findings and confirmation\nthat the target's data handling practices meet Tenant Beta's compliance\nrequirements. Details of Project Harborlight are restricted to the Tenant\nBeta deal team until a public announcement, if any, is made.",
"score": 0.7503
},
{
"vector_id": "BETA-001-C01",
"tenant_id": "TENANT-BETA",
"document_id": "BETA-001",
"chunk_id": "BETA-001-C01",
"title": "Beta Acquisition Planning Brief",
"classification": "confidential",
"owner": "beta-deal-team@meridian-demo.example",
"text": "# Beta Acquisition Planning Brief\n\n> SYNTHETIC DOCUMENT. Meridian Demo Corporation, Tenant Beta, and everything\n> below are fictional, invented for the Shah AI Security Lab (ASL-008).\n\nMarker: BETA-CANARY-ASL008-2202\n\nTenant Beta is evaluating the acquisition of a small fictional logistics\nanalytics vendor, codenamed Project Harborlight, as part of this quarter's\nacquisition planning. Due diligence covers the target's technology stack,\ncustomer contract terms, and outstanding liabilities.\n\nThe current acquisition planning timeline targets a signed letter of intent\nwithin six weeks, contingent",
"score": 0.6714
}
]
}
}
]
}
}Configuration
Documentation and configuration templates.
README.mdPurpose, scope, architecture, API, authorization model, how to build the index, run the lab, harness and tests, and how to read the results.
# ASL-008 — Vector Database Authorization Failure
Status: see `src/content/labs/asl-008-vector-database-authorization-failure.md`
for the public write-up and current status.
## Purpose
Tests what happens when the vector-data interface itself — not the RAG
application layered on top of it — accepts caller-controlled scope
(namespace, tenant, document ID, metadata filter) without independently
enforcing authorization.
> A caller being able to specify `tenant=B` does not mean the caller is
> allowed to access Tenant B.
### Distinguishing this from ASL-005
- **ASL-005** (RAG data exfiltration): an application bug. A normal RAG
application performs global retrieval with no authorization check in its
own application code; the vector layer itself is not the thing being
tested.
- **ASL-008**: a vector-data/API authorization bug, one layer lower. A
caller manipulates tenant/namespace/document scope directly against the
vector service/API and receives unauthorized vector records — with or
without any application or LLM sitting on top of it. **The LLM is not
required to prove the primary vulnerability**: if unauthorized vector
records can be retrieved directly, authorization has already failed. See
"Why there is no LLM in the primary evidence" below.
## Threat model
An authenticated caller of a multi-tenant vector-data API supplies
scope-related parameters — a tenant ID, a namespace, a document ID, a
metadata filter — that the service trusts as authoritative instead of
deriving authorization from the caller's own identity. No credential
compromise is involved: every request in this lab uses a legitimate,
successfully-authenticated identity. The question is whether the service
independently checks that identity's actual entitlements before returning
data, or only checks that *a* request happened to be well-formed.
## Why there is no LLM in the primary evidence
Primary evidence comes directly from the vector API's own responses —
returned record identity, tenant metadata, and HTTP status — never from
model output. This lab intentionally has **no chat-completion calls at
all**: `vector_store/build_index.py` calls only the embedding model (to
build the search index), and the attack harness calls only the local HTTP
API. A downstream "vector API → LLM → answer" demonstration was considered
and deliberately left out as out-of-scope secondary impact — the
authorization vulnerability is complete and fully demonstrated without it,
and adding an LLM step would not change any of the eight measured fields
below.
## Lab scenario
Fictional **Meridian Demo Corporation**, three tenants:
- `TENANT-ALPHA` — one document, `ALPHA-001` (quarterly planning).
- `TENANT-BETA` — one document, `BETA-001` (acquisition planning,
classification `confidential`).
- `TENANT-GAMMA` — one document, `GAMMA-001` (personnel planning).
Five synthetic identities (`entitlements.py`), each mapped **server-side
only** to the tenant(s) it may access — never derived from anything a
request supplies:
| Identity | Authorized tenant(s) |
| --- | --- |
| ALICE | TENANT-ALPHA |
| BOB | TENANT-BETA |
| CAROL | TENANT-GAMMA |
| AUDITOR | TENANT-ALPHA, TENANT-BETA (set-based — proves the policy isn't a hardcoded single-tenant equality check) |
| DAVE | *(none — a known, authenticable identity with no entitlement record, modeling a provisioning gap)* |
Each document contains one synthetic canary (`ALPHA-CANARY-ASL008-1101`,
`BETA-CANARY-ASL008-2202`, `GAMMA-CANARY-ASL008-3303`) for deterministic
disclosure detection.
## Vector record model
Every stored chunk (`vector_store/vector_store.py::StoredChunk`) carries:
`vector_id`, `tenant_id`, `document_id`, `chunk_id`, `title`,
`classification`, `owner`, `text` — the fields listed in the lab spec,
implemented directly (no elaborate metadata wrapper).
## Local vector service
`app/server.py`, stdlib-only, bound to `127.0.0.1` only:
- `POST /search` — `{query, tenant_id?, filter?}`
- `GET /vectors/{vector_id}` — direct object fetch by ID
- `POST /fetch` — `{vector_ids: [...]}`, batch direct fetch
- `GET /documents/{document_id}/chunks` — every chunk of one document
Mode (`vulnerable` | `hardened`) and identity are chosen **per request**
(`X-Lab-Mode`, `X-Lab-User` headers) — one running server serves both
modes, same convention as ASL-005/006/007.
## Index architecture
**One shared physical vector index** across all three tenants
(`vector_store/build_index.py`), queried identically in every mode. This is
deliberate: the lab demonstrates that a shared index is not automatically
insecure *if authorization is enforced correctly elsewhere* — the failure
mode under test is missing server-side scope binding, not the index
topology. See the public write-up's "Shared vs Separate Indexes" for the
tradeoffs against a per-tenant index/collection (isolation, operability,
cost, blast radius) — neither is claimed to be universally superior.
## Vulnerable design
The server authenticates the caller (a valid `X-Lab-User` is still
required — see "Authentication vs Authorization" below) but trusts
caller-supplied scope directly: `requested_tenant = request.body.tenant_id`
is used to filter the search with no check that the caller's identity is
actually entitled to that tenant. Filters are applied the same way — any
`tenant_id` or `classification` clause in a caller-supplied `filter` is
ANDed into the query verbatim (`app/server.py::_vulnerable_predicate`). A
missing `tenant_id` means no tenant condition at all: a genuinely global
search across every tenant. Direct fetch (`GET /vectors/{id}`, `POST
/fetch`, `GET /documents/{id}/chunks`) returns a record merely because the
ID exists, regardless of which tenant it belongs to.
## Hardened design
`mitigation/authorization.py` is the one control the vulnerable path never
calls — the vulnerability is that module's *absence* from that code path,
not a bug inside it.
- **`resolve_search_scope`** — server-side entitlement binding. Looks up
the caller's authorized tenants from `entitlements.py` (never from the
request). A requested `tenant_id` or filter-referenced tenant must be a
member of that set or the whole request is rejected (403) — never
silently narrowed. No `tenant_id` at all defaults the scope to the
identity's *own* authorized tenants, never to every tenant that exists.
- **`authorize_object`** — object-level authorization for direct fetch: an
object's *actual* `tenant_id` (looked up after finding the record) is
checked against the caller's authorized set, independent of anything the
caller claimed. This is the boundary that stops "I happen to know the ID
`BETA-001-C01`" from being sufficient.
- **Fail-closed**: an identity with no entitlement record (DAVE) — or any
identity the server cannot resolve entitlements for — is denied
outright. It never defaults to global scope.
## Authentication vs authorization
Both modes require a recognized `X-Lab-User` (401 otherwise) — the
vulnerable path is not "no authentication," it is authenticated identity
with **no server-side authorization check on top of it**. That is the
lab's central point made structural: authentication answers *who is this*;
only the hardened path's `mitigation/authorization.py` answers *what may
they touch*.
## HTTP status policy
Documented once, applied consistently:
- **401** — unauthenticated: missing or unrecognized `X-Lab-User`.
- **403** — authenticated but forbidden: an explicit search request (a
`tenant_id` or filter) names a tenant outside the caller's authorized
set, or the caller has no entitlement record at all.
- **404** — direct object/vector/chunk access denied *or* the object
genuinely doesn't exist. These are intentionally indistinguishable by
status code: an object-level authorization boundary that concealed
*"exists, but not yours"* would still confirm cross-tenant IDs by
contrast with a genuine 404 for a made-up one.
- **200** — authorized; every returned record belongs to the caller's
authorized scope.
## Requirements
- Python 3.9+ (stdlib only — no `pip install`).
- An OpenAI-compatible backend serving an embedding model (tested with
local Ollama: `ollama pull nomic-embed-text`). No chat model is required.
## Configuration
Copy `.env.example` to `.env` and edit, or export the variables directly.
`LAB_PORT` is 8794 — ASL-001..007 use 8787..8793.
## Build the index
```sh
cd labs/asl-008-vector-database-authorization-failure
python3 vector_store/build_index.py # embeds 6 chunks from 3 documents
python3 vector_store/build_index.py --check # verify the manifest matches the corpus/settings; no model needed
```
## Run the lab
```sh
python3 app/server.py # http://127.0.0.1:8794
```
Pick a mode and identity, choose an endpoint, and send a request. The
response includes the server's own `authorization_decision` (event +
reason) alongside whatever records were returned.
## Run the attack harness
With the index built and the server running:
```sh
python3 attack-harness/harness.py
```
Runs `attack-harness/test_cases.json` (VDA-01 to VDA-08, fixed before the
first run) against both modes — pure HTTP calls to the local API, no LLM
involved — and writes `evidence/latest.json`, a timestamped
`evidence/run-<stamp>.json`, and `evidence/summary.json`. **Commit before
running** — the harness records the git commit and dirty flag it ran
against, read from git, never entered by hand.
## Run the tests
```sh
python3 -m unittest discover -s tests -v
```
34 tests; none needs the embedding model, a chat model, or a network (a
deterministic stand-in embedder in `tests/context.py` is used only for the
one test that needs a real search). Covers: every VDA-0x policy decision
directly (ALICE authorized for Alpha / denied Beta, BOB authorized for
Beta, the filter-override rejection), the AUDITOR set-based entitlement,
fail-closed behavior for DAVE's missing entitlement record and for
unauthenticated requests, object-level authorization, vector-store
mechanics, catalog validation, and the critical invariant (see below).
## The critical invariant
For every hardened-mode search, across every identity and every tenant it
is authorized for:
```
for every returned record: record.tenant_id IN identity.authorized_tenants
```
Asserted directly against the real corpus and store in
`tests/test_authorization.py::CriticalInvariant`, and checked independently
in every harness run (`evidence` records `invariant_violations` per mode —
empty for hardened in the published run).
## The eight measured fields
Tracked and reported **separately**, computed independently from the
server's own observable response (never from its self-reported
`authorization_decision`) — see `attack-harness/measure.py`:
`UNAUTHORIZED_SEARCH_ACCEPTED`, `UNAUTHORIZED_RECORD_RETURNED`,
`UNAUTHORIZED_TEXT_EXPOSED`, `UNAUTHORIZED_METADATA_EXPOSED`,
`DIRECT_OBJECT_ACCESS_BYPASS`, `GLOBAL_SEARCH_OCCURRED`, `POLICY_DENIED`,
`AUTHORIZED_FUNCTIONALITY_PRESERVED`.
## Test rounds
All eight cases were written and fixed in `test_cases.json` before the
first run. VDA-04 (filter override) was corrected once, before any run
against the server, after reasoning through the predicate logic showed the
original request (a top-level `tenant_id` combined with a *conflicting*
`filter.tenant_id`) would self-cancel in the vulnerable path rather than
demonstrate a filter-based bypass — fixed by removing the conflicting
top-level parameter so the filter is the sole scoping mechanism under test.
No case was changed after seeing a result from the running server.
## Important limitations
- Three tenants, one document each, five identities, eight fixed cases on
one round: this measures one architecture, not vector-database
authorization failures in general.
- The filter syntax this reference service accepts is a flat two-key
dict (`tenant_id`, `classification`) — deliberately not a general query
language; a real vector database's richer filter/query surface could have
additional bypass shapes this lab does not model.
- 404-for-concealment is a policy choice, not a proof that no timing or
other side-channel could distinguish "denied" from "doesn't exist" — see
README "HTTP status policy".
- No LLM step exists in this lab by design — see "Why there is no LLM in
the primary evidence."
## Safety warning
- The server binds to `127.0.0.1` only. Do not expose it.
- Every tenant, document, name and canary is synthetic and fictional.
- The application is never hosted; its source is published read-only for
review (see `labSourceCode` in `src/data/labs.ts`) — no file listed there
may ever contain a real credential.
## Cleanup
```sh
pkill -f "asl-008-vector-database-authorization-failure/app/[s]erver.py"
rm -rf index/ # derived; rebuilt by vector_store/build_index.py
rm -f evidence/request-log.jsonl # local request log
```
.env.exampleConfiguration template — contains no secrets.
# ASL-008 — Vector Database Authorization Failure lab configuration.
# Copy to .env and edit. Never commit a real .env file.
# OpenAI-compatible embeddings endpoint, used only to build the search
# index (this lab makes no chat-completion calls at all — see README "Why
# there is no LLM in the primary evidence").
# - Ollama (local): http://localhost:11434/v1
LLM_BASE_URL=http://localhost:11434/v1
# Leave blank for a local backend that doesn't require auth.
LLM_API_KEY=
# Same embedding model as ASL-005/006/007 (Ollama: `ollama pull nomic-embed-text`).
EMBEDDING_MODEL=nomic-embed-text
# vulnerable | hardened — only sets the UI's default at startup; the
# running server accepts either per request (X-Lab-Mode header).
LAB_MODE=vulnerable
# Local server port. ASL-001..007 use 8787..8793, so this lab uses 8794.
LAB_PORT=8794