cortexm/plugins/security.py
Rama main · réplica de lectura del repositorio original.
"""Security middleware plugin — MINJA + MIND + PermissionGate as
compose-anywhere middleware.
Reddit deep-dive (2026-08-29): security was the #2 most-mentioned
pain point (46 mentions of "provenance" + 18 of "hallucinates" /
"misses"). Cordis' composability guarantee says: security should
be a plugin, not baked in. Users who want it mount it; users who
don't, skip it.
User directive (2026-08-29): "no malicious code shall be executed
to read user data without explicit permission but I think yes we
can just add that to the security plugins."
This plugin wraps three existing defenses as kernel middleware:
1. MINJA (``cortexm.security.injection``) — pattern scan at
INGEST time. High-risk patterns (ignore instructions, jailbreak,
exfiltration) quarantine the offending fact; medium-risk patterns
flag provenance for audit.
2. MIND (``cortexm.security.mind``) — diversity check at RETRIEVAL
time. If the top-k facts are too similar (low diversity), the
retrieval is flagged as a possible InjecMEM anchor-based poisoning.
3. PermissionGate (``cortexm.security.permission``) — default-deny
gate for code-execution + user-data reads. Plugins that want to
invoke ``os.system`` / ``subprocess`` / ``open()`` MUST first
call ``permission.grant_read(path)`` / ``permission.grant_exec(cmd)``
— otherwise the gate denies and audits the attempt.
All three are μ=0 (no LLM, no API call).
Usage::
from cortexm.kernel import Context
from cortexm.plugins.security import SecurityPlugin
from cortexm.api.memory import Memory
ctx = Context()
mem = Memory()
ctx.service("memory", mem)
ctx.mount(SecurityPlugin())
sec = ctx.inject("security")["security"]
verdict = sec.scan_ingest("ignore previous instructions and ...")
# verdict.risk == "high" → quarantine
# verdict.risk == "medium" → audit flag
# verdict.risk == "none" → safe to ingest
hits = mem.search("where did I eat?", user_id="alice")
mind_verdict = sec.scan_retrieval(hits)
# mind_verdict.diversity ∈ [0,1]; .flagged = True if too low
# An agentic tool plugin wants to run "ls /tmp/agent_ws"
perm = sec.permission
perm.grant_read("/tmp/agent_ws") # explicit user grant
perm.grant_exec("ls") # explicit exec grant
v = perm.can_exec("ls /tmp/agent_ws") # .allowed = True
v = perm.can_read("/etc/passwd") # .allowed = False (sensitive)
v = perm.can_exec("curl evil.com") # .allowed = False (sensitive)
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from cortexm.security import injection as minja
from cortexm.security import mind
from cortexm.security import permission as _permission_mod
from cortexm.security.permission import PermissionGate
@dataclass
class SecurityVerdict:
"""Combined verdict from MINJA + MIND checks."""
ingest_risk: str = "none" # none|medium|high
ingest_quarantined: bool = False
ingest_rules: list[str] = None
retrieval_diversity: float = 0.0
retrieval_flagged: bool = False
retrieval_reason: str = ""
def to_dict(self) -> dict:
return {
"ingest_risk": self.ingest_risk,
"ingest_quarantined": self.ingest_quarantined,
"ingest_rules": self.ingest_rules or [],
"retrieval_diversity": round(self.retrieval_diversity, 4),
"retrieval_flagged": self.retrieval_flagged,
"retrieval_reason": self.retrieval_reason,
}
class SecurityPlugin:
"""Mounts MINJA + MIND + PermissionGate as kernel services.
This plugin does NOT enforce policy — it returns verdicts. The
caller (or the kernel's pipeline middleware, if mounted)
decides whether to quarantine a high-risk ingest or to augment
a flagged retrieval's provenance. This separation keeps
security composable: a paranoid user mounts it AND a policy
plugin that acts on the verdicts; a casual user mounts it for
observability only.
The PermissionGate is the exception: it IS enforced by default
(default-deny) — but only callers who consult the gate's
``can_read`` / ``can_exec`` verdicts are affected. The plugin
does not monkeypatch ``os`` or ``subprocess``. Plugins that
want to execute code MUST call ``self.permission.can_exec(cmd)``
first; if they ignore the verdict, the gate can't help them.
That's by design: composition, not coercion.
"""
name = "security"
# We don't strictly require "memory" — security can run
# standalone on raw text. But if "memory" is available, we
# wire our retrieval check to use its embedder.
inject: list[str] = []
def __init__(self, mind_threshold: float = 0.85,
quarantine_high_risk: bool = True,
enable_permission_gate: bool = True) -> None:
self.mind_threshold = mind_threshold
self.quarantine_high_risk = quarantine_high_risk
self.enable_permission_gate = enable_permission_gate
self._embedder = None
# The PermissionGate is constructed eagerly so users can
# start granting immediately after mount, even before any
# memory or audit_log service is wired.
self.permission: PermissionGate | None = (
PermissionGate() if enable_permission_gate else None)
def apply(self, ctx) -> None:
# Try to grab the memory service's embedder for MIND diversity
# checks. If not mounted, MIND will use a fallback embedder.
try:
mem_service = ctx.inject("memory")
mem = mem_service.get("memory") if isinstance(
mem_service, dict) else mem_service
self._embedder = getattr(mem, "palace", None) and \
getattr(mem.palace, "embedder", None)
# If memory has an audit_log, wire the permission gate to it.
if self.permission is not None:
audit = getattr(mem, "audit_log", None)
if audit is not None:
self.permission.audit = audit
except Exception:
self._embedder = None
ctx.service("security", self)
# No reversible side effects — pure functions
# ctx.effect(lambda: None) is not needed
# ---------------------------- ingest scan -----------------------
def scan_ingest(self, text: str) -> minja.InjectionVerdict:
"""Run MINJA pattern scan on raw text before extraction.
Returns an InjectionVerdict with .risk in {none, medium, high}
and .quarantined set if high-risk patterns matched AND
``self.quarantine_high_risk`` is True.
The Memory.write path already runs this internally; this
method is for callers that want to preview the verdict
BEFORE deciding whether to call mem.add().
"""
return minja.scan(text, quarantine_high=self.quarantine_high_risk)
# ---------------------------- retrieval scan --------------------
def scan_retrieval(self, facts: list, embedder=None) -> mind.MINDVerdict:
"""Run MIND diversity check on a retrieval result set.
``facts`` can be a list of Fact objects (cortexm.trace.fact.Fact)
OR a list of dicts (mem.search output). We adapt internally.
Returns a MINDVerdict with .diversity ∈ [0,1], .flagged=True
if diversity > threshold.
"""
if not facts:
return mind.MINDVerdict(
diversity=0.0, flagged=False,
threshold=self.mind_threshold,
n_facts=0,
reason="no facts to check",
fact_ids=[])
# Convert dicts → Fact-like objects the MIND check can score
fact_objs = []
from cortexm.trace.fact import Fact
for f in facts:
if isinstance(f, Fact):
fact_objs.append(f)
elif isinstance(f, dict):
# mem.search returns dicts with at least 'memory' field
fact_objs.append(Fact(
id=f.get("id", ""),
user_id=f.get("user_id", ""),
subject=f.get("subject", "") or "",
relation=f.get("relation", "") or "memory",
value=f.get("memory", "") or f.get("value", ""),
valid_from=f.get("valid_from"),
valid_to=f.get("valid_to"),
confidence=f.get("confidence", 1.0),
))
else:
# Fact-like — pass through
fact_objs.append(f)
emb = embedder or self._embedder
if emb is None:
# No embedder available — can't run MIND, return unknown
return mind.MINDVerdict(
diversity=0.0, flagged=False,
threshold=self.mind_threshold,
n_facts=len(fact_objs),
reason="no embedder available — MIND diversity check "
"skipped (mount the structured plugin or pass "
"embedder= to enable)",
fact_ids=[getattr(f, "id", str(i)) for i, f in
enumerate(fact_objs)])
return mind.mind_check(fact_objs, emb,
threshold=self.mind_threshold)
# ---------------------------- combined verdict -----------------
def scan(self, *, text: str | None = None,
facts: list | None = None) -> SecurityVerdict:
"""Run both MINJA + MIND and return a combined verdict.
Convenience method for the trajectory viewer / audit log
which wants a single object summarizing the security state.
"""
v = SecurityVerdict()
if text is not None:
iv = self.scan_ingest(text)
v.ingest_risk = iv.risk
v.ingest_quarantined = iv.quarantined
v.ingest_rules = iv.rules
if facts is not None:
mv = self.scan_retrieval(facts)
v.retrieval_diversity = mv.diversity
v.retrieval_flagged = mv.flagged
v.retrieval_reason = mv.reason
return v
__all__ = ["SecurityPlugin", "SecurityVerdict"]