AirlockTools/utils/configmanager.py

#    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)