JiriUhlir
143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
"""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
|