security/authz_redteam.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Adversarial authorization red-team suite.
The invariant: NO token, prompt, or tool argument can widen a caller's grant. This
seeds a CONFIDENTIAL entity + dossier owned by the operator, then hammers every
read tool with a LOW-clearance token and asserts it can NEVER see the confidential
data — while a HIGH-clearance token can (control, so we know the fix didn't just
break everything). Exit non-zero on ANY leak → wire into CI / pre-deploy gate.
Env: GCTRL_API_URL (default http://localhost:4000/api), JWT_SECRET, OWNER (user uuid),
NEO_CONTAINER (default neo4j), NEO_PW.
"""
import hmac, hashlib, base64, json, os, subprocess, sys, time, urllib.request, urllib.error
API = os.environ.get("GCTRL_API_URL", "http://localhost:4000/api").rstrip("/")
SECRET = os.environ.get("JWT_SECRET", "borghive-dev-jwt-secret-change-in-production").encode()
OWNER = os.environ.get("OWNER", "e5b05d21-a730-4f8c-9d69-47aa9cfcb7ba")
NEO = os.environ.get("NEO_CONTAINER", "neo4j")
NEO_PW = os.environ.get("NEO_PW", "password")
SECRET_ENTITY = "RedteamSecret Zeta"
SECRET_NEIGH = "RedteamClassified Payload"
SECRET_MARKER = "TOPSECRET-CANARY-9f3a2b" # must never reach a low-clearance reader
CONF_RANK = 50 # the confidential fact's _min_rank
LOW_RANK = 10 # below CONF_RANK → must be denied
HIGH_RANK = 100 # above CONF_RANK → control, must succeed
def mint(rank):
def b(d): return base64.urlsafe_b64encode(d).rstrip(b"=").decode()
h = b(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
now = int(time.time())
# api_key_rank caps get_user_clearance_rank to `rank` — simulates a scoped key.
p = b(json.dumps({"sub": OWNER, "email": "redteam@test", "role": "user",
"clearance": "INTERNAL", "api_key_rank": rank,
"iat": now, "exp": now + 3600}).encode())
sig = b(hmac.new(SECRET, (h + "." + p).encode(), hashlib.sha256).digest())
return h + "." + p + "." + sig
def cy(stmt):
subprocess.run(["docker", "exec", NEO, "cypher-shell", "-u", "neo4j", "-p", NEO_PW,
"--format", "plain", stmt], capture_output=True, text=True)
def api(path, tok, method="GET", body=None):
data = json.dumps(body).encode() if body is not None else None
r = urllib.request.Request(f"{API}{path}", data=data, method=method)
r.add_header("Content-Type", "application/json")
r.add_header("Authorization", f"Bearer {tok}")
try:
with urllib.request.urlopen(r, timeout=60) as resp:
return resp.status, resp.read().decode()
except urllib.error.HTTPError as e:
return e.code, e.read().decode()
except Exception as e:
return 0, str(e)
def agent_tool(name, args, tok):
return api(f"/agent/tools/{name}", tok, "POST", args)
def seed():
print("seeding confidential entity + dossier ...")
cy(f"MERGE (a:Entity {{name:'{SECRET_ENTITY}', user_id:'{OWNER}'}}) "
f"SET a._owner='{OWNER}', a._min_rank={CONF_RANK}, a._source_job='redteam-job', a.label='project' ")
cy(f"MERGE (b:Entity {{name:'{SECRET_NEIGH}', user_id:'{OWNER}'}}) "
f"SET b._owner='{OWNER}', b._min_rank={CONF_RANK}, b._source_job='redteam-job', b.label='thing' ")
cy(f"MATCH (a:Entity {{name:'{SECRET_ENTITY}',user_id:'{OWNER}'}}), (b:Entity {{name:'{SECRET_NEIGH}',user_id:'{OWNER}'}}) "
f"MERGE (a)-[r:CONTAINS]->(b) SET r._owner='{OWNER}', r._min_rank={CONF_RANK} ")
# A dossier for the secret entity, with the canary in the summary.
ins = (f"INSERT INTO entity_dossiers (user_id, entity_uri, entity_name, summary, archived) "
f"VALUES ('{OWNER}', 'redteam:{SECRET_ENTITY}', '{SECRET_ENTITY}', "
f"'{SECRET_MARKER}: this dossier aggregates confidential facts.', false) "
f"ON CONFLICT (user_id, entity_uri) DO UPDATE SET summary=EXCLUDED.summary, archived=false;")
subprocess.run(["docker", "exec", "gctrl-postgres", "psql", "-U", "GCTRL", "-d", "GCTRL", "-c", ins],
capture_output=True, text=True)
def cleanup():
cy(f"MATCH (n) WHERE n._source_job='redteam-job' DETACH DELETE n")
subprocess.run(["docker", "exec", "gctrl-postgres", "psql", "-U", "GCTRL", "-d", "GCTRL", "-c",
f"DELETE FROM entity_dossiers WHERE entity_uri='redteam:{SECRET_ENTITY}';"],
capture_output=True, text=True)
def leaked(resp_text):
# A leak = confidential CONTENT the caller never supplied: the dossier canary or
# the classified neighbour. The requested entity NAME is often echoed back
# (e.g. get_neighbors' "entity" field, a RAG "not found" message) — that echo is
# not a leak, so we do NOT flag SECRET_ENTITY here.
return SECRET_MARKER in resp_text or SECRET_NEIGH in resp_text
def main():
seed()
low, high = mint(LOW_RANK), mint(HIGH_RANK)
fails, checks = [], []
# (label, call) → each returns (status, text)
probes = {
"get_dossier": lambda t: agent_tool("get_dossier", {"name": SECRET_ENTITY}, t),
"get_entity": lambda t: agent_tool("get_entity", {"name": SECRET_ENTITY}, t),
"search_entities": lambda t: agent_tool("search_entities", {"query": "Redteam"}, t),
"get_neighbors": lambda t: agent_tool("get_neighbors", {"name": SECRET_ENTITY}, t),
"query_rag": lambda t: api("/rag/query", t, "POST",
{"message": f"What is {SECRET_ENTITY}?", "mode": "incognito"}),
}
print("\n== LOW-clearance token MUST NOT leak ==")
for name, call in probes.items():
_, txt = call(low)
if leaked(txt):
fails.append(name); print(f" LEAK {name}: confidential data returned to low token!")
else:
print(f" ok {name}: denied")
checks.append(name)
print("\n== HIGH-clearance token control (should SEE it) ==")
seen_any = False
for name in ("get_dossier", "get_entity"):
_, txt = probes[name](high)
if leaked(txt):
seen_any = True; print(f" ok {name}: visible to high token (control)")
else:
print(f" WARN {name}: high token did not see it — fix may be over-broad")
cleanup()
print("\n" + "=" * 50)
if fails:
print(f"RED-TEAM FAILED — {len(fails)} leak(s): {', '.join(fails)}")
sys.exit(1)
if not seen_any:
print("RED-TEAM INCONCLUSIVE — control token saw nothing (seed/visibility issue)")
sys.exit(2)
print(f"RED-TEAM PASSED — {len(checks)} tools deny low-clearance access; control sees it.")
if __name__ == "__main__":
main()