Files
NexusOS/synapse/nexus_config.py
T
AthenaandCursor fe12eb1821 feat(security): bind services to loopback with Host + CORS allowlists
The Synapse backend and memory service bound 0.0.0.0 with wildcard CORS and no
auth, exposing the full unauthenticated admin/data API to the LAN. Default the
uvicorn bind to 127.0.0.1 (NEXUS_BIND_HOST override), scope CORS to known local
origins instead of "*", and add TrustedHostMiddleware to reject foreign Host
headers (which defeats DNS-rebinding, something same-origin CORS cannot stop).

NEXUS_ALLOWED_HOSTS / NEXUS_ALLOWED_ORIGINS allow opt-in LAN exposure, intended
to be paired with real authentication.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-07 09:40:12 -05:00

186 lines
6.3 KiB
Python

# config.py
from __future__ import annotations
import os
from pathlib import Path
from typing import Dict, Any
# --- ENV ---
try:
from dotenv import load_dotenv # optional
load_dotenv()
except Exception:
pass
# --- PROJECT ROOT ---
PROJECT_ROOT = Path(__file__).resolve().parent.parent
# --- VERSION (single source of truth: the VERSION file at the repo root) ---
try:
VERSION = (PROJECT_ROOT / "VERSION").read_text(encoding="utf-8").strip() or "0.0.0"
except Exception:
VERSION = "0.0.0"
# --- MODEL DEFAULTS ---
# Single source of truth for the three models NexusOS ships with. The installers
# pull and pin these, the runtime falls back to them; keeping them in one place
# is what stops installer and backend from drifting apart.
#
# Chat: llama3.1:8b - a strong non-reasoning instruct model (~4.9 GB; overflows a
# 4 GB GPU into CPU/RAM). Chosen over Qwen3 because Qwen3 is a reasoning model:
# smart only with its slow <think> step, weak without it.
# Memory: mistral - the curator that extracts facts and titles conversations.
# Embed: nomic-embed-text - powers semantic recall of past conversations
# (OllamaManager.embed / store.semantic_search_conversations). Without it,
# recall silently degrades to lexical substring matching.
DEFAULT_CHAT_MODEL = "llama3.1:8b"
DEFAULT_MEMORY_MODEL = "mistral:latest"
DEFAULT_EMBED_MODEL = "nomic-embed-text"
# --- CORE DIRECTORIES ---
DATA_DIR = PROJECT_ROOT / "data"
MODELS_DIR = PROJECT_ROOT / "models"
RUNTIME_DIR = PROJECT_ROOT / "runtime"
MEMORY_DIR = PROJECT_ROOT / "synapse" / "memory"
LOGS_DIR = RUNTIME_DIR / "logs"
CACHE_DIR = RUNTIME_DIR / "cache"
TEMP_DIR = RUNTIME_DIR / "tmp"
# --- APPLICATION SUBSYSTEM DIRECTORIES ---
PLAYBOOK_DIR = DATA_DIR / "playbooks" # YAML playbook files (PlaybookFileStore)
UPLOADS_DIR = DATA_DIR / "uploads"
EXPORTS_DIR = DATA_DIR / "exports"
# --- DATABASE / STORAGE FILES (match your repo) ---
MEMORY_DB = MEMORY_DIR / "memory.db"
# --- LOG FILES ---
BACKEND_LOG = RUNTIME_DIR / "backend.log"
OLLAMA_LOG = LOGS_DIR / "ollama.log"
CHAT_LOG = LOGS_DIR / "chat.log"
# --- ENSURE REQUIRED DIRECTORIES EXIST ---
for d in (
DATA_DIR,
MODELS_DIR,
RUNTIME_DIR,
LOGS_DIR,
MEMORY_DIR,
CACHE_DIR,
TEMP_DIR,
PLAYBOOK_DIR,
UPLOADS_DIR,
EXPORTS_DIR,
):
d.mkdir(parents=True, exist_ok=True)
# --- PATH ACCESSOR (fail-fast) ---
def path(name: str) -> Path:
"""
Return a Path for a known name. Raises KeyError if name is unknown.
"""
mapping = {
"root": PROJECT_ROOT,
"data": DATA_DIR,
"models": MODELS_DIR,
"runtime": RUNTIME_DIR,
"logs": LOGS_DIR,
"memory": MEMORY_DIR,
"cache": CACHE_DIR,
"tmp": TEMP_DIR,
"playbooks": PLAYBOOK_DIR,
"uploads": UPLOADS_DIR,
"exports": EXPORTS_DIR,
"memory_db": MEMORY_DB,
"backend_log": BACKEND_LOG,
"ollama_log": OLLAMA_LOG,
"chat_log": CHAT_LOG,
}
try:
return mapping[name]
except KeyError:
raise KeyError(f"Unknown config path name: {name}")
# --- Settings class and exported instance ---
class Settings:
"""
Lightweight settings container. Use `settings` instance for runtime access,
or `Settings` class for typing/tests.
"""
def __init__(self) -> None:
self.version: str = VERSION
self.project_root: Path = PROJECT_ROOT
self.data_dir: Path = DATA_DIR
self.models_dir: Path = MODELS_DIR
self.runtime_dir: Path = RUNTIME_DIR
self.memory_dir: Path = MEMORY_DIR
self.logs_dir: Path = LOGS_DIR
# DB files
self.memory_db: Path = MEMORY_DB
# Logs
self.backend_log: Path = BACKEND_LOG
self.ollama_log: Path = OLLAMA_LOG
self.chat_log: Path = CHAT_LOG
# Env overrides
self.ollama_host: str = os.getenv("OLLAMA_HOST", "http://127.0.0.1:11434")
self.ollama_timeout: int = int(os.getenv("OLLAMA_TIMEOUT", "120"))
def as_dict(self) -> Dict[str, Any]:
return {
"version": self.version,
"project_root": str(self.project_root),
"data_dir": str(self.data_dir),
"models_dir": str(self.models_dir),
"runtime_dir": str(self.runtime_dir),
"memory_dir": str(self.memory_dir),
"memory_db": str(self.memory_db),
"ollama_host": self.ollama_host,
"ollama_timeout": self.ollama_timeout,
}
# --- local-access allowlists (shared by the backend + memory FastAPI apps) ---
# The REST APIs are unauthenticated, so they are meant to be reached only from
# this machine. Two independent browser-side defenses depend on these lists:
# * ALLOWED_ORIGINS drives CORS — blocks a malicious page from *reading*
# responses cross-origin (was previously "*", which let any site read them).
# * ALLOWED_HOSTS drives TrustedHostMiddleware — rejects a foreign Host header,
# which is what stops DNS-rebinding (same-origin from the browser's view, so
# CORS can't help there).
# Both accept a comma-separated env override for the intentional-LAN case, to be
# paired with real auth. NEXUS_ALLOWED_HOSTS=* disables the Host check.
def _csv_env(name: str, default: list) -> list:
raw = os.getenv(name, "").strip()
return [p.strip() for p in raw.split(",") if p.strip()] if raw else list(default)
_LOCAL_HOSTS = ["localhost", "127.0.0.1", "[::1]", "::1", "testserver"]
_LOCAL_ORIGINS = [
f"http://{h}:{p}"
for h in ("localhost", "127.0.0.1")
for p in (8000, 8001, 5173)
]
ALLOWED_HOSTS = _csv_env("NEXUS_ALLOWED_HOSTS", _LOCAL_HOSTS)
ALLOWED_ORIGINS = _csv_env("NEXUS_ALLOWED_ORIGINS", _LOCAL_ORIGINS)
# exported instance
settings = Settings()
# explicit exports for static checkers and IDEs
__all__ = ["Settings", "settings", "path", "VERSION",
"DEFAULT_CHAT_MODEL", "DEFAULT_MEMORY_MODEL", "DEFAULT_EMBED_MODEL",
"PROJECT_ROOT", "DATA_DIR", "MODELS_DIR", "RUNTIME_DIR",
"MEMORY_DIR", "LOGS_DIR", "PLAYBOOK_DIR", "UPLOADS_DIR",
"EXPORTS_DIR", "MEMORY_DB",
"BACKEND_LOG", "OLLAMA_LOG", "CHAT_LOG",
"ALLOWED_HOSTS", "ALLOWED_ORIGINS"]
# --- quick runtime sanity check when run directly (no side effects on import) ---
if __name__ == "__main__":
print("Config paths:")
for key in ("root", "data", "models", "runtime", "memory", "memory_db"):
print(f" {key}: {path(key)}")