furtka/furtka/installer.py

import shutil
from pathlib import Path

from furtka import sources
from furtka.manifest import Manifest, ManifestError, load_manifest
from furtka.paths import apps_dir

# Values that an app's .env.example may use as obvious "fill me in" markers.
# If any of these reach the live .env, install refuses — otherwise we'd ship
# an SMB share with password "changeme" out of the box, which is the kind of
# default that ends up screenshotted on Hacker News.
PLACEHOLDER_SECRETS: frozenset[str] = frozenset({"changeme"})

# System paths that must never be accepted as a user-supplied `path`-type
# setting. The user is root on their own box, so this is about preventing
# accidental footguns (typing `/etc` when they meant `/mnt/etc`), not
# defending against an attacker. Matches exact paths and their subtrees
# after `Path.resolve()` — so `/mnt/../etc` also lands here.
DENIED_PATH_PREFIXES: tuple[str, ...] = (
    "/etc",
    "/root",
    "/boot",
    "/proc",
    "/sys",
    "/dev",
    "/bin",
    "/sbin",
    "/usr/bin",
    "/usr/sbin",
    "/var/lib/furtka",
)


class InstallError(RuntimeError):
    pass


def _placeholder_keys(env_path: Path) -> list[str]:
    if not env_path.exists():
        return []
    bad: list[str] = []
    for raw in env_path.read_text().splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        value = value.strip().strip('"').strip("'")
        if value in PLACEHOLDER_SECRETS:
            bad.append(key.strip())
    return bad


def _is_denied_system_path(resolved: str) -> bool:
    if resolved == "/":
        return True
    for bad in DENIED_PATH_PREFIXES:
        if resolved == bad or resolved.startswith(bad + "/"):
            return True
    return False


def _path_setting_errors(m: Manifest, env_path: Path) -> list[str]:
    """Validate the filesystem paths named by `path`-type settings.

    Returns one human-readable message per offending setting. Empty values
    on non-required settings are allowed — the required-field check in the
    caller already refuses blanks on required fields before write.
    """
    if not env_path.exists():
        return []
    values = _read_env(env_path)
    errors: list[str] = []
    for s in m.settings:
        if s.type != "path":
            continue
        value = values.get(s.name, "")
        if not value:
            continue
        p = Path(value)
        if not p.is_absolute():
            errors.append(f"{s.name}={value!r} must be an absolute path (start with /)")
            continue
        try:
            resolved = p.resolve(strict=False)
        except (OSError, RuntimeError) as e:
            errors.append(f"{s.name}={value!r} cannot be resolved: {e}")
            continue
        if _is_denied_system_path(str(resolved)):
            errors.append(f"{s.name}={value!r} resolves into a system path and is not allowed")
            continue
        if not resolved.exists():
            errors.append(f"{s.name}={value!r} does not exist on this box")
            continue
        if not resolved.is_dir():
            errors.append(f"{s.name}={value!r} is not a directory")
            continue
    return errors


def _format_env_value(v: str) -> str:
    # Quote values that contain whitespace, quotes, or shell metacharacters so
    # docker-compose's env substitution reads them back intact. Simple values
    # stay unquoted to keep the file readable when a user SSHes in.
    if v == "" or any(c in v for c in " \t\"'$`\\#"):
        escaped = v.replace("\\", "\\\\").replace('"', '\\"')
        return f'"{escaped}"'
    return v


def write_env(env_path: Path, values: dict[str, str]) -> None:
    """Write a KEY=VALUE .env file atomically with 0600 perms.

    Preserves insertion order of `values` so the file reads in the same order
    the user filled in the form.
    """
    lines = [f"{k}={_format_env_value(v)}" for k, v in values.items()]
    body = "\n".join(lines) + ("\n" if lines else "")
    tmp = env_path.with_suffix(env_path.suffix + ".tmp")
    tmp.write_text(body)
    tmp.chmod(0o600)
    tmp.replace(env_path)


def resolve_source(source: str) -> Path:
    """Resolve a `furtka app install <source>` arg to a real source folder.

    If `source` looks like a path (or exists on disk), use it. Otherwise treat
    it as an app name and look it up via `furtka.sources.resolve_app_name` —
    which checks the synced catalog first and falls back to the bundled seed.
    """
    p = Path(source)
    if p.is_dir():
        return p
    if "/" in source or source.startswith("."):
        raise InstallError(f"{source!r} is not a directory")
    resolved = sources.resolve_app_name(source)
    if resolved is None:
        raise InstallError(f"{source!r} not found as a path, catalog app, or bundled app")
    return resolved.path


