85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""The Brain — central registry for Sanad_lite.
|
|
|
|
Lite scope: only audio + voice client + callback whitelist. Motion, macro,
|
|
skill-execution, and live-voice trigger pieces were removed when those
|
|
subsystems left the project.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib
|
|
from typing import Any, Callable
|
|
|
|
from Project.Sanad.core.event_bus import bus # noqa: F401 (kept for downstream emitters)
|
|
from Project.Sanad.core.logger import get_logger
|
|
from Project.Sanad.core.skill_registry import SkillRegistry
|
|
|
|
log = get_logger("brain")
|
|
|
|
# Whitelist of module path prefixes allowed for skill callbacks.
|
|
# Prevents arbitrary code execution via dashboard-editable skills.json.
|
|
from Project.Sanad.core.config_loader import section as _cfg_section
|
|
_BRAIN_CFG = _cfg_section("core", "brain")
|
|
ALLOWED_CALLBACK_PREFIXES = tuple(_BRAIN_CFG.get("allowed_callback_prefixes", [
|
|
"Project.Sanad.voice.",
|
|
"voice.",
|
|
]))
|
|
|
|
|
|
class Brain:
|
|
"""Lite Brain — just the bits the surviving routes need."""
|
|
|
|
def __init__(self):
|
|
self.registry = SkillRegistry()
|
|
self._lock = asyncio.Lock()
|
|
self._voice = None # gemini.client.GeminiVoiceClient
|
|
self._audio_mgr = None # voice.audio_manager.AudioManager
|
|
|
|
# -- dependency injection --
|
|
|
|
def attach_voice(self, client):
|
|
self._voice = client
|
|
log.info("Voice client attached")
|
|
|
|
def attach_audio_manager(self, mgr):
|
|
self._audio_mgr = mgr
|
|
log.info("Audio manager attached")
|
|
|
|
# -- callback resolution (security whitelist) --
|
|
|
|
def _resolve_callback(self, callback_str: str) -> Callable | None:
|
|
"""Resolve 'module.submodule:function_name' → callable.
|
|
|
|
SECURITY: only modules under ALLOWED_CALLBACK_PREFIXES may be imported.
|
|
Skill JSON is dashboard-editable and otherwise an arbitrary-import RCE.
|
|
"""
|
|
if not callback_str:
|
|
return None
|
|
if ":" not in callback_str:
|
|
log.error("Invalid callback (missing ':'): %s", callback_str)
|
|
return None
|
|
module_path, func_name = callback_str.rsplit(":", 1)
|
|
if not any(module_path.startswith(prefix) or module_path == prefix.rstrip(".")
|
|
for prefix in ALLOWED_CALLBACK_PREFIXES):
|
|
log.error(
|
|
"Callback %s rejected — module '%s' not in whitelist",
|
|
callback_str, module_path,
|
|
)
|
|
return None
|
|
try:
|
|
mod = importlib.import_module(module_path)
|
|
return getattr(mod, func_name)
|
|
except Exception:
|
|
log.exception("Cannot resolve callback '%s'", callback_str)
|
|
return None
|
|
|
|
# -- status --
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
return {
|
|
"voice_attached": self._voice is not None,
|
|
"audio_manager_attached": self._audio_mgr is not None,
|
|
"total_skills": len(self.registry.list_skills()),
|
|
}
|