346 lines
11 KiB
Python
346 lines
11 KiB
Python
# Copyright (C) 2025 James Brotosky, Brandon Wickline
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
from typing import Any, Callable, Optional, TypeVar
|
|
|
|
T = TypeVar("T")
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# System config keys - these are immutable and come from system_config.json (bundled in exe)
|
|
SYSTEM_CONFIG_KEYS = [
|
|
"URL",
|
|
"APPNAME",
|
|
"LOG_LEVEL",
|
|
"BAD_PATH_PARTS",
|
|
"BAD_PUBLISHERS",
|
|
"PUPS",
|
|
"PATH_EXCLUSION_CONST",
|
|
"MIN_FILES_FOR_PATH",
|
|
"VT_THREAT_TOLERANCE",
|
|
"POLICY_MAP_ENF_AUD",
|
|
]
|
|
|
|
# User config keys - these can be changed by the end user
|
|
USER_CONFIG_KEYS = [
|
|
"TELEMETRY", # User opt-in/out for telemetry
|
|
"TELEM_URL",
|
|
"TEXTUAL_THEME", # UI theme preference
|
|
"EXTRAS", # Feature flags
|
|
]
|
|
|
|
# In-memory config storage
|
|
_system_config = {}
|
|
_user_config = {}
|
|
|
|
|
|
def get_system_config_path() -> Path:
|
|
"""
|
|
Get path to system_config.json.
|
|
Priority:
|
|
1. Bundled in exe (_MEIPASS)
|
|
2. Next to this file (development)
|
|
"""
|
|
# Check inside bundled EXE directory first
|
|
bundled_dir = Path(getattr(sys, "_MEIPASS", ""))
|
|
bundled_path = bundled_dir / "system_config.json"
|
|
if bundled_path.exists():
|
|
return bundled_path
|
|
|
|
# Fallback to development location (next to this file)
|
|
return Path(__file__).parent.parent / "system_config.json"
|
|
|
|
|
|
def load_system_config() -> dict:
|
|
"""
|
|
Load system configuration from system_config.json.
|
|
This should only be called once at startup.
|
|
Returns the full system config dict.
|
|
"""
|
|
global _system_config
|
|
|
|
try:
|
|
config_path = get_system_config_path()
|
|
with open(config_path, "r") as f:
|
|
_system_config = json.load(f)
|
|
logger.debug(f"✅ Loaded system config from {config_path}")
|
|
except FileNotFoundError:
|
|
logger.warning("⚠️ system_config.json not found. Using minimal defaults.")
|
|
# Minimal defaults for development without system_config.json
|
|
_system_config = {
|
|
"APPNAME": "Loxide",
|
|
"LOG_LEVEL": "INFO",
|
|
"PATH_EXCLUSION_CONST": 4,
|
|
"MIN_FILES_FOR_PATH": 4,
|
|
"VT_THREAT_TOLERANCE": 4,
|
|
"POLICY_MAP_ENF_AUD": {},
|
|
}
|
|
|
|
return _system_config
|
|
|
|
|
|
def get_system_value(
|
|
key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
|
|
) -> Optional[T]:
|
|
"""
|
|
Get a value from system config (immutable).
|
|
|
|
Parameters:
|
|
key: The config key to retrieve
|
|
cast_type: Function to cast the value to desired type
|
|
default: Default value if key not found
|
|
|
|
Returns:
|
|
The config value cast to the desired type, or default
|
|
"""
|
|
value = _system_config.get(key)
|
|
if value is None:
|
|
logger.warning(f"System config key '{key}' not found.")
|
|
return default
|
|
|
|
try:
|
|
if isinstance(value, str):
|
|
value = value.strip("'\"")
|
|
return cast_type(value)
|
|
except (ValueError, TypeError):
|
|
logger.warning(
|
|
f"Invalid value for system key '{key}': {value}. Expected type {cast_type.__name__}."
|
|
)
|
|
return default
|
|
|
|
|
|
def get_system_json(key: str, default: Optional[dict] = None) -> dict:
|
|
"""
|
|
Get a JSON/dict value from system config.
|
|
Handles both dict values and JSON strings.
|
|
"""
|
|
if default is None:
|
|
default = {}
|
|
|
|
raw = _system_config.get(key, default)
|
|
if isinstance(raw, dict):
|
|
return raw
|
|
|
|
try:
|
|
return json.loads(raw)
|
|
except (json.JSONDecodeError, TypeError) as e:
|
|
logger.error(f"Failed to parse system JSON key '{key}': {e}")
|
|
return default
|
|
|
|
|
|
def get_system_list(key: str, default: Optional[list] = None) -> list:
|
|
"""
|
|
Get a list value from system config.
|
|
Handles both list values and JSON strings.
|
|
|
|
Parameters:
|
|
key: The config key to retrieve
|
|
default: Default value if key not found or parsing fails
|
|
|
|
Returns:
|
|
The list value or default
|
|
"""
|
|
if default is None:
|
|
default = []
|
|
|
|
raw = _system_config.get(key, default)
|
|
if isinstance(raw, list):
|
|
return raw
|
|
|
|
try:
|
|
result = json.loads(raw) if isinstance(raw, str) else raw
|
|
if isinstance(result, list):
|
|
return result
|
|
logger.warning(f"System config key '{key}' is not a list: {type(result)}")
|
|
return default
|
|
except (json.JSONDecodeError, TypeError) as e:
|
|
logger.error(f"Failed to parse system list key '{key}': {e}")
|
|
return default
|
|
|
|
|
|
def load_user_config(config_dir: Path) -> dict:
|
|
"""
|
|
Load user configuration from user_config.json.
|
|
Creates the file with defaults if it doesn't exist.
|
|
|
|
Parameters:
|
|
config_dir: Directory containing user_config.json
|
|
|
|
Returns:
|
|
The user config dict
|
|
"""
|
|
global _user_config
|
|
|
|
user_config_path = config_dir / "user_config.json"
|
|
|
|
if not user_config_path.exists():
|
|
# Create default user config
|
|
default_user_config = {
|
|
"TELEMETRY": False,
|
|
"TELEM_URL": "",
|
|
"TEXTUAL_THEME": "gruvbox",
|
|
"EXTRAS": "NOTTODAY",
|
|
}
|
|
user_config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(user_config_path, "w") as f:
|
|
json.dump(default_user_config, f, indent=4)
|
|
logger.debug(f"Created default user config at {user_config_path}")
|
|
_user_config = default_user_config
|
|
else:
|
|
with open(user_config_path, "r") as f:
|
|
_user_config = json.load(f)
|
|
logger.debug(f"✅ Loaded user config from {user_config_path}")
|
|
|
|
return _user_config
|
|
|
|
|
|
def save_user_config(config_dir: Path, updates: dict) -> None:
|
|
"""
|
|
Save updates to user configuration.
|
|
Only keys in USER_CONFIG_KEYS are allowed.
|
|
|
|
Parameters:
|
|
config_dir: Directory containing user_config.json
|
|
updates: Dict of key-value pairs to update
|
|
"""
|
|
global _user_config
|
|
|
|
# Validate that only user-configurable keys are being updated
|
|
invalid_keys = [k for k in updates.keys() if k not in USER_CONFIG_KEYS]
|
|
if invalid_keys:
|
|
logger.error(f"Attempted to save invalid user config keys: {invalid_keys}")
|
|
raise ValueError(f"Cannot modify system config keys: {invalid_keys}")
|
|
|
|
# Update in-memory config
|
|
_user_config.update(updates)
|
|
|
|
# Write to file
|
|
user_config_path = config_dir / "user_config.json"
|
|
user_config_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(user_config_path, "w") as f:
|
|
json.dump(_user_config, f, indent=4)
|
|
|
|
logger.debug(f"✅ Saved user config to {user_config_path}: {updates}")
|
|
|
|
|
|
def get_user_value(
|
|
key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
|
|
) -> Optional[T]:
|
|
"""
|
|
Get a value from user config (mutable).
|
|
|
|
Parameters:
|
|
key: The config key to retrieve
|
|
cast_type: Function to cast the value to desired type
|
|
default: Default value if key not found
|
|
|
|
Returns:
|
|
The config value cast to the desired type, or default
|
|
"""
|
|
value = _user_config.get(key)
|
|
if value is None:
|
|
logger.warning(f"User config key '{key}' not found.")
|
|
return default
|
|
|
|
try:
|
|
if isinstance(value, str):
|
|
value = value.strip("'\"")
|
|
return cast_type(value)
|
|
except (ValueError, TypeError):
|
|
logger.warning(
|
|
f"Invalid value for user key '{key}': {value}. Expected type {cast_type.__name__}."
|
|
)
|
|
return default
|
|
|
|
|
|
def load_env(
|
|
key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
|
|
) -> Optional[T]:
|
|
"""
|
|
Safely retrieves an environment variable from .env and casts it to the desired type.
|
|
This should ONLY be used for runtime/dynamic values like WORKING_DIR.
|
|
|
|
For system config, use get_system_value().
|
|
For user config, use get_user_value().
|
|
|
|
Parameters:
|
|
key: The name of the environment variable
|
|
cast_type: Function to cast the value. Defaults to str
|
|
default: Default value if the variable is not set or invalid
|
|
|
|
Returns:
|
|
The casted value or the default
|
|
"""
|
|
value = os.getenv(key)
|
|
if value is None:
|
|
logger.debug(f"Environment variable '{key}' not set, using default.")
|
|
return default
|
|
|
|
try:
|
|
value = value.strip("'\"") # Strip surrounding quotes
|
|
return cast_type(value)
|
|
except (ValueError, TypeError):
|
|
logger.warning(
|
|
f"Invalid value for env var '{key}': {value}. Expected type {cast_type.__name__}."
|
|
)
|
|
return default
|
|
|
|
|
|
def load_env_json(key: str, default: str = "[]") -> Any:
|
|
"""
|
|
Load a JSON value from environment or system config.
|
|
|
|
DEPRECATED: This function is kept for backward compatibility.
|
|
- For system config lists (BAD_PUBLISHERS, PUPS, BAD_PATH_PARTS), use get_system_list()
|
|
- For system config dicts, use get_system_json()
|
|
- For actual .env JSON values, parse manually
|
|
|
|
This function automatically redirects known system config keys to system config.
|
|
"""
|
|
# Known system config list keys - redirect to system config
|
|
system_list_keys = ["BAD_PUBLISHERS", "PUPS", "BAD_PATH_PARTS"]
|
|
if key in system_list_keys:
|
|
logger.debug(f"Redirecting load_env_json('{key}') to get_system_list()")
|
|
return get_system_list(key, json.loads(default) if default else [])
|
|
|
|
# Known system config dict keys - redirect to system config
|
|
system_dict_keys = ["POLICY_MAP_ENF_AUD"]
|
|
if key in system_dict_keys:
|
|
logger.debug(f"Redirecting load_env_json('{key}') to get_system_json()")
|
|
return get_system_json(key, json.loads(default) if default else {})
|
|
|
|
# Fall back to reading from .env (backward compatibility for unknown keys)
|
|
raw = os.getenv(key, default)
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
try:
|
|
escaped = raw.encode("unicode_escape").decode("utf-8")
|
|
return json.loads(escaped)
|
|
except Exception as e:
|
|
logger.error(f"Failed to parse {key}: {e}")
|
|
return json.loads(default)
|
|
|
|
|
|
# Backwards compatibility aliases (deprecated - use get_system_value instead)
|
|
get_protected_value = get_system_value
|
|
get_protected_json = get_system_json
|
|
load_protected_config = load_system_config
|
|
PROTECTED_KEYS = SYSTEM_CONFIG_KEYS # For backwards compatibility
|