furtka/furtka/installer.py

import shutil
from pathlib import Path

from furtka import sources
from furtka.manifest import APP_NAME_RE, 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 app_dir(name: str) -> Path:
    """/var/lib/furtka/apps/<name>, refusing anything that isn't a plain app name.

    Names reach us from API bodies and URL paths; without this check "..",
    "." or "/" would point remove()'s rmtree at /var/lib/furtka, every app,
    or the root filesystem.
    """
    if not isinstance(name, str) or not APP_NAME_RE.fullmatch(name):
        raise InstallError(f"{name!r} is not a valid app name")
    return apps_dir() / name


def _placeholder_values(values: dict[str, str]) -> list[str]:
    return [k for k, v in values.items() if v.strip() in PLACEHOLDER_SECRETS]


def _placeholder_keys(env_path: Path) -> list[str]:
    if not env_path.exists():
        return []
    return _placeholder_values(_read_env(env_path))


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, values: dict[str, str]) -> 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.
    """
    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 _validate_settings(m: Manifest, values: dict[str, str]) -> None:
    """Refuse a merged settings view before it is written to .env.

    Runs in memory so a rejected value (placeholder, system path, line
    break) never reaches the live .env where the next reconcile would use it.
    """
    for k, v in values.items():
        if any(c in v for c in "\r\n\0"):
            raise InstallError(f"{m.name}: setting {k!r} must not contain line breaks")
    bad = _placeholder_values({s.name: values.get(s.name, "") for s in m.settings})
    if bad:
        raise InstallError(
            f"{m.name}: please replace the example value for {', '.join(bad)} with your own"
        )
    path_errors = _path_setting_errors(m, values)
    if path_errors:
        raise InstallError(f"{m.name}: {'; '.join(path_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 any(c in v for c in "\r\n\0"):
        raise InstallError("setting values must not contain line breaks")
    if "$" in v and "'" not in v:
        # Compose expands ${VAR} inside double quotes, so "pa$word" would
        # silently turn into "pa". Single-quoted values are taken literally.
        return f"'{v}'"
    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 = app_dir(m.name)
    target.mkdir(parents=True, exist_ok=True)

    for item in src.iterdir():
        dest = target / item.name
        # Subdirectories (e.g. scripts/ holding a provider's on_install/on_start
        # dependency hooks) must come along too — copy the whole tree, replacing
        # any stale copy from a previous install so removed files don't linger.
        if item.is_dir():
            if dest.exists():
                shutil.rmtree(dest)
            shutil.copytree(item, dest)
            continue
        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, dest)

    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")
        _validate_settings(m, merged)
        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, read_env_values(env))
    if path_errors:
        raise InstallError(f"{m.name}: {'; '.join(path_errors)}")

    return target


def parse_env_text(text: str) -> dict[str, str]:
    """Parse KEY=VALUE lines from a string into a dict. Unquotes quoted values.

    Reusable by anything that needs the same lenient .env parsing logic
    without reading a file — e.g. hook script stdout merged into an app's
    .env during install (see install_runner._fire_install_hooks).
    """
    out: dict[str, str] = {}
    for raw in text.splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        value = value.strip()
        if len(value) >= 2 and value.startswith('"') and value.endswith('"'):
            value = value[1:-1].replace('\\"', '"').replace("\\\\", "\\")
        elif len(value) >= 2 and value.startswith("'") and value.endswith("'"):
            # Single quotes are literal in compose's dotenv — no escapes.
            value = value[1:-1]
        out[key.strip()] = value
    return out


def _read_env(env_path: Path) -> dict[str, str]:
    """Parse a simple KEY=VALUE .env into a dict. Unquotes quoted values."""
    return parse_env_text(env_path.read_text())


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 = app_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")
    _validate_settings(m, merged)
    write_env(env, merged)
    return target


def install_plan(plan, settings_target: dict[str, str] | None = None) -> list[Path]:
    """Run the synchronous install phase for every app in `plan.to_install`.

    Each name is resolved via `resolve_source()` and copied via `install_from`
    in plan order, so providers land before consumers. Only the target app
    receives user-supplied settings — transitive providers install from their
    catalog/bundled `.env.example` and rely on the placeholder-secret check
    to refuse if anyone shipped a "changeme" default.

    No rollback on partial failure. Re-running install is the recovery path;
    stopping providers a user may already rely on for other apps is more
    destructive than a partial state. Returns the list of target folders in
    install order.
    """
    targets: list[Path] = []
    for name in plan.to_install:
        src = resolve_source(name)
        settings = settings_target if name == plan.target else None
        targets.append(install_from(src, settings=settings))
    return targets


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 = app_dir(name)
    if not target.exists():
        raise InstallError(f"{name!r} is not installed")
    shutil.rmtree(target)
    return target