coulomb/emergent-security
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bind knowledge retrieval to sealed bytes and invalidate changed releases

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a10238-7ecc-78e2-903d-075f50e913cc
This commit is contained in:
tegwick 2026-10-03 17:44:52 +02:00
parent 416f9919c5
commit d0bf65e964
16 changed files with 335 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

65
src/emergent_security/knowledge.py
View file

@ -40,16 +40,27 @@ def _mapping(loader, node, deep=False):
UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _mapping)
def read_yaml(path: Path) -> dict:
def _read_bytes(path: Path) -> bytes:
try:
value = yaml.load(path.read_text(encoding="utf-8"), Loader=UniqueLoader)
except (OSError, UnicodeError, yaml.YAMLError):
return path.read_bytes()
except OSError:
raise KnowledgeError("unreadable content") from None
def _yaml_mapping(data: bytes) -> dict:
try:
value = yaml.load(data.decode("utf-8"), Loader=UniqueLoader)
except (UnicodeError, yaml.YAMLError):
raise KnowledgeError("unreadable YAML document") from None
if not isinstance(value, dict):
raise KnowledgeError("expected YAML mapping")
return value
def read_yaml(path: Path) -> dict:
return _yaml_mapping(_read_bytes(path))
def confined(root: Path, name: str) -> Path:
if not isinstance(name, str) or not name or Path(name).is_absolute() or ".." in Path(name).parts:
raise KnowledgeError("invalid relative path")
@ -90,19 +101,20 @@ def default_bundle() -> Path:
class Knowledge:
def __init__(self, root: Path | None = None):
self.root = (root or default_bundle()).resolve()
self.release = read_yaml(confined(self.root, "release.yaml"))
release_bytes = _read_bytes(confined(self.root, "release.yaml"))
self.package_digest = hashlib.sha256(release_bytes).hexdigest()
self.release = _yaml_mapping(release_bytes)
if self.release.get("schema") != "emergent.release.v1":
raise KnowledgeError("unsupported release schema")
hashes = self.release.get("files")
if not isinstance(hashes, dict) or not {"catalog.yaml", "sources.yaml"} <= hashes.keys():
raise KnowledgeError("incomplete release manifest")
for name, expected in hashes.items():
for expected in hashes.values():
if not isinstance(expected, str) or not re.fullmatch(r"[a-f0-9]{64}", expected):
raise KnowledgeError("invalid release digest")
if digest(confined(self.root, name)) != expected:
raise KnowledgeError("release content changed; review and reseal")
self.catalog = read_yaml(confined(self.root, "catalog.yaml"))
sources = read_yaml(confined(self.root, "sources.yaml"))
contents = self._verified_contents()
self.catalog = _yaml_mapping(contents["catalog.yaml"])
sources = _yaml_mapping(contents["sources.yaml"])
if self.catalog.get("schema") != "emergent.knowledge.v1" or sources.get("schema") != "emergent.sources.v1":
raise KnowledgeError("unsupported knowledge schema")
if self.catalog.get("version") != self.release.get("version"):
@ -146,9 +158,31 @@ class Knowledge:
raise KnowledgeError("invalid source digest")
if not str(document.get("url", "")).startswith("https://forgejo.coulomb.social/"):
raise KnowledgeError("invalid canonical source URL")
self.package_digest = digest(self.root / "release.yaml")
def _verified_contents(self) -> dict[str, bytes]:
"""Bind every operation to this load's seal and return verified bytes.
A changed seal requires an explicit reload, never silent adoption. Read
each file once so the bytes served cannot differ from the bytes hashed.
"""
release_path = confined(self.root, "release.yaml")
if digest(release_path) != self.package_digest:
raise KnowledgeError("release changed; reload the reviewed bundle")
contents = {}
for name, expected in self.release["files"].items():
data = _read_bytes(confined(self.root, name))
if hashlib.sha256(data).hexdigest() != expected:
raise KnowledgeError("release content changed; review and reseal")
contents[name] = data
if digest(confined(self.root, "release.yaml")) != self.package_digest:
raise KnowledgeError("release changed; reload the reviewed bundle")
return contents
def source_status(self, ref: str, repositories: Path | None, today: date) -> str:
self._verified_contents()
return self._source_status(ref, repositories, today)
def _source_status(self, ref: str, repositories: Path | None, today: date) -> str:
source = self.sources[ref]
if source["status"] != "active":
return source["status"]
@ -174,6 +208,7 @@ class Knowledge:
def retrieve(self, card_id: str, *, audience: str = "local",
repositories: Path | None = None, today: date | None = None) -> dict:
contents = self._verified_contents()
if card_id not in self.cards:
raise KnowledgeError("unknown card; use list")
card = self.cards[card_id]
@ -182,7 +217,7 @@ class Knowledge:
raise KnowledgeError("private or tenant content is not served")
if audience not in card.get("audiences", []):
raise KnowledgeError("audience not approved for this card")
statuses = {ref: self.source_status(ref, repositories, today or date.today())
statuses = {ref: self._source_status(ref, repositories, today or date.today())
for ref in card["sources"]}
withdrawn = any(value in {"withdrawn", "superseded"} for value in statuses.values())
stale = any(value in {"changed", "review-due", "future-verification"} for value in statuses.values())
@ -190,7 +225,10 @@ class Knowledge:
# dated snapshot useful for orientation, with no current-authority claim.
state = "withheld" if withdrawn or stale else (
"verified-files" if all(v == "verified-files" for v in statuses.values()) else "offline-snapshot")
text = "" if state == "withheld" else confined(self.root, card["path"]).read_text(encoding="utf-8")
try:
text = "" if state == "withheld" else contents[card["path"]].decode("utf-8")
except UnicodeError:
raise KnowledgeError("unreadable card") from None
if card_id == "start" and len(text.encode()) > self.catalog["startup_max_bytes"]:
raise KnowledgeError("startup context budget exceeded")
return {
@ -207,7 +245,8 @@ class Knowledge:
}
def check(self, repositories: Path | None = None, today: date | None = None) -> dict:
statuses = {ref: self.source_status(ref, repositories, today or date.today()) for ref in self.sources}
self._verified_contents()
statuses = {ref: self._source_status(ref, repositories, today or date.today()) for ref in self.sources}
return {"schema": "emergent.knowledge-check.v1", "version": self.catalog["version"],
"package_digest": self.package_digest, "sources": statuses,
"live_state_verified": False,