UmbraTrans/umbratrans_web/tests/test_cli.py

"""Operator CLI tests: production refusal, TTY requirement, hidden prompt, output."""

from __future__ import annotations

import io
import json
import os
import stat
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import TextIO

import pytest

from umbratrans_web import cli
from umbratrans_web.cli import (
    EXIT_ERROR,
    EXIT_NOT_INTERACTIVE,
    EXIT_OK,
    EXIT_PASSWORD_REJECTED,
    EXIT_REFUSED_PRODUCTION,
    bootstrap_admin,
)
from umbratrans_web.config import Environment, Settings
from umbratrans_web.dev_repository import JsonDevelopmentRepository
from umbratrans_web.service import AccountService

from .conftest import random_email, random_password


class _FakeTty(io.StringIO):
    """A stream that claims to be a terminal, for the hidden-prompt guard."""

    def isatty(self) -> bool:
        return True


class _Sink:
    """Minimal writable stream so CLI output can be asserted on."""

    def __init__(self) -> None:
        self.chunks: list[str] = []

    def write(self, value: str) -> int:
        self.chunks.append(value)
        return len(value)

    def flush(self) -> None:
        return None

    @property
    def text(self) -> str:
        return "".join(self.chunks)


@contextmanager
def _environment(values: dict[str, str] | None) -> Iterator[None]:
    previous = {key: os.environ.get(key) for key in (values or {})}
    try:
        for key, value in (values or {}).items():
            os.environ[key] = value
        yield
    finally:
        for key, value in previous.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value


def _run(
    argv: list[str],
    *,
    entries: list[str] | None = None,
    stdin: TextIO | None = None,
    env: dict[str, str] | None = None,
) -> tuple[int, str, str]:
    """Run the CLI with an injected prompt. Values are generated per run."""

    # users.email is NOT NULL in the control plane, so --email is required. A
    # generated address is supplied unless a test is exercising the flag.
    if "--email" not in argv:
        argv = [*argv, "--email", random_email("ops-admin")]

    out = _Sink()
    err = _Sink()
    # The command always prompts twice; a single value is reused for both.
    typed = list(entries) if entries else [random_password(28)]
    if len(typed) == 1:
        typed.append(typed[0])

    def fake_prompt(_label: str) -> str:
        return typed.pop(0)

    with _environment(env):
        code = bootstrap_admin(
            argv,
            stdin=stdin if stdin is not None else _FakeTty(),
            stdout=out,  # type: ignore[arg-type]
            stderr=err,  # type: ignore[arg-type]
            prompt=fake_prompt,
        )
    return code, out.text, err.text


def test_bootstrap_creates_a_must_change_account_in_a_private_file(tmp_path: Path) -> None:
    path = tmp_path / "state" / "dev-accounts.json"
    password = random_password(28)

    code, out, err = _run(["--repository-path", str(path)], entries=[password, password])

    assert code == EXIT_OK, err
    assert "Bootstrap account created." in out
    assert "Umbraadmin" in out
    assert "must_change_password: true" in out
    assert "platform_admin" in out
    # Nothing sensitive is printed.
    assert password not in out
    assert password not in err
    assert "$argon2" not in out

    assert stat.S_IMODE(path.stat().st_mode) == 0o600
    assert stat.S_IMODE(path.parent.stat().st_mode) == 0o700

    settings = Settings(
        environment=Environment.TEST,
        repository_kind="json",
        repository_path=path,
        argon2_memory_cost_kib=8 * 1024,
        argon2_time_cost=1,
        argon2_parallelism=1,
    )
    service = AccountService(JsonDevelopmentRepository(settings), settings)
    result = service.authenticate(username="Umbraadmin", password=password)
    assert result.session.must_change_password is True

    raw = path.read_text(encoding="utf-8")
    assert password not in raw
    payload = json.loads(raw)
    assert len(payload["credentials"]) == 1
    assert payload["credentials"][0]["credential_type"] == "local_password"
    assert payload["credentials"][0]["must_change_password"] is True
    assert payload["credentials"][0]["password_hash"].startswith("$argon2id$")


