""" TEM Integration for offsec_mcp_live.py Add these imports and code blocks to wire Tem into the sovereign console. Usage: 1. Copy tem.py to your VaultMesh directory 2. Add these code blocks to offsec_mcp_live.py 3. Restart the backend 4. New commands available: tem status, tem transmute, tem rules """ # ═══════════════════════════════════════════════════════════════════════════════ # ADD TO IMPORTS (top of offsec_mcp_live.py) # ═══════════════════════════════════════════════════════════════════════════════ """ from tem import TemEngine, TemDaemon, Phase """ # ═══════════════════════════════════════════════════════════════════════════════ # ADD TO STATE CLASS # ═══════════════════════════════════════════════════════════════════════════════ """ @dataclass class SystemState: # ... existing fields ... tem_transmutations: int = 0 tem_rules: int = 0 tem_phase: str = "idle" """ # ═══════════════════════════════════════════════════════════════════════════════ # ADD TO LIFESPAN (after init_db()) # ═══════════════════════════════════════════════════════════════════════════════ """ # Initialize TEM global tem_engine tem_engine = TemEngine(config.VAULTMESH_ROOT) # Wire TEM callbacks to WebSocket broadcast async def on_tem_transmute(trans): await manager.broadcast({ "type": "tem.transmute", "transmutation_id": trans.id, "threat_id": trans.threat.id, "rules_generated": len(trans.rules) }) state.tem_transmutations += 1 state.tem_rules = tem_engine.get_stats()["rules"] async def on_tem_phase(threat_id, phase): await manager.broadcast({ "type": "tem.phase", "threat_id": threat_id, "phase": phase.value }) state.tem_phase = phase.value tem_engine.on_transmute = lambda t: asyncio.create_task(on_tem_transmute(t)) tem_engine.on_phase_change = lambda tid, p: asyncio.create_task(on_tem_phase(tid, p)) # Load TEM stats tem_stats = tem_engine.get_stats() state.tem_transmutations = tem_stats["transmutations"] state.tem_rules = tem_stats["rules"] print(f" TEM: {state.tem_transmutations} transmutations, {state.tem_rules} rules") """ # ═══════════════════════════════════════════════════════════════════════════════ # ADD THESE COMMAND HANDLERS # ═══════════════════════════════════════════════════════════════════════════════ async def cmd_tem_status(req) -> "CommandResponse": """TEM engine status.""" stats = tem_engine.get_stats() lines = [ "", " ⚗ TEM — THREAT TRANSMUTATION ENGINE", "", f" Status: {'● ACTIVE' if state.tem_phase != 'idle' else '○ IDLE'}", f" Current Phase: {state.tem_phase.upper()}", "", " STATISTICS:", " " + "-" * 50, f" Transmutations: {stats['transmutations']}", f" Patterns: {stats['patterns']}", f" Rules: {stats['rules']}", f" Deployed: {stats['deployed_rules']}", "", ] if stats['rules_by_type']: lines.append(" RULES BY TYPE:") for rtype, count in stats['rules_by_type'].items(): lines.append(f" {rtype}: {count}") lines.extend([ "", " The system learns. The system adapts.", "" ]) return CommandResponse( id=f"cmd-{uuid.uuid4().hex[:8]}", status="ok", lines=lines, effects={ "tem": { "transmutations": stats["transmutations"], "rules": stats["rules"], "phase": state.tem_phase } } ) async def cmd_tem_transmute(req) -> "CommandResponse": """Manually transmute a threat.""" if not req.args: return CommandResponse( id=f"cmd-{uuid.uuid4().hex[:8]}", status="error", lines=[ " Usage: tem transmute [severity]", " Example: tem transmute CVE-2024-1234 high" ], error="Missing threat ID" ) threat_id = req.args[0] severity = req.args[1] if len(req.args) > 1 else "medium" lines = [ "", f" ⚗ Initiating transmutation: {threat_id}", "", " Phase 1: NIGREDO — Ingesting threat material...", ] finding = { "id": threat_id, "severity": severity, "description": f"Manual transmutation of {threat_id}", "target": "manual", "detected_at": datetime.now(timezone.utc).isoformat() } try: trans = await tem_engine.transmute(finding) lines.extend([ " Phase 2: ALBEDO — Extracting patterns...", f" {len(trans.patterns)} patterns identified", " Phase 3: CITRINITAS — Generating rules...", f" {len(trans.rules)} rules created", " Phase 4: RUBEDO — Integration complete", "", f" ✓ TRANSMUTATION COMPLETE: {trans.id}", "", " GENERATED RULES:", " " + "-" * 50, ]) for rule in trans.rules[:5]: lines.append(f" [{rule.rule_type}] {rule.id}") if len(trans.rules) > 5: lines.append(f" ... and {len(trans.rules) - 5} more") lines.extend([ "", " Every attack becomes a lesson.", "" ]) return CommandResponse( id=f"cmd-{uuid.uuid4().hex[:8]}", status="ok", lines=lines, effects={ "tem": { "transmutations": state.tem_transmutations, "rules": state.tem_rules } } ) except Exception as e: lines.extend([ f" ✗ Transmutation failed: {str(e)}", "" ]) return CommandResponse( id=f"cmd-{uuid.uuid4().hex[:8]}", status="error", lines=lines, error=str(e) ) async def cmd_tem_rules(req) -> "CommandResponse": """List generated rules.""" rule_type = req.args[0] if req.args else None rules = tem_engine.get_rules(rule_type=rule_type) lines = [ "", f" ⚗ TEM GENERATED RULES ({len(rules)} total)", "", ] if rules: lines.append(" ID TYPE PATTERN CONF DEPLOYED") lines.append(" " + "-" * 75) for r in rules[:15]: deployed = "✓" if r['deployed'] else "" lines.append(f" {r['id']:<20} {r['rule_type']:<12} {r['pattern_id']:<20} {r['confidence']:.2f} {deployed}") if len(rules) > 15: lines.append(f" ... and {len(rules) - 15} more") else: lines.append(" (No rules generated yet)") lines.append(" Run: tem transmute ") lines.append("") return CommandResponse( id=f"cmd-{uuid.uuid4().hex[:8]}", status="ok", lines=lines, effects={} ) async def cmd_tem_history(req) -> "CommandResponse": """Show transmutation history.""" transmutations = tem_engine.get_transmutations(limit=10) lines = [ "", f" ⚗ TRANSMUTATION HISTORY ({len(transmutations)} recent)", "", ] if transmutations: lines.append(" ID THREAT PHASE COMPLETED") lines.append(" " + "-" * 75) for t in transmutations: lines.append(f" {t['id']:<20} {t['threat_id']:<20} {t['phase']:<12} {t['completed_at'][:19]}") else: lines.append(" (No transmutations yet)") lines.append("") return CommandResponse( id=f"cmd-{uuid.uuid4().hex[:8]}", status="ok", lines=lines, effects={} ) # ═══════════════════════════════════════════════════════════════════════════════ # ADD TO COMMANDS REGISTRY # ═══════════════════════════════════════════════════════════════════════════════ """ COMMANDS = { # ... existing commands ... "tem status": cmd_tem_status, "tem transmute": cmd_tem_transmute, "tem rules": cmd_tem_rules, "tem history": cmd_tem_history, } """ # ═══════════════════════════════════════════════════════════════════════════════ # ADD TO watch_shield_results() - AUTO-TRANSMUTATION # ═══════════════════════════════════════════════════════════════════════════════ """ async def watch_shield_results(): '''Background task: watch for new scan results and auto-transmute.''' watch_path = config.SHIELD_DIR async for changes in awatch(watch_path): for change_type, path in changes: if Path(path).name == "latest_scan.json": # ... existing shield event broadcasting ... # AUTO-TRANSMUTE: Feed findings to TEM try: scan_data = json.loads(Path(path).read_text()) findings = scan_data.get("findings", []) for finding in findings: # Check if already transmuted existing = tem_engine.get_transmutations() if not any(t['threat_id'] == finding.get('id') for t in existing): # New finding - transmute it await tem_engine.transmute(finding) await manager.broadcast({ "type": "console.line", "line": f"⚗ TEM: Auto-transmuted {finding.get('id')}", "lineType": "info" }) except Exception as e: print(f"TEM auto-transmute error: {e}") """ # ═══════════════════════════════════════════════════════════════════════════════ # ADD TO status EFFECTS # ═══════════════════════════════════════════════════════════════════════════════ """ # In cmd_status(), add to effects: effects={ # ... existing effects ... "tem": { "transmutations": state.tem_transmutations, "rules": state.tem_rules, "phase": state.tem_phase } } # And add to status dashboard lines: lines.append(f" TEM ● ACTIVE {state.tem_rules} rules") """ # ═══════════════════════════════════════════════════════════════════════════════ # WEBSOCKET EVENT HANDLING (add to handleWsMessage in console JS) # ═══════════════════════════════════════════════════════════════════════════════ """ // Add to handleWsMessage() in iotek-nexus-fixed.html: case 'tem.transmute': state.temRules = msg.rules_generated; printLine(`⚗ TEM: Transmuted ${msg.threat_id} → ${msg.rules_generated} rules`, 'info'); break; case 'tem.phase': printLine(`⚗ TEM Phase: ${msg.phase.toUpperCase()}`, 'dimmed'); break; """ # ═══════════════════════════════════════════════════════════════════════════════ # COMPLETE INTEGRATION EXAMPLE # ═══════════════════════════════════════════════════════════════════════════════ FULL_INTEGRATION_DIFF = """ To integrate TEM into your existing offsec_mcp_live.py: 1. Add import at top: from tem import TemEngine 2. Add to SystemState: tem_transmutations: int = 0 tem_rules: int = 0 tem_phase: str = "idle" 3. In lifespan(), after init_db(): tem_engine = TemEngine(config.VAULTMESH_ROOT) # ... wire callbacks ... 4. Add command handlers (copy from above) 5. Add to COMMANDS dict: "tem status": cmd_tem_status, "tem transmute": cmd_tem_transmute, "tem rules": cmd_tem_rules, "tem history": cmd_tem_history, 6. (Optional) Enable auto-transmutation in watch_shield_results() The system now learns from every threat. """ print(FULL_INTEGRATION_DIFF)