""" Mythos diagnostics — capability availability matrix + lightweight self-checks. Importable from anywhere (no heavy side-effects). Used by: * ``mythos.__main__`` (the ``python -m mythos`` self-test CLI) * ``app.py`` (the Gradio "🜲 Mythos" tab status box) * ``mythos.api.fastapi_server`` health endpoint hooks """ from __future__ import annotations from typing import Any def _probe(modpath: str, attr: str) -> dict[str, Any]: try: mod = __import__(modpath, fromlist=[attr]) cls = getattr(mod, attr) inst = cls() ok = bool(getattr(inst, "available", lambda: True)()) return {"available": ok} except Exception as exc: # noqa: BLE001 return {"available": False, "error": f"{type(exc).__name__}: {exc}"} def availability_matrix() -> dict[str, dict[str, Any]]: """Return ``{component_name: {available: bool, error?: str}}``.""" return { "static.treesitter": _probe("mythos.static.treesitter_cpg", "TreeSitterCPG"), "static.joern": _probe("mythos.static.joern_bridge", "JoernBridge"), "static.codeql": _probe("mythos.static.codeql_bridge", "CodeQLBridge"), "static.semgrep": _probe("mythos.static.semgrep_bridge", "SemgrepBridge"), "dynamic.aflpp": _probe("mythos.dynamic.aflpp_runner", "AFLPlusPlusRunner"), "dynamic.klee": _probe("mythos.dynamic.klee_runner", "KLEERunner"), "dynamic.qemu": _probe("mythos.dynamic.qemu_harness", "QEMUHarness"), "dynamic.frida": _probe("mythos.dynamic.frida_instr", "FridaInstrumenter"), "dynamic.gdb": _probe("mythos.dynamic.gdb_automation", "GDBAutomation"), "exploit.pwntools": _probe("mythos.exploit.pwntools_synth", "PwntoolsSynth"), "exploit.rop": _probe("mythos.exploit.rop_chain", "ROPChainBuilder"), "exploit.heap": _probe("mythos.exploit.heap_exploit", "HeapExploitKit"), "exploit.privesc": _probe("mythos.exploit.privesc_kb", "PrivEscKB"), } def reasoning_check() -> dict[str, Any]: from .reasoning.probabilistic import HypothesisEngine from .reasoning.attack_graph import AttackGraph he = HypothesisEngine(seed=7) hyps = he.sample({"languages": ["python", "c"], "frameworks": ["flask"], "dependencies": ["pickle", "yaml"]}, n=5) g = AttackGraph() for h in hyps: g.add_hypothesis(h) g.connect() return {"backend": getattr(he, "backend", "?"), "n_hypotheses": len(hyps), "graph_nodes": len(g.nodes), "graph_edges": len(g.edges)} def learning_check() -> dict[str, Any]: from .learning.rl_planner import RLPlanner from .learning.episodic_memory import EpisodicMemory from .learning.mlflow_tracker import MLflowTracker return {"rl_backend": RLPlanner().backend, "episodic": bool(EpisodicMemory()), "mlflow_backend": MLflowTracker(experiment="self-test").backend} def api_check() -> dict[str, Any]: from .api import fastapi_server return {"fastapi_available": bool(fastapi_server.app), "title": getattr(fastapi_server.app, "title", None) if fastapi_server.app else None} def mcp_check() -> dict[str, Any]: out: dict[str, Any] = {} for name in ("static_analysis_mcp", "dynamic_analysis_mcp", "exploit_generation_mcp", "vulnerability_database_mcp", "web_security_mcp", "reconnaissance_mcp", # ARCHITECT additions "browser_agent_mcp", "scope_parser_mcp", "subdomain_enum_mcp", "httpx_probe_mcp", "shodan_mcp", "wayback_mcp", "frida_runtime_mcp", "ghidra_bridge_mcp", "can_bus_mcp", "sdr_analysis_mcp"): try: mod = __import__(f"mythos.mcp.{name}", fromlist=["server"]) out[name] = {"tools": [t["name"] for t in mod.server.list_tools()]} except Exception as exc: # noqa: BLE001 out[name] = {"error": f"{type(exc).__name__}: {exc}"} return out