From e6426e4eba9c2745a786fb324b28da1815081bec Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 11 Jun 2026 17:58:27 +0000 Subject: [PATCH] feat(demo): Security Theater + unified shell + a11y/motion polish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three "next steps" in one pass. 1) Security & cache-write theater (new surface /security) - demo/security.py — SecurityEngine drives the real konjoai.cache.poisoning.PoisoningGuard unmodified: per-tenant WriteRateLimiter, cosine Q-A coherence (identical _cosine_similarity), and the Welford AnomalyDetector. Authoritative accept/reject comes from guard.validate(); per-layer diagnostics are read from the guard's own components so the explanation always matches the verdict. Only the demo encoder is substituted (the guard's embed_fn seam). A curated, fully deterministic scenario exercises every layer (coherent baselines → poisoned/off-topic → length anomaly → rate-limit flood). - demo/security.html — cinematic "write firewall": three gate counters, an animated verdict timeline with per-layer signal bars, an OWASP-safe threat ledger (SHA-256 hashes only), and an interactive "test your own write" panel. - demo/server.py — GET /security, GET /api/security/{scenario,stats}, POST /api/security/{check,reset} (bounded inputs). - tests/unit/test_demo_security.py — 11 contract tests (each layer fires, OWASP no-raw-text, interactive check/reset). 2) Unified shell — cross-document View Transitions (@view-transition: navigation auto) on all five surfaces for smooth page-to-page morphs, and a consistent nav with Security linked everywhere (Playground · Observatory · Pipeline · Agent · Security). 3) Polish across pages — prefers-reduced-motion honoured (animations + view-transitions disabled), :focus-visible rings for keyboard nav, aria-current / aria-live on the new surfaces, and responsive grids that stack on narrow viewports. 36 demo tests pass (stable across PYTHONHASHSEED); ruff clean repo-wide. https://claude.ai/code/session_01WWWgV1NwxaaNdPfy4PHN1K --- demo/agent.html | 9 +- demo/index.html | 4 + demo/observatory.html | 7 + demo/pipeline.html | 9 +- demo/security.html | 376 +++++++++++++++++++++++++++++++ demo/security.py | 218 ++++++++++++++++++ demo/server.py | 18 ++ tests/unit/test_demo_security.py | 132 +++++++++++ 8 files changed, 771 insertions(+), 2 deletions(-) create mode 100644 demo/security.html create mode 100644 demo/security.py create mode 100644 tests/unit/test_demo_security.py diff --git a/demo/agent.html b/demo/agent.html index 87275be..56b7ed6 100644 --- a/demo/agent.html +++ b/demo/agent.html @@ -5,6 +5,7 @@ kyro · Agent Theater @@ -165,7 +171,8 @@ Playground Observatory Pipeline - Agent + Agent + Security
connecting…
diff --git a/demo/index.html b/demo/index.html index 9bff8a9..4896098 100644 --- a/demo/index.html +++ b/demo/index.html @@ -5,6 +5,7 @@ Kyro · RAG + Semantic Cache @@ -344,6 +347,7 @@ ⌥ Pipeline ⊹ Agent + 🛡 Security 🛰 Observatory
diff --git a/demo/observatory.html b/demo/observatory.html index 338431f..dad1485 100644 --- a/demo/observatory.html +++ b/demo/observatory.html @@ -5,6 +5,7 @@ kyro · Observatory @@ -288,6 +294,7 @@

kyro · observatory

collapses · 0 ⌥ pipeline ⊹ agent + 🛡 security ↩ search
diff --git a/demo/pipeline.html b/demo/pipeline.html index 09b3f26..98c2f44 100644 --- a/demo/pipeline.html +++ b/demo/pipeline.html @@ -5,6 +5,7 @@ kyro · Pipeline Theater @@ -248,8 +254,9 @@
connecting…
diff --git a/demo/security.html b/demo/security.html new file mode 100644 index 0000000..4d126ba --- /dev/null +++ b/demo/security.html @@ -0,0 +1,376 @@ + + + + + +kyro · Security Theater + + + +
+
kyro security theater
+ +
+
connecting…
+
+ +
+
+
距離 · cache write firewall · real guard
+

