115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
CONFIG_PATH = Path(os.getenv("APP_CONFIG", "/app/config.json"))
|
|
|
|
DEFAULT_CONFIG: dict[str, Any] = {
|
|
"general": {
|
|
"title": "AssetManager",
|
|
"company_name": "",
|
|
"logo": "",
|
|
"custom_css": "",
|
|
},
|
|
"authentication": {
|
|
"mode": "none",
|
|
"session_secret_env": "SESSION_SECRET",
|
|
"ldap": {
|
|
"server": "",
|
|
"port": 389,
|
|
"use_ssl": False,
|
|
"start_tls": False,
|
|
"bind_dn": "",
|
|
"bind_password_env": "LDAP_BIND_PASSWORD",
|
|
"user_base_dn": "",
|
|
"user_filter": "(sAMAccountName={username})",
|
|
"display_name_attribute": "displayName",
|
|
"email_attribute": "mail"
|
|
}
|
|
},
|
|
"software": {
|
|
"dispatch_delay_seconds": 2,
|
|
"callback_base_url": "",
|
|
"callback_timeout_minutes": 30,
|
|
"callback_worker_count": 4,
|
|
"callback_test_timeout_seconds": 10,
|
|
"callback_test_verify_tls": True,
|
|
"debug_logging": True,
|
|
"debug_log_max_lines": 1000,
|
|
"automatic_retry_enabled": True,
|
|
"automatic_retry_max_age_hours": 12,
|
|
"automatic_retry_interval_seconds": 60,
|
|
"automatic_retry_max_retries": 3,
|
|
"inventory_exclusion_rules": []
|
|
},
|
|
"meshcentral": {
|
|
"enabled": False,
|
|
"url": "",
|
|
"username": "",
|
|
"password": "",
|
|
"password_env": "MESHCENTRAL_PASSWORD",
|
|
"tenant": "",
|
|
"meshctrl_path": "/opt/meshcentral/node_modules/meshcentral/meshctrl.js",
|
|
"verify_tls": True,
|
|
"fallback_category_id": 1,
|
|
"category_mapping": {"1": 1, "2": 1, "3": 1, "4": 1, "5": 1, "6": 1, "7": 1, "8": 1},
|
|
"timeout_seconds": 120,
|
|
"presence_enabled": True,
|
|
"presence_interval_seconds": 30,
|
|
"match_order": ["node_id", "serial_number"],
|
|
"create_missing_assets": True,
|
|
"mark_missing_devices": True,
|
|
"field_rules": {},
|
|
},
|
|
}
|
|
|
|
|
|
def _merge(base: dict[str, Any], supplied: dict[str, Any]) -> dict[str, Any]:
|
|
for section, values in supplied.items():
|
|
if isinstance(values, dict) and isinstance(base.get(section), dict):
|
|
base[section].update(values)
|
|
else:
|
|
base[section] = values
|
|
return base
|
|
|
|
|
|
def load_config() -> dict[str, Any]:
|
|
config = json.loads(json.dumps(DEFAULT_CONFIG))
|
|
if CONFIG_PATH.exists():
|
|
with CONFIG_PATH.open("r", encoding="utf-8") as handle:
|
|
supplied = json.load(handle)
|
|
_merge(config, supplied)
|
|
mesh = config["meshcentral"]
|
|
env_name = mesh.get("password_env")
|
|
if env_name and os.getenv(env_name):
|
|
mesh["password"] = os.getenv(env_name)
|
|
return config
|
|
|
|
|
|
def save_config(config: dict[str, Any]) -> None:
|
|
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
current: dict[str, Any] = {}
|
|
if CONFIG_PATH.exists():
|
|
with CONFIG_PATH.open("r", encoding="utf-8") as handle:
|
|
current = json.load(handle)
|
|
_merge(current, config)
|
|
temporary = CONFIG_PATH.with_suffix(CONFIG_PATH.suffix + ".tmp")
|
|
with temporary.open("w", encoding="utf-8") as handle:
|
|
json.dump(current, handle, ensure_ascii=False, indent=2)
|
|
handle.write("\n")
|
|
temporary.replace(CONFIG_PATH)
|
|
|
|
|
|
def public_config() -> dict[str, Any]:
|
|
config = load_config()
|
|
mesh = dict(config["meshcentral"])
|
|
mesh["password"] = "***" if mesh.get("password") else ""
|
|
return {
|
|
"config_path": str(CONFIG_PATH),
|
|
"general": dict(config.get("general", {})),
|
|
"authentication": dict(config.get("authentication", {})),
|
|
"software": dict(config.get("software", {})),
|
|
"meshcentral": mesh,
|
|
}
|