def test_bootstrap_uses_a_configurable_non_secret_username(tmp_path: Path) -> None:
    path = tmp_path / "dev-accounts.json"
    code, out, err = _run(["--repository-path", str(path), "--username", "Ops.Admin", "--role", "organization_admin"])
    assert code == EXIT_OK, err
    assert "Ops.Admin" in out
    assert "organization_admin" in out
    assert path.is_file()


def test_bootstrap_refuses_production_mode(tmp_path: Path) -> None:
    code, out, err = _run(
        ["--repository-path", str(tmp_path / "dev-accounts.json")],
        env={"UMBRA_ENV": "production"},
    )
    assert code == EXIT_REFUSED_PRODUCTION
    assert "development-only" in err
    assert "Refusing to run" in err
    assert not (tmp_path / "dev-accounts.json").exists()


def test_bootstrap_refuses_a_non_interactive_terminal(tmp_path: Path) -> None:
    path = tmp_path / "dev-accounts.json"
    code, out, err = _run(["--repository-path", str(path)], stdin=io.StringIO())
    assert code == EXIT_NOT_INTERACTIVE
    assert "interactive terminal" in err
    assert not path.exists()


def test_bootstrap_refuses_mismatched_entries(tmp_path: Path) -> None:
    path = tmp_path / "dev-accounts.json"
    code, out, err = _run(["--repository-path", str(path)], entries=[random_password(28), random_password(28)])
    assert code == EXIT_PASSWORD_REJECTED
    assert "did not match" in err
    assert not path.exists()


def test_bootstrap_rejects_a_password_that_fails_the_policy(tmp_path: Path) -> None:
    path = tmp_path / "dev-accounts.json"
    code, out, err = _run(["--repository-path", str(path)], entries=["short", "short"])
    assert code == EXIT_PASSWORD_REJECTED
    assert "policy" in err
    assert "short" not in err
    assert not path.exists()


def test_bootstrap_reports_a_duplicate_username_without_disclosing_secrets(tmp_path: Path) -> None:
    path = tmp_path / "dev-accounts.json"
    first = random_password(28)
    assert _run(["--repository-path", str(path)], entries=[first, first])[0] == EXIT_OK
    second = random_password(28)
    code, out, err = _run(["--repository-path", str(path)], entries=[second, second])
    assert code == EXIT_ERROR
    assert first not in err
    assert second not in err
    assert "$argon2" not in err


def test_serve_refuses_non_loopback_and_production(monkeypatch: pytest.MonkeyPatch) -> None:
    err = _Sink()
    monkeypatch.delenv("UMBRA_ENV", raising=False)
    assert cli.serve(["--host", "0.0.0.0"], stderr=err) == EXIT_REFUSED_PRODUCTION  # type: ignore[arg-type]
    assert "loopback" in err.text

    err = _Sink()
    monkeypatch.setenv("UMBRA_ENV", "production")
    assert cli.serve([], stderr=err) == EXIT_REFUSED_PRODUCTION  # type: ignore[arg-type]
    assert "production" in err.text


def test_module_entry_point_rejects_unknown_commands() -> None:
    from umbratrans_web.__main__ import main

    assert main(["not-a-command"]) == 2


def test_parser_exposes_no_password_option() -> None:
    parser = cli.build_parser()
    text = parser.format_help()
    assert "hidden" in text
    assert "password" in text.lower()
    option_strings = {
        option
        for action in parser._actions
        for option in action.option_strings  # noqa: SLF001
    }
    assert not any("password" in option for option in option_strings)
    assert not any("secret" in option for option in option_strings)


def test_interactive_guard_rejects_a_closed_stream() -> None:
    from umbratrans_web.errors import OperatorRefusalError

    closed = io.StringIO()
    closed.close()
    with pytest.raises(OperatorRefusalError) as error:
        cli._require_interactive(closed)  # noqa: SLF001
    assert error.value.code == "terminal_unavailable"


def test_prompt_defaults_to_getpass(monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[str] = []

    def fake_getpass(prompt: str) -> str:
        calls.append(prompt)
        return random_password(28)

    monkeypatch.setattr(cli.getpass, "getpass", fake_getpass)
    prompt: Callable[[str], str] = cli._prompt_hidden_password  # noqa: SLF001
    value = prompt("One-time password (input hidden): ")

    assert value
    assert calls == ["One-time password (input hidden): "]
    assert cli.getpass.getpass is fake_getpass