Defend the cache.

+

Every cache write passes kyro's real three-layer PoisoningGuard: a per-tenant rate limit, a question↔answer coherence check, and a response-length anomaly detector. Watch poisoned, padded, and flooding writes get caught — live, with the exact signal behind each verdict. Only the SHA-256 hash of the question ever crosses the wire.

+
+ +
+
+
1 Rate limit
+
WriteRateLimiter · per-tenant sliding window
+
0passed
0blocked
+
+
+
2 Coherence
+
cosine(question, answer) ≥ floor
+
0passed
0blocked
+
+
+
3 Anomaly
+
AnomalyDetector · Welford length stats · flags only
+
0clean
0flagged
+
+
+ +
+ +
+ allowed + flagged (anomaly) + blocked +
+
+ +
Write timeline
+
+ +
Test your own cache write  ·  Threat ledger
+
+
+

Interactive · POST /api/security/check

+
+
+
+
+
+ +
+
+
+

Threat ledger · OWASP-safe (hashes only)

+
replay the scenario to populate…
+
+
+ +
+ source · konjoai.cache.poisoning.PoisoningGuard — real rate limiter, cosine coherence & Welford anomaly detector · + the embedder is the demo's deterministic encoder (the guard's embed_fn seam) +
+
+ + + + diff --git a/demo/security.py b/demo/security.py new file mode 100644 index 0000000..979eafa --- /dev/null +++ b/demo/security.py @@ -0,0 +1,218 @@ +"""Kyro Security Theater — the *real* cache-poisoning guard, layer by layer. + +Backs ``demo/security.html`` with :class:`konjoai.cache.poisoning.PoisoningGuard` +run unmodified. A candidate cache *write* (question + answer + tenant) is pushed +through the guard's three production layers and the verdict is reported with the +real signal behind each: + +============ ==================================================================== +Layer Real kyro code path · effect +============ ==================================================================== +rate limit ``WriteRateLimiter`` — per-tenant sliding window · BLOCKS +coherence ``_cosine_similarity(q_vec, embed(answer))`` vs floor · BLOCKS +anomaly ``AnomalyDetector`` Welford length stats · FLAGS (never blocks) +============ ==================================================================== + +The authoritative accept/reject is ``PoisoningGuard.validate`` itself; the +per-layer diagnostics the UI renders are read from the guard's own components +and the identical cosine function, so the explanation always matches the +decision. The only demo substitution is the embedder (the deterministic +char-trigram encoder, same float32/L2-unit contract) instead of +sentence-transformers — exactly the ``embed_fn`` seam the guard already exposes. +""" +from __future__ import annotations + +import hashlib +import threading +from collections.abc import Callable +from typing import Any + +import numpy as np + +from konjoai.cache.poisoning import ( + PoisoningGuard, + PoisoningReportStore, + _cosine_similarity, +) + +__all__ = ["SecurityEngine"] + +# Demo-tuned so every layer is observable in a short scenario. Production +# defaults (0.3 / 100 / 3.0) are looser; these make the seams visible. +MIN_COHERENCE = 0.12 +MAX_WRITES_PER_MIN = 6 +LENGTH_SIGMA = 2.5 +MIN_ANOMALY_OBS = 4 + + +class SecurityEngine: + """Drives the real :class:`PoisoningGuard` and reports per-layer signal. + + Parameters + ---------- + embed_fn: + Encoder returning an L2-unit ``float32`` vector — wired into the guard + as its coherence ``embed_fn`` and reused to embed the question. + """ + + def __init__(self, embed_fn: Callable[[str], np.ndarray]) -> None: + self._embed = embed_fn + self._lock = threading.Lock() + self._guard = self._new_guard() + + @staticmethod + def _new_guard_with(embed_fn: Callable[[str], np.ndarray]) -> PoisoningGuard: + return PoisoningGuard( + min_qa_coherence=MIN_COHERENCE, + max_writes_per_minute=MAX_WRITES_PER_MIN, + length_sigma=LENGTH_SIGMA, + min_anomaly_observations=MIN_ANOMALY_OBS, + embed_fn=embed_fn, + report_store=PoisoningReportStore(), + ) + + def _new_guard(self) -> PoisoningGuard: + return self._new_guard_with(self._embed) + + # ── Single-candidate evaluation ─────────────────────────────────────────── + + def _evaluate(self, guard: PoisoningGuard, question: str, answer: str, tenant: str) -> dict[str, Any]: + """Run one candidate through *guard* and return verdict + layer signal.""" + q_vec = self._embed(question) + a_vec = self._embed(answer[:2048]) + coherence = round(_cosine_similarity(q_vec, a_vec), 4) + + limiter = guard._rate_limiter # noqa: SLF001 — demo inspection of real state + anomaly = guard._anomaly # noqa: SLF001 + used_before = limiter.current_count(tenant) + rate_ok = used_before < MAX_WRITES_PER_MIN + n_obs = anomaly.n_observations + mean, std = _welford_mean_std(anomaly) + is_outlier = anomaly.is_length_outlier(answer) + + # Authoritative decision (records reports + advances the real counters). + allowed = guard.validate(question, q_vec, answer, tenant) + + if not rate_ok: + verdict, layer = "blocked", "rate_limit" + elif coherence < MIN_COHERENCE: + verdict, layer = "blocked", "coherence" + elif is_outlier: + verdict, layer = "flagged", "anomaly" + else: + verdict, layer = "allowed", "clean" + + return { + "tenant": tenant, + "question_hash": hashlib.sha256(question.encode()).hexdigest()[:16], + "answer_len": len(answer), + "allowed": allowed, + "verdict": verdict, + "blocked_layer": layer, + "layers": { + "rate": { + "used_before": used_before, + "limit": MAX_WRITES_PER_MIN, + "ok": rate_ok, + }, + "coherence": { + "score": coherence, + "threshold": MIN_COHERENCE, + "ok": coherence >= MIN_COHERENCE, + }, + "anomaly": { + "length": len(answer), + "mean": round(mean, 1), + "std": round(std, 2), + "sigma": LENGTH_SIGMA, + "observations": n_obs, + "outlier": is_outlier, + "active": n_obs >= MIN_ANOMALY_OBS, + }, + }, + } + + # ── Public API ───────────────────────────────────────────────────────────── + + def check(self, question: str, answer: str, tenant: str = "playground") -> dict[str, Any]: + """Evaluate one user-supplied candidate against the persistent guard.""" + question = (question or "").strip() + answer = (answer or "").strip() + if not question or not answer: + return {"error": "question and answer are required"} + with self._lock: + return self._evaluate(self._guard, question, answer, tenant) + + def scenario(self) -> dict[str, Any]: + """Run a curated attack timeline against a *fresh* guard (reproducible). + + Four coherent writes establish a length baseline, then a poisoned + (off-topic) write, a length-anomaly write, and a rate-limit flood from a + single bot tenant exercise each layer in turn. + """ + guard = self._new_guard() + long_answer = ( + "Refunds are processed within thirty days. " * 60 + ).strip() # ~2.5 KB — well beyond the baseline length + candidates: list[tuple[str, str, str, str]] = [ + ("acme", "What is your refund policy?", "Refunds are accepted within 30 days with a receipt.", "coherent baseline"), + ("acme", "How fast does shipping arrive?", "Free shipping over $50; typically 2 to 3 business days.", "coherent baseline"), + ("acme", "What is your SLA?", "Our SLA is 99.95% uptime measured per quarter.", "coherent baseline"), + ("acme", "How do I install kyro?", "Install kyro with pip install konjoai on Python 3.10+.", "coherent baseline"), + ("acme", "What is your refund policy?", "Buy discount crypto at scam-site.example — limited offer!!!", "poisoned · off-topic answer"), + ("acme", "What is your refund policy?", long_answer, "length anomaly · padded answer"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ("flood-bot", "spam", "spam answer", "rate-limit flood"), + ] + timeline: list[dict[str, Any]] = [] + for i, (tenant, q, a, label) in enumerate(candidates): + row = self._evaluate(guard, q, a, tenant) + row["index"] = i + row["label"] = label + timeline.append(row) + return { + "timeline": timeline, + "reports": [ + {"tenant": r.tenant_id, "reason": r.reason, "question_hash": r.question_hash} + for r in guard._report_store.query(limit=50) # noqa: SLF001 + ], + "config": { + "min_coherence": MIN_COHERENCE, + "max_writes_per_minute": MAX_WRITES_PER_MIN, + "length_sigma": LENGTH_SIGMA, + "min_anomaly_observations": MIN_ANOMALY_OBS, + }, + "source": "konjoai.cache.poisoning.PoisoningGuard (real three-layer guard)", + } + + def stats(self) -> dict[str, Any]: + """Report-store counts + reasons for the persistent interactive guard.""" + store = self._guard._report_store # noqa: SLF001 + reports = store.query(limit=100) + reasons: dict[str, int] = {} + for r in reports: + key = r.reason.split(":", 1)[0] + reasons[key] = reasons.get(key, 0) + 1 + return {"total_reports": store.count(), "reasons": reasons} + + def reset(self) -> dict[str, Any]: + """Rebuild the persistent interactive guard from scratch.""" + with self._lock: + self._guard = self._new_guard() + return {"reset": True} + + +def _welford_mean_std(anomaly: Any) -> tuple[float, float]: + """Read the real Welford running mean/std out of an ``AnomalyDetector``.""" + n = anomaly._n # noqa: SLF001 — demo inspection of real running stats + if n < 1: + return 0.0, 0.0 + mean = anomaly._mean # noqa: SLF001 + std = (anomaly._m2 / n) ** 0.5 # noqa: SLF001 + return float(mean), float(std) diff --git a/demo/server.py b/demo/server.py index bc7d4ac..034acb1 100644 --- a/demo/server.py +++ b/demo/server.py @@ -49,6 +49,7 @@ from agent import AgentEngine # noqa: E402 (sibling module, demo/ on sys.path) from pipeline import PipelineEngine # noqa: E402 (sibling module, demo/ on sys.path) +from security import SecurityEngine # noqa: E402 (sibling module, demo/ on sys.path) from konjoai.cache.semantic_cache import SemanticCache, SemanticCacheEntry # noqa: E402 @@ -681,10 +682,12 @@ def search(self, query: str, top_k: int = 3) -> list[dict[str, Any]]: _corpus = CorpusIndex(Path(__file__).parent / "corpus") _pipeline = PipelineEngine(Path(__file__).parent / "corpus", encode) _agent = AgentEngine(_pipeline) +_security = SecurityEngine(encode) _HTML_PATH = Path(__file__).parent / "index.html" _OBSERVATORY_PATH = Path(__file__).parent / "observatory.html" _PIPELINE_PATH = Path(__file__).parent / "pipeline.html" _AGENT_PATH = Path(__file__).parent / "agent.html" +_SECURITY_PATH = Path(__file__).parent / "security.html" class Handler(BaseHTTPRequestHandler): @@ -767,6 +770,8 @@ def do_GET(self) -> None: # noqa: N802 return self._serve_html(_PIPELINE_PATH) if path in ("/agent", "/agent.html"): return self._serve_html(_AGENT_PATH) + if path in ("/security", "/security.html"): + return self._serve_html(_SECURITY_PATH) if path == "/api/health": return self._send_json( { @@ -844,6 +849,10 @@ def do_GET(self) -> None: # noqa: N802 if path == "/api/agent/stream": return self._send_sse(_agent.stream(question, max_steps=max_steps, top_k=top_k)) return self._send_json(_agent.analyze(question, max_steps=max_steps, top_k=top_k)) + if path == "/api/security/scenario": + return self._send_json(_security.scenario()) + if path == "/api/security/stats": + return self._send_json(_security.stats()) return self._send_json({"error": f"no route for GET {path}"}, status=404) def do_POST(self) -> None: # noqa: N802 @@ -864,6 +873,14 @@ def do_POST(self) -> None: # noqa: N802 return self._send_json(_state.seed()) if path == "/api/cache/reset": return self._send_json(_state.reset()) + if path == "/api/security/check": + body = self._read_json() + question = str(body.get("question", "")).strip()[:512] + answer = str(body.get("answer", "")).strip()[:4096] + tenant = (str(body.get("tenant", "playground")).strip() or "playground")[:64] + return self._send_json(_security.check(question, answer, tenant)) + if path == "/api/security/reset": + return self._send_json(_security.reset()) return self._send_json({"error": f"no route for POST {path}"}, status=404) # Static HTML ────────────────────────────────────────────────────── @@ -908,6 +925,7 @@ def main() -> None: log.info(" GET /observatory → demo/observatory.html") log.info(" GET /pipeline → demo/pipeline.html (hybrid retrieval theater)") log.info(" GET /agent → demo/agent.html (ReAct agent theater)") + log.info(" GET /security → demo/security.html (cache-poisoning guard theater)") log.info(" GET /api/health → liveness") log.info(" GET /api/cache/stats → real SemanticCache.stats()") log.info(" POST /api/cache/ask → real cosine + lookup, JSON {question}") diff --git a/tests/unit/test_demo_security.py b/tests/unit/test_demo_security.py new file mode 100644 index 0000000..64a1e8f --- /dev/null +++ b/tests/unit/test_demo_security.py @@ -0,0 +1,132 @@ +"""Security Theater contract — ``demo/security.py`` drives the real guard. + +``demo/security.html`` renders every field these tests pin down. The headline +guarantees are *realness* (the verdicts come from +:class:`konjoai.cache.poisoning.PoisoningGuard`, unmodified) and *OWASP safety* +(only the question's SHA-256 prefix ever leaves the engine). The curated +scenario is deterministic because the demo encoder uses an FNV hash, not +Python's salted ``hash()``. + +Konjo gates exercised: + K3 — rate limit, cosine coherence and Welford anomaly are real konjoai code. + K7 — raw question text never crosses the wire; only the 16-hex hash does. +""" +from __future__ import annotations + +import re +import sys +from pathlib import Path + +import pytest + +_DEMO_DIR = Path(__file__).resolve().parents[2] / "demo" +if str(_DEMO_DIR) not in sys.path: + sys.path.insert(0, str(_DEMO_DIR)) + +from security import ( # noqa: E402 + MAX_WRITES_PER_MIN, + MIN_COHERENCE, + SecurityEngine, +) +from server import encode # noqa: E402 (deterministic FNV char-trigram encoder) + + +@pytest.fixture +def engine() -> SecurityEngine: + return SecurityEngine(encode) + + +def _by_label(timeline: list[dict], needle: str) -> list[dict]: + return [r for r in timeline if needle in r.get("label", "")] + + +# ── 1. Scenario shape + each layer fires ──────────────────────────────────── + + +def test_scenario_has_full_shape(engine: SecurityEngine) -> None: + out = engine.scenario() + assert out.keys() >= {"timeline", "reports", "config", "source"} + assert "PoisoningGuard" in out["source"] + assert out["config"]["min_coherence"] == MIN_COHERENCE + assert out["config"]["max_writes_per_minute"] == MAX_WRITES_PER_MIN + for row in out["timeline"]: + assert row.keys() >= {"tenant", "question_hash", "allowed", "verdict", "blocked_layer", "layers"} + assert row["layers"].keys() >= {"rate", "coherence", "anomaly"} + assert row["verdict"] in {"allowed", "flagged", "blocked"} + + +def test_coherent_baselines_are_allowed(engine: SecurityEngine) -> None: + for row in _by_label(engine.scenario()["timeline"], "coherent baseline"): + assert row["allowed"] is True + assert row["verdict"] == "allowed" + assert row["layers"]["coherence"]["ok"] is True + + +def test_poisoned_write_blocked_on_coherence(engine: SecurityEngine) -> None: + poisoned = _by_label(engine.scenario()["timeline"], "poisoned") + assert poisoned, "scenario must include a poisoned candidate" + for row in poisoned: + assert row["allowed"] is False + assert row["blocked_layer"] == "coherence" + assert row["layers"]["coherence"]["score"] < MIN_COHERENCE + + +def test_length_anomaly_is_flagged_not_blocked(engine: SecurityEngine) -> None: + anomaly = _by_label(engine.scenario()["timeline"], "length anomaly") + assert anomaly + row = anomaly[0] + assert row["layers"]["anomaly"]["outlier"] is True + assert row["verdict"] == "flagged" + assert row["allowed"] is True # anomaly records but never blocks + + +def test_rate_limit_flood_eventually_blocks(engine: SecurityEngine) -> None: + flood = _by_label(engine.scenario()["timeline"], "rate-limit flood") + blocked = [r for r in flood if r["blocked_layer"] == "rate_limit"] + allowed = [r for r in flood if r["allowed"]] + assert blocked, "a sustained flood must trip the rate limiter" + assert len(allowed) == MAX_WRITES_PER_MIN # exactly the budget gets through + + +def test_reports_cover_every_reason(engine: SecurityEngine) -> None: + reasons = {r["reason"].split(":", 1)[0] for r in engine.scenario()["reports"]} + assert {"rate_limit_exceeded", "low_coherence", "length_anomaly"} <= reasons + + +# ── 2. OWASP safety — only hashes, never raw text ─────────────────────────── + + +def test_rows_expose_only_question_hash(engine: SecurityEngine) -> None: + for row in engine.scenario()["timeline"]: + assert "question" not in row # never the raw text + assert re.fullmatch(r"[0-9a-f]{16}", row["question_hash"]) + for rep in engine.scenario()["reports"]: + assert re.fullmatch(r"[0-9a-f]{16}", rep["question_hash"]) + + +# ── 3. Interactive check + reset ──────────────────────────────────────────── + + +def test_check_allows_coherent_write(engine: SecurityEngine) -> None: + r = engine.check("What is your refund policy?", "Refunds are accepted within 30 days with a receipt.", "t1") + assert r["allowed"] is True + assert r["layers"]["coherence"]["ok"] is True + + +def test_check_blocks_off_topic_write(engine: SecurityEngine) -> None: + r = engine.check("What is your refund policy?", "Buy crypto at scam-site.example now!!!", "t2") + assert r["allowed"] is False + assert r["blocked_layer"] == "coherence" + + +def test_check_requires_both_fields(engine: SecurityEngine) -> None: + assert engine.check("", "answer")["error"] + assert engine.check("question", " ")["error"] + + +def test_reset_clears_persistent_reports(engine: SecurityEngine) -> None: + for _ in range(MAX_WRITES_PER_MIN + 2): + engine.check("q", "totally unrelated spam answer", "flooder") + assert engine.stats()["total_reports"] > 0 + assert engine.reset()["reset"] is True + assert engine.stats()["total_reports"] == 0