fix(server, client): auto-resolve db, cert, state and config paths to ProgramData when running in read-only directories
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 1m35s
Release Binaries & Packages / Build & Release Standalone Binaries (Linux & Windows Cross-Compilation) (push) Canceled after 2m2s

This commit is contained in:
2026-09-26 19:56:37 +02:00
parent fd34fe9911
commit 48bb279601
2 changed files with 230 additions and 125 deletions
+72 -8
View File
@@ -384,11 +384,45 @@ def generate_server_keypair(server_name: str):
return private_key_armored, public_key_armored, fingerprint
def get_default_data_dir(app_name: str = "LOGAR-Server") -> str:
"""Returns a suitable writable application data directory."""
if sys.platform == "win32":
prog_data = os.environ.get("ProgramData", r"C:\ProgramData")
target = os.path.join(prog_data, app_name)
cwd = os.getcwd()
if os.path.exists(target) or "Program Files" in cwd or not os.access(cwd, os.W_OK):
try:
os.makedirs(target, exist_ok=True)
return target
except Exception:
pass
return os.getcwd()
def resolve_writable_path(path: str, app_name: str = "LOGAR-Server", default_filename: str = "logar_state.db") -> str:
"""Resolves a file or directory path, falling back to ProgramData if running in a read-only directory (e.g. Program Files)."""
if not path:
path = default_filename
if os.path.isabs(path):
return path
cwd = os.getcwd()
if sys.platform == "win32" and ("Program Files" in cwd or not os.access(cwd, os.W_OK)):
data_dir = get_default_data_dir(app_name)
return os.path.join(data_dir, path)
return os.path.abspath(path)
def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
"""Loads existing server_config.json or creates a new one on first run."""
if os.path.exists(config_path):
print(f"[*] Loading server configuration from: {os.path.abspath(config_path)}")
with open(config_path, "r", encoding="utf-8") as f:
actual_config_path = config_path
if not os.path.isabs(actual_config_path) and not os.path.exists(actual_config_path):
data_dir_cfg = os.path.join(get_default_data_dir("LOGAR-Server"), os.path.basename(config_path))
if os.path.exists(data_dir_cfg):
actual_config_path = data_dir_cfg
if os.path.exists(actual_config_path):
print(f"[*] Loading server configuration from: {os.path.abspath(actual_config_path)}")
with open(actual_config_path, "r", encoding="utf-8") as f:
config = json.load(f)
if "enrollment_secret" not in config:
config["enrollment_secret"] = secrets.token_hex(24)
@@ -404,11 +438,22 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
config["admin_user"] = DEFAULT_ADMIN_USER
if "session_secret" not in config:
config["session_secret"] = secrets.token_hex(32)
if "db_path" not in config:
config["db_path"] = DEFAULT_DB_FILE
# Resolve relative db_path and cert_dir to writable paths
config["db_path"] = resolve_writable_path(config["db_path"], "LOGAR-Server", DEFAULT_DB_FILE)
config["cert_dir"] = resolve_writable_path(config["cert_dir"], "LOGAR-Server", "certs")
global SESSION_SECRET
if not os.environ.get("LOGAR_SESSION_SECRET"):
SESSION_SECRET = config["session_secret"]
return config
# Fallback to ProgramData if CWD is read-only / Program Files
if not os.path.isabs(config_path) and ("Program Files" in os.getcwd() or not os.access(os.getcwd(), os.W_OK)):
config_path = os.path.join(get_default_data_dir("LOGAR-Server"), os.path.basename(config_path))
print(f"[!] Config '{config_path}' not found. Initializing first-run configuration...")
server_name = "LOGAR-Cloud-Hub"
private_key, public_key, fingerprint = generate_server_keypair(server_name)
@@ -416,6 +461,9 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
enrollment_secret = secrets.token_hex(24)
session_sec = secrets.token_hex(32)
db_path = resolve_writable_path(DEFAULT_DB_FILE, "LOGAR-Server", DEFAULT_DB_FILE)
cert_dir = resolve_writable_path("certs", "LOGAR-Server", "certs")
config = {
"server_name": server_name,
"tcp_host": "0.0.0.0",
@@ -428,9 +476,9 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
"session_secret": session_sec,
"max_seats": 10,
"license_server_url": "https://logar.eibl.tech",
"cert_dir": "certs",
"cert_dir": cert_dir,
"tls_enabled": True,
"db_path": DEFAULT_DB_FILE,
"db_path": db_path,
"evaluation_window_hours": EVALUATION_WINDOW_HOURS,
"min_persistence_runs": RUN_THRESHOLD,
"server_fingerprint": fingerprint,
@@ -440,6 +488,7 @@ def load_or_init_config(config_path: str = CONFIG_FILE_NAME) -> Dict[str, Any]:
if not os.environ.get("LOGAR_SESSION_SECRET"):
SESSION_SECRET = session_sec
os.makedirs(os.path.dirname(os.path.abspath(config_path)), exist_ok=True)
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config, f, indent=2)
@@ -554,9 +603,13 @@ def format_relative_time(dt_str: Optional[str]) -> str:
return dt_str[:16] if dt_str else "N/A"
def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: int = 10):
def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: int = 10) -> str:
"""Initializes the SQLite schema for multi-run temporal tracking, client tracking, users, and license quota."""
conn = sqlite3.connect(db_path)
def _execute_schema(target_path: str):
target_dir = os.path.dirname(os.path.abspath(target_path))
if target_dir:
os.makedirs(target_dir, exist_ok=True)
conn = sqlite3.connect(target_path)
# 1. Active Issues
conn.execute("""
@@ -675,6 +728,17 @@ def init_db(db_path: str, enrollment_secret: Optional[str] = None, max_seats: in
conn.commit()
conn.close()
try:
_execute_schema(db_path)
return db_path
except (sqlite3.OperationalError, PermissionError, OSError) as e:
if "readonly" in str(e).lower() or "permission" in str(e).lower() or isinstance(e, PermissionError):
fallback_path = os.path.join(get_default_data_dir("LOGAR-Server"), "logar_state.db")
print(f"[!] Warning: Database path '{db_path}' is not writable ({e}). Falling back to '{fallback_path}'...")
_execute_schema(fallback_path)
return fallback_path
raise
def export_license_request_lic(config: Dict[str, Any]) -> Tuple[Dict[str, Any], str]:
"""Generates CA CSR and packages into exportable .lic dictionary and JSON string."""
@@ -2601,7 +2665,7 @@ def main():
args = parser.parse_args()
config = load_or_init_config(args.config)
init_db(
config["db_path"] = init_db(
config["db_path"],
enrollment_secret=config.get("enrollment_secret"),
max_seats=config.get("max_seats", 10)
+44 -3
View File
@@ -115,10 +115,40 @@ def get_tls_socket(hub_host: str, hub_port: int, cert_dir: str):
return ctx.wrap_socket(raw_sock, server_hostname=hub_host)
def get_default_data_dir(app_name: str = "LOGAR") -> str:
"""Returns a suitable writable application data directory."""
if sys.platform == "win32":
prog_data = os.environ.get("ProgramData", r"C:\ProgramData")
target = os.path.join(prog_data, app_name)
cwd = os.getcwd()
if os.path.exists(target) or "Program Files" in cwd or not os.access(cwd, os.W_OK):
try:
os.makedirs(target, exist_ok=True)
return target
except Exception:
pass
return os.getcwd()
def resolve_writable_path(path: str, app_name: str = "LOGAR", default_filename: str = "client_state.json") -> str:
"""Resolves a file or directory path, falling back to ProgramData if running in a read-only directory (e.g. Program Files)."""
if not path:
path = default_filename
if os.path.isabs(path):
return path
cwd = os.getcwd()
if sys.platform == "win32" and ("Program Files" in cwd or not os.access(cwd, os.W_OK)):
data_dir = get_default_data_dir(app_name)
return os.path.join(data_dir, path)
return os.path.abspath(path)
def get_state_path(config_path: str, custom_state_path: Optional[str] = None) -> str:
if custom_state_path:
return custom_state_path
config_dir = os.path.dirname(os.path.abspath(config_path))
if sys.platform == "win32" and ("Program Files" in config_dir or not os.access(config_dir, os.W_OK)):
return os.path.join(get_default_data_dir("LOGAR"), STATE_FILE_NAME)
return os.path.join(config_dir, STATE_FILE_NAME)
@@ -135,6 +165,7 @@ def load_state(state_path: str) -> dict:
def save_state(state_path: str, state: dict):
try:
os.makedirs(os.path.dirname(os.path.abspath(state_path)), exist_ok=True)
temp_path = f"{state_path}.tmp"
with open(temp_path, "w", encoding="utf-8") as f:
json.dump(state, f, indent=2)
@@ -154,13 +185,23 @@ def commit_state(state: dict, state_path: str):
def load_config(config_path: str = CONFIG_FILE_NAME):
if not os.path.exists(config_path):
actual_config_path = config_path
if not os.path.exists(actual_config_path):
data_dir_cfg = os.path.join(get_default_data_dir("LOGAR"), os.path.basename(config_path))
if os.path.exists(data_dir_cfg):
actual_config_path = data_dir_cfg
if not os.path.exists(actual_config_path):
raise FileNotFoundError(
f"Client configuration file not found at: {config_path}\n"
f"Generate one from the server using: python Server.py --create-client-config --client-out {config_path}"
)
with open(config_path, "r", encoding="utf-8") as f:
return json.load(f)
with open(actual_config_path, "r", encoding="utf-8") as f:
config = json.load(f)
if "cert_dir" in config:
config["cert_dir"] = resolve_writable_path(config["cert_dir"], "LOGAR", "certs")
return config
def get_machine_identifier() -> str: