appfactory-portal/app/environment.py

import os
import re
import subprocess
from pathlib import Path

from app.db.apps import get_app_variables, get_apps

APP_DATA_ROOT = Path("/opt/appfactory/data/apps")
ENV_HEADER = "# GENERATED BY APPFACTORY - DO NOT EDIT"
ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


class AppEnvironmentError(RuntimeError):
    pass


def apply_app_environment(app_id: str) -> dict:
    app_dir = _app_runtime_dir(app_id)
    env_path = app_dir / ".env"
    variables = get_app_variables(app_id)
    new_content = _render_env_content(variables)
    old_content = _read_text(env_path)

    if old_content == new_content:
        return {"changed": False, "restarted": False, "container": None}

    _atomic_write(env_path, new_content)
    container = _find_running_container(app_id)
    if not container:
        return {"changed": True, "restarted": False, "container": None}

    _restart_container(container)
    return {"changed": True, "restarted": True, "container": container}


def apply_all_app_environments() -> dict:
    results = {}
    errors = {}
    for app in get_apps():
        app_id = app.get("id") or ""
        try:
            results[app_id] = apply_app_environment(app_id)
        except AppEnvironmentError as exc:
            errors[app_id] = str(exc)

    if errors:
        failed = ", ".join(sorted(errors))
        raise AppEnvironmentError(f"Environment apply failed for apps: {failed}")

    return results


def _app_runtime_dir(app_id: str) -> Path:
    if not app_id or "/" in app_id or "\\" in app_id or app_id in {".", ".."}:
        raise AppEnvironmentError("Invalid app id")

    root = APP_DATA_ROOT.resolve()
    app_dir = (root / app_id).resolve()
    if root not in app_dir.parents and app_dir != root:
        raise AppEnvironmentError("Invalid app runtime path")

    return app_dir


def _render_env_content(variables: list[dict]) -> str:
    lines = [ENV_HEADER]
    seen_keys: set[str] = set()

    for variable in variables:
        key = (variable.get("key") or "").strip()
        validate_environment_key(key)
        if key in seen_keys:
            raise AppEnvironmentError(f"Duplicate environment variable key: {key}")

        value = variable.get("value")
        if value is None:
            raise AppEnvironmentError(f"Missing value for environment variable key: {key}")

        seen_keys.add(key)
        lines.append(f"{key}={_format_env_value(str(value))}")

    return "\n".join(lines) + "\n"


def validate_environment_key(key: str) -> None:
    if not ENV_KEY_RE.match(key):
        raise AppEnvironmentError(f"Invalid environment variable key: {key or '<empty>'}")


def _format_env_value(value: str) -> str:
    if value == "":
        return ""

    needs_quotes = (
        value != value.strip()
        or any(char in value for char in (' ', '\t', '\r', '\n', '"', "'", "\\", "#", "$"))
    )
    if not needs_quotes:
        return value

    escaped = (
        value
        .replace("\\", "\\\\")
        .replace("\r", "\\r")
        .replace("\n", "\\n")
        .replace("\t", "\\t")
        .replace('"', '\\"')
        .replace("$", "\\$")
    )
    return f'"{escaped}"'


def _read_text(path: Path) -> str | None:
    try:
        return path.read_text(encoding="utf-8")
    except FileNotFoundError:
        return None
    except OSError as exc:
        raise AppEnvironmentError(f"Could not read runtime environment file: {exc}") from exc


def _atomic_write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = path.with_name(f".{path.name}.tmp.{os.getpid()}")
    fd = None
    try:
        fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as tmp_file:
            fd = None
            tmp_file.write(content)
            tmp_file.flush()
            os.fsync(tmp_file.fileno())
        os.chmod(tmp_path, 0o600)
        os.replace(tmp_path, path)
        _fsync_directory(path.parent)
    except OSError as exc:
        raise AppEnvironmentError(f"Could not write runtime environment file: {exc}") from exc
    finally:
        if fd is not None:
            os.close(fd)
        try:
            tmp_path.unlink()
        except FileNotFoundError:
            pass


def _fsync_directory(path: Path) -> None:
    if os.name == "nt":
        return

    fd = os.open(path, os.O_RDONLY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def _find_running_container(app_id: str) -> str | None:
    for container in _container_candidates(app_id):
        result = _run_docker(["inspect", "-f", "{{.State.Running}}", container])
        if result.returncode == 0 and result.stdout.strip().lower() == "true":
            return container

    return None


def _container_candidates(app_id: str) -> list[str]:
    normalized = app_id.replace("_", "-")
    candidates = [app_id, normalized, f"appfactory-{app_id}", f"appfactory-{normalized}"]
    unique = []
    for candidate in candidates:
        if candidate not in unique:
            unique.append(candidate)
    return unique


def _restart_container(container: str) -> None:
    result = _run_docker(["restart", container])
    if result.returncode != 0:
        raise AppEnvironmentError(f"Could not restart container {container}: {_safe_command_error(result.stderr)}")


def _run_docker(args: list[str]) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(["docker", *args], capture_output=True, text=True, timeout=30)
    except FileNotFoundError as exc:
        raise AppEnvironmentError("Docker command is not available") from exc
    except subprocess.TimeoutExpired as exc:
        raise AppEnvironmentError("Docker command timed out") from exc


def _safe_command_error(value: str) -> str:
    value = (value or "").strip()
    if not value:
        return "no error output"
    return value[:500]