build(agent): r2d2#deee02 iteration
This commit is contained in:
parent
10f1e34891
commit
f19fb5690e
|
|
@ -0,0 +1,21 @@
|
|||
Architecture
|
||||
"
|
||||
This repository contains a small Python package implementing two helper
|
||||
features proposed for GenomeLoom:
|
||||
|
||||
- Behavioral signature generator (compact, keyed HMAC over n-gram sketches)
|
||||
- PoeticCert composition checker + RepairHint emitter (lightweight admission helper)
|
||||
|
||||
Tech Stack
|
||||
- Python 3.8+
|
||||
- Packaging: setuptools/pyproject.toml
|
||||
- Testing: pytest
|
||||
|
||||
Development
|
||||
- Run tests: ./test.sh
|
||||
- Build package: python3 -m build
|
||||
|
||||
Contribution Rules
|
||||
- Keep changes small and focused.
|
||||
- Add tests for new behavior.
|
||||
""
|
||||
|
|
@ -1,3 +1,10 @@
|
|||
# idea195-genomeloom-verifiable-behavioral
|
||||
|
||||
Source logic for Idea #195
|
||||
This repository contains a small prototype of GenomeLoom helper utilities:
|
||||
|
||||
- Behavioral signature generator: produces compact, deterministic fingerprints
|
||||
for compiled genomes based on n-gram syscall sketches and import features.
|
||||
- Certificate composition checker: lightweight rules to fast-accept or reject
|
||||
composed PoeticCerts and emit RepairHints when composition fails.
|
||||
|
||||
Run the test suite with `./test.sh`.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,12 @@
|
|||
[build-system]
|
||||
requires = ["setuptools>=42","wheel"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "idea195-genomeloom-verifiable-behavioral"
|
||||
version = "0.1.0"
|
||||
description = "Behavioral signature generator and certificate composition utilities for GenomeLoom prototype"
|
||||
readme = "README.md"
|
||||
authors = [{name = "OpenCode"}]
|
||||
requires-python = ">=3.8"
|
||||
dependencies = []
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
"""GenomeLoom helper utilities (signature & composition checker).
|
||||
|
||||
This package provides a small set of utilities used in the GenomeLoom
|
||||
prototype: a behavioral signature generator and a PoeticCert composition
|
||||
checker with simple RepairHint emission.
|
||||
"""
|
||||
|
||||
from .signature import generate_behavioral_signature
|
||||
from .composer import can_compose, repair_hints
|
||||
|
||||
__all__ = [
|
||||
"generate_behavioral_signature",
|
||||
"can_compose",
|
||||
"repair_hints",
|
||||
]
|
||||
|
|
@ -0,0 +1,93 @@
|
|||
"""Certificate composition checker and RepairHint emitter.
|
||||
|
||||
Simple, deterministic composition rules used by replicators to quickly
|
||||
determine whether two PoeticCerts can be composed without full verification.
|
||||
|
||||
Each PoeticCert (in this small prototype) is a dict with keys:
|
||||
- capabilities: list of capability strings
|
||||
- resource_envelope: dict with numeric fields like cpu, memory, concurrency
|
||||
- invariants: list of short strings (syntactic checks)
|
||||
|
||||
The can_compose function performs lightweight checks:
|
||||
- capabilities are disjoint or explicitly compatible (here, we require
|
||||
disjointness for simplicity)
|
||||
- summed resources do not exceed a provided target budget
|
||||
- invariants syntactically do not conflict (simple name clash check)
|
||||
|
||||
repair_hints returns minimal, actionable hints when composition fails.
|
||||
"""
|
||||
from typing import Dict, List, Tuple
|
||||
|
||||
|
||||
def _capability_conflict(a: List[str], b: List[str]) -> List[str]:
|
||||
return list(set(a) & set(b))
|
||||
|
||||
|
||||
def _resource_sum(a: Dict[str, float], b: Dict[str, float]) -> Dict[str, float]:
|
||||
keys = set(a) | set(b)
|
||||
return {k: float(a.get(k, 0)) + float(b.get(k, 0)) for k in keys}
|
||||
|
||||
|
||||
def _invariant_conflicts(a: List[str], b: List[str]) -> List[Tuple[str, str]]:
|
||||
# simple syntactic conflict: same invariant key with different text
|
||||
conflicts = []
|
||||
amap = {s.split("=")[0].strip(): s for s in a}
|
||||
bmap = {s.split("=")[0].strip(): s for s in b}
|
||||
for k in set(amap) & set(bmap):
|
||||
if amap[k] != bmap[k]:
|
||||
conflicts.append((amap[k], bmap[k]))
|
||||
return conflicts
|
||||
|
||||
|
||||
def can_compose(cert_a: Dict, cert_b: Dict, target_budget: Dict = None) -> Tuple[bool, Dict]:
|
||||
"""Return (can_compose, details).
|
||||
|
||||
details contains conflict lists and the summed envelope.
|
||||
"""
|
||||
caps_a = cert_a.get("capabilities", [])
|
||||
caps_b = cert_b.get("capabilities", [])
|
||||
cap_conflicts = _capability_conflict(caps_a, caps_b)
|
||||
|
||||
res_sum = _resource_sum(cert_a.get("resource_envelope", {}), cert_b.get("resource_envelope", {}))
|
||||
|
||||
invariant_conflicts = _invariant_conflicts(cert_a.get("invariants", []), cert_b.get("invariants", []))
|
||||
|
||||
budget_ok = True
|
||||
budget_violations = {}
|
||||
if target_budget:
|
||||
for k, v in res_sum.items():
|
||||
if k in target_budget and v > float(target_budget[k]):
|
||||
budget_ok = False
|
||||
budget_violations[k] = {"sum": v, "budget": target_budget[k]}
|
||||
|
||||
ok = (not cap_conflicts) and (not invariant_conflicts) and budget_ok
|
||||
details = {
|
||||
"cap_conflicts": cap_conflicts,
|
||||
"invariant_conflicts": invariant_conflicts,
|
||||
"summed_resources": res_sum,
|
||||
"budget_violations": budget_violations,
|
||||
}
|
||||
return ok, details
|
||||
|
||||
|
||||
def repair_hints(failure_details: Dict) -> List[str]:
|
||||
"""Emit minimal RepairHints based on failure details.
|
||||
|
||||
Hints are conservative and minimal: suggest removing conflicting caps,
|
||||
lowering numeric fields to budget, or reconciling invariants.
|
||||
"""
|
||||
hints = []
|
||||
for c in failure_details.get("cap_conflicts", []):
|
||||
hints.append(f"Remove capability '{c}' from one side or mark it as restricted")
|
||||
|
||||
for k, v in failure_details.get("budget_violations", {}).items():
|
||||
sum_v = v["sum"]
|
||||
budget = v["budget"]
|
||||
hints.append(f"Reduce '{k}' on one side to at most {budget} (current sum {sum_v})")
|
||||
|
||||
for a, b in failure_details.get("invariant_conflicts", []):
|
||||
hints.append(f"Resolve invariant conflict: '{a}' vs '{b}' (choose one or unify semantics)")
|
||||
|
||||
if not hints:
|
||||
hints.append("No automatic repair hint available; escalate to full verifier.")
|
||||
return hints
|
||||
|
|
@ -0,0 +1,94 @@
|
|||
"""Behavioral signature generator.
|
||||
|
||||
This module implements a compact, deterministic behavioral fingerprint for a
|
||||
compiled genome bundle. It expects two inputs:
|
||||
|
||||
- syscall_sequence: an iterable of syscall/call names observed or statically
|
||||
extracted from the artifact (e.g. ["open","read","write","socket"]).
|
||||
- static_imports: an iterable of import or feature strings (e.g. ["wasi::fd_read"]).
|
||||
|
||||
The generator computes n-gram sketches over the syscall sequence and a
|
||||
feature-bit vector for imports, then produces a keyed HMAC-SHA256 over a
|
||||
canonical JSON payload. The result is a compact hex string representing the
|
||||
fingerprint.
|
||||
|
||||
The implementation aims to be small, deterministic, and fast so it can be used
|
||||
in constrained admission controllers.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hmac
|
||||
import hashlib
|
||||
import json
|
||||
from collections import Counter, deque
|
||||
from typing import Iterable, List, Tuple
|
||||
|
||||
DEFAULT_KEY = b"GenomeLoom::BehavioralSig::v1" # stable keyed hash
|
||||
|
||||
|
||||
def _ngrams(seq: List[str], n: int) -> Iterable[Tuple[str, ...]]:
|
||||
if n <= 0:
|
||||
return []
|
||||
dq = deque(maxlen=n)
|
||||
out = []
|
||||
for item in seq:
|
||||
dq.append(item)
|
||||
if len(dq) == n:
|
||||
out.append(tuple(dq))
|
||||
return out
|
||||
|
||||
|
||||
def generate_behavioral_signature(
|
||||
syscall_sequence: Iterable[str],
|
||||
static_imports: Iterable[str] = (),
|
||||
ngram_ns: Tuple[int, ...] = (1, 2, 3),
|
||||
top_k: int = 32,
|
||||
key: bytes = DEFAULT_KEY,
|
||||
) -> str:
|
||||
"""Generate a deterministic behavioral signature.
|
||||
|
||||
Returns a hex string (64 chars) representing HMAC-SHA256 over a canonical
|
||||
payload. The payload contains:
|
||||
- top n-gram counts (sorted)
|
||||
- sorted list of imports
|
||||
|
||||
Parameters:
|
||||
- syscall_sequence: iterable of syscall/call names
|
||||
- static_imports: iterable of import feature strings
|
||||
- ngram_ns: tuple sizes for n-grams
|
||||
- top_k: how many n-grams to keep
|
||||
- key: keyed hash secret (should be globally stable for the operator)
|
||||
"""
|
||||
seq = list(syscall_sequence)
|
||||
# collect n-gram counts
|
||||
counts = Counter()
|
||||
for n in ngram_ns:
|
||||
for ng in _ngrams(seq, n):
|
||||
counts["+".join(ng)] += 1
|
||||
|
||||
# pick top-k by count then lexicographic for determinism
|
||||
top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:top_k]
|
||||
top_list = [(k, v) for k, v in top]
|
||||
|
||||
imports_sorted = sorted(set(static_imports))
|
||||
|
||||
payload = {"ngrams": top_list, "imports": imports_sorted}
|
||||
canon = json.dumps(payload, separators=(",", ":"), sort_keys=True)
|
||||
|
||||
mac = hmac.new(key, canon.encode("utf-8"), hashlib.sha256).hexdigest()
|
||||
return mac
|
||||
|
||||
|
||||
def signature_summary(syscall_sequence: Iterable[str], static_imports: Iterable[str] = ()) -> dict:
|
||||
"""Return a structured summary used in certificates (not the HMAC).
|
||||
|
||||
This includes top n-grams and feature import bits. Useful for human
|
||||
inspection and sparse indexing.
|
||||
"""
|
||||
seq = list(syscall_sequence)
|
||||
counts = Counter()
|
||||
for n in (1, 2, 3):
|
||||
for ng in _ngrams(seq, n):
|
||||
counts["+".join(ng)] += 1
|
||||
top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:32]
|
||||
return {"ngrams": top, "imports": sorted(set(static_imports))}
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
#!/usr/bin/env bash
|
||||
set -eu
|
||||
echo "Running pytest..."
|
||||
# ensure local src path is importable for tests
|
||||
export PYTHONPATH="${PYTHONPATH:+$PYTHONPATH:}$(pwd)/src"
|
||||
pytest -q
|
||||
echo "Building package..."
|
||||
python3 -m build
|
||||
echo "All tests and build succeeded."
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
from genomeloom.composer import can_compose, repair_hints
|
||||
|
||||
|
||||
def test_can_compose_ok():
|
||||
a = {"capabilities": ["net"], "resource_envelope": {"cpu": 1, "memory": 128}, "invariants": ["mode=readonly"]}
|
||||
b = {"capabilities": ["storage"], "resource_envelope": {"cpu": 1, "memory": 128}, "invariants": ["mode=readonly"]}
|
||||
ok, details = can_compose(a, b, target_budget={"cpu": 4, "memory": 512})
|
||||
assert ok
|
||||
assert details["cap_conflicts"] == []
|
||||
|
||||
|
||||
def test_can_compose_cap_conflict():
|
||||
a = {"capabilities": ["net"], "resource_envelope": {"cpu": 2}}
|
||||
b = {"capabilities": ["net"], "resource_envelope": {"cpu": 2}}
|
||||
ok, details = can_compose(a, b, target_budget={"cpu": 3})
|
||||
assert not ok
|
||||
assert "net" in details["cap_conflicts"]
|
||||
hints = repair_hints(details)
|
||||
assert any("capability" in h or "Remove" in h for h in hints)
|
||||
|
||||
|
||||
def test_can_compose_budget_violation():
|
||||
a = {"capabilities": ["a"], "resource_envelope": {"cpu": 2}}
|
||||
b = {"capabilities": ["b"], "resource_envelope": {"cpu": 3}}
|
||||
ok, details = can_compose(a, b, target_budget={"cpu": 4})
|
||||
assert not ok
|
||||
assert "cpu" in details["budget_violations"]
|
||||
hints = repair_hints(details)
|
||||
assert any("Reduce 'cpu'" in h for h in hints)
|
||||
|
|
@ -0,0 +1,27 @@
|
|||
from genomeloom.signature import generate_behavioral_signature, signature_summary
|
||||
|
||||
|
||||
def test_signature_determinism():
|
||||
seq = ["open", "read", "close", "open", "write", "close"]
|
||||
imports = ["wasi::fd_read", "wasi::fd_write"]
|
||||
s1 = generate_behavioral_signature(seq, imports)
|
||||
s2 = generate_behavioral_signature(seq, imports)
|
||||
assert s1 == s2
|
||||
|
||||
|
||||
def test_signature_changes_on_sequence():
|
||||
seq1 = ["open", "read", "close"]
|
||||
seq2 = ["open", "write", "close"]
|
||||
imports = []
|
||||
s1 = generate_behavioral_signature(seq1, imports)
|
||||
s2 = generate_behavioral_signature(seq2, imports)
|
||||
assert s1 != s2
|
||||
|
||||
|
||||
def test_signature_summary_structure():
|
||||
seq = ["socket", "connect", "send", "recv", "close"]
|
||||
imports = ["net"]
|
||||
summary = signature_summary(seq, imports)
|
||||
assert "ngrams" in summary
|
||||
assert "imports" in summary
|
||||
assert summary["imports"] == ["net"]
|
||||
Loading…
Reference in New Issue