"""Bezpečné čtení a zápis hlavního AppFactory env souboru (/opt/appfactory/config/appfactory.env). Zdroj pravdy zůstává soubor na disku — nikdy se neukládá do DB. Modul pracuje výhradně s pevnou cestou APPFACTORY_ENV; žádná jiná cesta není povolená. """ import os import re import shutil from datetime import datetime from app.config import APPFACTORY_ENV ENV_PATH = APPFACTORY_ENV # Povolený tvar klíče (shell env konvence). KEY_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$") # Hodnota, kterou lze zapsat bez uvozovek (žádné mezery ani shell-speciální znaky). _SIMPLE_VALUE_RE = re.compile(r"^[A-Za-z0-9_./:@%+,=-]*$") def _parse_line(line: str): """Vrátí (key, value) pro řádek typu KEY=value, jinak None (komentář/prázdné/neparsovatelné).""" stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in line: return None key, _, raw_value = line.partition("=") key = key.strip() if not KEY_RE.match(key): return None return key, _parse_value(raw_value) def _unescape_double(value: str) -> str: out = [] i = 0 while i < len(value): char = value[i] if char == "\\" and i + 1 < len(value) and value[i + 1] in '"\\$`': out.append(value[i + 1]) i += 2 else: out.append(char) i += 1 return "".join(out) def _parse_value(raw: str) -> str: value = raw.strip() if len(value) >= 2 and value[0] == '"' and value[-1] == '"': return _unescape_double(value[1:-1]) if len(value) >= 2 and value[0] == "'" and value[-1] == "'": return value[1:-1] return value def format_value(value: str) -> str: """Naformátuje hodnotu pro shell env soubor — prosté hodnoty bez uvozovek, hodnoty s mezerami/speciálními znaky v uvozovkách s bezpečným escapem.""" if value == "": return "" if _SIMPLE_VALUE_RE.match(value): return value escaped = ( value.replace("\\", "\\\\") .replace('"', '\\"') .replace("$", "\\$") .replace("`", "\\`") ) return f'"{escaped}"' def read_entries() -> list[tuple[str, str]]: """Načte uspořádaný seznam (key, value) ze souboru. Duplicitní klíče bere jen poprvé.""" try: with open(ENV_PATH, "r", encoding="utf-8") as handle: lines = handle.readlines() except FileNotFoundError: return [] entries: list[tuple[str, str]] = [] seen: set[str] = set() for line in lines: parsed = _parse_line(line) if not parsed: continue key, value = parsed if key in seen: continue seen.add(key) entries.append((key, value)) return entries def save_entries(desired_pairs: list[tuple[str, str]]) -> str | None: """Zapíše nový obsah souboru z požadovaného uspořádaného seznamu (key, value). Zachová komentáře, prázdné řádky a původní pořadí klíčů. Smazané klíče vypustí, nové přidá na konec. Před zápisem vytvoří časově označený backup. Vrací cestu backupu. """ desired_map = dict(desired_pairs) desired_order = [key for key, _ in desired_pairs] try: with open(ENV_PATH, "r", encoding="utf-8") as handle: original_lines = handle.readlines() except FileNotFoundError: original_lines = [] out_lines: list[str] = [] emitted: set[str] = set() for line in original_lines: parsed = _parse_line(line) if not parsed: out_lines.append(line.rstrip("\n")) continue key, _value = parsed if key not in desired_map or key in emitted: continue # smazaný klíč nebo duplicitní původní řádek out_lines.append(f"{key}={format_value(desired_map[key])}") emitted.add(key) for key in desired_order: if key not in emitted: out_lines.append(f"{key}={format_value(desired_map[key])}") emitted.add(key) content = "\n".join(out_lines) + "\n" backup_path = None if os.path.exists(ENV_PATH): timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") backup_path = f"{ENV_PATH}.bak-{timestamp}" shutil.copy2(ENV_PATH, backup_path) tmp_path = f"{ENV_PATH}.tmp-{os.getpid()}" with open(tmp_path, "w", encoding="utf-8", newline="\n") as handle: handle.write(content) os.replace(tmp_path, ENV_PATH) return backup_path