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