def install_from(src: Path, settings: dict[str, str] | None = None) -> Path:
    """Copy a validated app folder into /var/lib/furtka/apps/<name>/.

    If `settings` is provided, the .env is written from those values (this is
    what the Web UI / API does — user fills in a form, values land here).
    Otherwise, preserves an existing .env on upgrade and bootstraps from
    .env.example on first install.

    Refuses to finish (raises InstallError) if the resulting .env still has
    placeholder secrets — the target folder is left in place so the user can
    edit and re-run.

    Returns the target folder on success.
    """
    manifest_path = src / "manifest.json"
    if not manifest_path.exists():
        raise InstallError(f"{src} has no manifest.json")
    try:
        m = load_manifest(manifest_path)
    except ManifestError as e:
        raise InstallError(str(e)) from e

    if settings is not None:
        declared = {s.name for s in m.settings}
        for key in settings:
            if key not in declared:
                raise InstallError(f"{m.name}: unknown setting {key!r}")

    target = apps_dir() / m.name
    target.mkdir(parents=True, exist_ok=True)

    for item in src.iterdir():
        if not item.is_file():
            continue
        # Never overwrite an existing user .env — either settings-driven write
        # or previous manual edit has authority.
        if item.name == ".env" and (target / ".env").exists():
            continue
        shutil.copy2(item, target / item.name)

    env = target / ".env"
    env_example = target / ".env.example"

    if settings is not None:
        # Merge: start from existing .env (to preserve values the user didn't
        # change — e.g. when editing a single password), overlay the submitted
        # settings. Manifest-declared fields always appear in the final file.
        existing = _read_env(env) if env.exists() else {}
        merged: dict[str, str] = {}
        for s in m.settings:
            if s.name in settings:
                merged[s.name] = settings[s.name]
            elif s.name in existing:
                merged[s.name] = existing[s.name]
            elif s.default is not None:
                merged[s.name] = s.default
            else:
                merged[s.name] = ""
        # Preserve any non-manifest keys already in .env (forward-compat).
        for k, v in existing.items():
            if k not in merged:
                merged[k] = v
        # Required-field check runs on the merged view so that editing just
        # one field (e.g. password) doesn't trip on unsubmitted fields that
        # already have values in the existing .env.
        for s in m.settings:
            if s.required and not merged.get(s.name):
                raise InstallError(f"{m.name}: setting {s.name!r} is required")
        write_env(env, merged)
    elif not env.exists() and env_example.exists():
        # First install with no settings and no .env shipped: bootstrap from
        # .env.example so compose has values to substitute.
        shutil.copy2(env_example, env)

    # .env carries app secrets — lock to root-only. Done before the placeholder
    # check so even the half-installed state is at least not world-readable.
    if env.exists():
        env.chmod(0o600)

    bad = _placeholder_keys(env)
    if bad:
        raise InstallError(
            f"{m.name}: {env} still has placeholder values for {', '.join(bad)}. "
            f"Open the app in the Furtka UI to fill in real values, or edit the "
            f"file and re-run `furtka app install {m.name}`."
        )

    path_errors = _path_setting_errors(m, env)
    if path_errors:
        raise InstallError(f"{m.name}: {'; '.join(path_errors)}")

    return target


def _read_env(env_path: Path) -> dict[str, str]:
    """Parse a simple KEY=VALUE .env into a dict. Unquotes quoted values."""
    out: dict[str, str] = {}
    for raw in env_path.read_text().splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        value = value.strip()
        if (value.startswith('"') and value.endswith('"')) or (
            value.startswith("'") and value.endswith("'")
        ):
            value = value[1:-1].replace('\\"', '"').replace("\\\\", "\\")
        out[key.strip()] = value
    return out


def read_env_values(env_path: Path) -> dict[str, str]:
    """Public wrapper — returns {} if the file doesn't exist."""
    if not env_path.exists():
        return {}
    return _read_env(env_path)


def update_env(name: str, settings: dict[str, str]) -> Path:
    """Merge `settings` into the installed app's .env.

    Preserves values the user didn't submit. Validates required fields against
    the merged view. Leaves files/manifest untouched — for already-installed
    apps only. Returns the target folder; caller is expected to run
    reconcile to restart the containers.
    """
    target = apps_dir() / name
    manifest_path = target / "manifest.json"
    if not manifest_path.exists():
        raise InstallError(f"{name!r} is not installed")
    try:
        m = load_manifest(manifest_path)
    except ManifestError as e:
        raise InstallError(str(e)) from e

    declared = {s.name for s in m.settings}
    for key in settings:
        if key not in declared:
            raise InstallError(f"{m.name}: unknown setting {key!r}")

    env = target / ".env"
    existing = _read_env(env) if env.exists() else {}
    merged: dict[str, str] = {}
    for s in m.settings:
        if s.name in settings:
            merged[s.name] = settings[s.name]
        elif s.name in existing:
            merged[s.name] = existing[s.name]
        elif s.default is not None:
            merged[s.name] = s.default
        else:
            merged[s.name] = ""
    for k, v in existing.items():
        if k not in merged:
            merged[k] = v
    for s in m.settings:
        if s.required and not merged.get(s.name):
            raise InstallError(f"{m.name}: setting {s.name!r} is required")
    write_env(env, merged)
    bad = _placeholder_keys(env)
    if bad:
        raise InstallError(f"{m.name}: {env} still has placeholder values for {', '.join(bad)}.")
    path_errors = _path_setting_errors(m, env)
    if path_errors:
        raise InstallError(f"{m.name}: {'; '.join(path_errors)}")
    return target


def remove(name: str) -> Path:
    """Delete /var/lib/furtka/apps/<name>/. Volumes are NOT touched.

    Caller is responsible for stopping the compose project first.
    """
    target = apps_dir() / name
    if not target.exists():
        raise InstallError(f"{name!r} is not installed")
    shutil.rmtree(target)
    return target