"""Configuration loading and path management for the WolfPanel Agent. Every filesystem location and runtime setting lives here so the rest of the codebase never hard-codes a path. Production defaults follow the Linux layout documented in the README. For local development (including non-Linux machines) set WOLFPANEL_DEV_HOME to rebase every directory under a single writable folder. """ from __future__ import annotations import os from dataclasses import dataclass, field from pathlib import Path import state as state_module # Agent version is the single source of truth for the running build. # Read version from VERSION file if exists, otherwise fallback. try: _version_file = Path(__file__).parent.parent / "VERSION" if _version_file.is_file(): AGENT_VERSION = _version_file.read_text(encoding="utf-8").strip() else: AGENT_VERSION = "0.1.0-dev" except Exception: AGENT_VERSION = "0.1.0-dev" # Lifecycle states the agent/panel can move through. v1 does not implement the # full state machine, but the names exist so other modules can reference them # instead of magic strings. LIFECYCLE_STATES = ( "pending_install", "pairing_waiting", "registered", "online", "offline", "updating", "rollback", "degraded", "revoked", "uninstalled", ) # Production layout. These can be overridden individually via env vars, or all # at once by setting WOLFPANEL_DEV_HOME (handy on Windows/macOS dev machines). _DEFAULT_ETC = "/etc/wolfpanel" _DEFAULT_OPT = "/opt/wolfpanel" _DEFAULT_VAR = "/var/lib/wolfpanel" _DEFAULT_LOG = "/var/log/wolfpanel" def _resolve_dirs() -> dict[str, Path]: """Resolve the four base directories, honouring dev overrides.""" dev_home = os.environ.get("WOLFPANEL_DEV_HOME") if dev_home: base = Path(dev_home) return { "etc": base / "etc", "opt": base / "opt", "var": base / "var", "log": base / "log", } return { "etc": Path(os.environ.get("WOLFPANEL_ETC_DIR", _DEFAULT_ETC)), "opt": Path(os.environ.get("WOLFPANEL_OPT_DIR", _DEFAULT_OPT)), "var": Path(os.environ.get("WOLFPANEL_VAR_DIR", _DEFAULT_VAR)), "log": Path(os.environ.get("WOLFPANEL_LOG_DIR", _DEFAULT_LOG)), } @dataclass class Config: """Resolved agent configuration and all derived paths.""" api_url: str release_channel: str release_base: str server_id: str agent_version: str heartbeat_interval: int api_mock: bool connect_url: str mysql_user: str mysql_password: str mysql_socket: str update_timeout: int backup_retention: int update_retention: int download_timeout: int etc_dir: Path opt_dir: Path var_dir: Path log_dir: Path # Derived paths (populated in __post_init__). conf_file: Path = field(init=False) secrets_dir: Path = field(init=False) agent_token_file: Path = field(init=False) state_file: Path = field(init=False) spool_dir: Path = field(init=False) jobs_dir: Path = field(init=False) discovery_dir: Path = field(init=False) cache_dir: Path = field(init=False) agent_log: Path = field(init=False) jobs_log: Path = field(init=False) update_log: Path = field(init=False) status: str = field(default="") last_error: str = field(default="") registration_pending: bool = field(default=False) def __post_init__(self) -> None: self.conf_file = self.etc_dir / "agent.conf" self.secrets_dir = self.etc_dir / "secrets" self.agent_token_file = self.secrets_dir / "agent.token" self.spool_dir = self.var_dir / "spool" self.jobs_dir = self.var_dir / "jobs" self.discovery_dir = self.var_dir / "discovery" self.cache_dir = self.var_dir / "cache" # Runtime state lives under var/, never in the systemd EnvironmentFile. self.state_file = self.var_dir / "state.json" self.agent_log = self.log_dir / "agent.log" self.jobs_log = self.log_dir / "jobs.log" self.update_log = self.log_dir / "update.log" @property def manifest_url(self) -> str: """URL of the channel's latest-version manifest (update check).""" return f"{self.release_base.rstrip('/')}/{self.release_channel}/latest.json" def _parse_conf_file(path: Path) -> dict[str, str]: """Parse a simple KEY=VALUE env-style file. Missing file -> empty dict.""" values: dict[str, str] = {} if not path.is_file(): return values for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") values[key.strip()] = value.strip().strip('"').strip("'") return values def load_config() -> Config: """Build a Config from (in priority order) env vars, then the conf file. Environment variables always win so operators and tests can override the on-disk configuration without editing files. """ dirs = _resolve_dirs() file_values = _parse_conf_file(dirs["etc"] / "agent.conf") def get(key: str, default: str) -> str: return os.environ.get(key, file_values.get(key, default)) api_url = get("WOLFPANEL_API_URL", "https://api.wolfpanel.net") api_mock = get("WOLFPANEL_API_MOCK", "0").lower() in ("1", "true", "yes") server_id = get("WOLFPANEL_SERVER_ID", "") etc_dir = dirs["etc"] token_file = etc_dir / "secrets" / "agent.token" has_token = token_file.is_file() # Runtime status/error/flags come from state.json, NOT agent.conf, so the # systemd EnvironmentFile stays static. The version is always the VERSION # file (AGENT_VERSION); a stale WOLFPANEL_AGENT_VERSION in agent.conf must # never be trusted as the source of truth. runtime = state_module.load_state(dirs["var"] / "state.json") default_status = "registered" if (server_id and has_token) else "pending_install" status = runtime.get("status") or default_status last_error = runtime.get("last_error", "") registration_pending = bool(runtime.get("registration_pending", False)) mysql_user = get("WOLFPANEL_MYSQL_USER", get("user", "root")) mysql_password = get("WOLFPANEL_MYSQL_PASSWORD", get("password", "")) mysql_socket = get("WOLFPANEL_MYSQL_SOCKET", get("socket", "/var/run/mysqld/mysqld.sock")) return Config( api_url=api_url, release_channel=get("WOLFPANEL_RELEASE_CHANNEL", "stable"), # Must match install.sh's RELEASE_BASE default (see docs/RELEASE_GUIDE.md); # the installer and the running agent derive the same manifest URL. release_base=get("WOLFPANEL_RELEASE_BASE", "https://downloads.wolfpanel.net/agent"), # v1 identity is the backend-assigned server_id (no agent_id in v1). server_id=server_id, # Single source of truth for the running build is the VERSION file. agent_version=AGENT_VERSION, heartbeat_interval=int(get("WOLFPANEL_HEARTBEAT_INTERVAL", "60")), api_mock=api_mock, # Browser pairing endpoint. No live default domain yet — configurable so # we never hardcode a dead host. Empty means "pairing flow not set up". connect_url=get("WOLFPANEL_CONNECT_URL", ""), mysql_user=mysql_user, mysql_password=mysql_password, mysql_socket=mysql_socket, update_timeout=int(get("WOLFPANEL_UPDATE_TIMEOUT", "60")), backup_retention=int(get("WOLFPANEL_BACKUP_RETENTION", "5")), update_retention=int(get("WOLFPANEL_UPDATE_RETENTION", "3")), download_timeout=int(get("WOLFPANEL_DOWNLOAD_TIMEOUT", "30")), etc_dir=dirs["etc"], opt_dir=dirs["opt"], var_dir=dirs["var"], log_dir=dirs["log"], status=status, last_error=last_error, registration_pending=registration_pending, )