#!/usr/bin/env python3
"""Synthetic wizard/Keychain tests. Run with the openconnect-lite Python."""
import base64
import importlib.util
import json
import os
import fcntl
from pathlib import Path
import pty
import select
import subprocess
import sys
import tempfile
import termios
import time

sys.dont_write_bytecode = True

# The existing tests/logging.py is a test entry point, not stdlib logging.
sys.path = [p for p in sys.path if Path(p).resolve() != Path(__file__).resolve().parent]
from unittest.mock import patch

import pyotp

ROOT = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location("yandex_import", ROOT / "bin/lemana-vpn-import-yandex.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
SECRET = base64.b32encode(b"12345678901234567890").decode()
OTHER = "JBSWY3DPEHPK3PXP"
URI = f"otpauth://totp/Lemana:example?secret={SECRET}&algorithm=SHA1&digits=6&period=30"
ENTRY = {"name": "Lemana VPN", "secret": SECRET, "techInfo": URI}
ORIGINAL_CONFIG = 'LEMANA_VPN_USERNAME="example"\nLEMANA_VPN_MODE="bitwarden-auto"\nLEMANA_VPN_LOCAL_OTP="0"\nLEMANA_VPN_LOG_MAX_BYTES="12345"\n'
verify = pyotp.TOTP.verify
assert pyotp.TOTP(SECRET).at(59) == "287082"  # RFC 6238 SHA1 vector, six digits.


class FakeKeyring:
    def __init__(self, previous=OTHER):
        self.values = {"example": "unchanged-password"}
        if previous is not None:
            self.values["totp/example"] = previous
        self.writes = []

    def get_password(self, service, account):
        assert service == "openconnect-lite"
        return self.values.get(account)

    def set_password(self, service, account, value):
        assert service == "openconnect-lite" and account == "totp/example"
        self.writes.append((account, value))
        self.values[account] = value

    def delete_password(self, service, account):
        assert service == "openconnect-lite" and account == "totp/example"
        self.values.pop(account, None)


def wizard(directory, keyring, answers=None, entries=None, code="287082"):
    backup = directory / "key_backup"
    backup.write_text(json.dumps(entries if entries is not None else [
        {"name": "Yandex account", "techInfo": "otpauth://yaotp/test?secret=not-supported"}, ENTRY]))
    config = directory / "env"
    config.write_text(ORIGINAL_CONFIG)
    inputs = iter(answers if answers is not None else [str(backup), "2"])
    output = []

    def ask(prompt):
        output.append(prompt)
        try:
            return next(inputs)
        except StopIteration:
            raise EOFError from None

    def ask_secret(prompt):
        assert any("LDAP: example" in line for line in output), output
        assert any("Запись: Lemana VPN" in line for line in output), output
        output.append(prompt)
        if code is None:
            raise EOFError
        return code

    failure = None
    with patch.dict(sys.modules, {"keyring": keyring}), patch.object(pyotp.TOTP, "verify", lambda self, otp, valid_window: verify(self, otp, for_time=59, valid_window=valid_window)):
        try:
            module.configure("example", config, ask, ask_secret, output.append)
        except (module.ImportFailure, EOFError, KeyboardInterrupt, OSError) as error:
            failure = error
    captured = "\n".join(output)
    assert SECRET not in captured and OTHER not in captured and URI not in captured, captured
    assert "unchanged-password" not in captured
    return config, failure, output


with tempfile.TemporaryDirectory(prefix="lemana-yandex-") as temporary:
    directory = Path(temporary)
    keyring = FakeKeyring()
    config, error, output = wizard(directory, keyring)
    assert error is None, error
    assert keyring.writes == [("totp/example", SECRET)], keyring.writes
    assert keyring.values["example"] == "unchanged-password"
    assert 'LEMANA_VPN_MODE="keychain-auto"' in config.read_text()
    assert 'LEMANA_VPN_LOCAL_OTP="1"' in config.read_text()
    assert 'LEMANA_VPN_LOG_MAX_BYTES="12345"' in config.read_text()
    assert config.stat().st_mode & 0o777 == 0o600
    assert not list(directory.glob(".otp-env-*"))
    print("PASS wizard: mixed backup, explicit VPN selection, RFC code, one Keychain entry and private settings")

    for answers, code in (([], "287082"), ([str(directory / "key_backup")], "287082"),
                          ([str(directory / "key_backup"), ""], "287082"),
                          ([str(directory / "key_backup"), "0"], "287082"),
                          (None, "000000"), (None, None)):
        keyring = FakeKeyring()
        config, error, _ = wizard(directory, keyring, answers=answers, code=code)
        assert error is not None and keyring.writes == []
        assert config.read_text() == ORIGINAL_CONFIG
    print("PASS cancellation/invalid choice/wrong code: no writes")

    variants = [URI.replace("//totp/", "//hotp/"), URI.replace("//totp/", "//yaotp/"),
                URI.replace("SHA1", "SHA256"), URI.replace("digits=6", "digits=8"),
                URI.replace("period=30", "period=60"), URI + "&secret=" + OTHER,
                URI + "&algorithm=SHA1", URI + "&digits=6", URI + "&period=30",
                URI.replace(SECRET, "NOT-BASE32"), URI.replace(SECRET, OTHER),
                URI + "&counter=1", URI.replace("otpauth:", "https:")]
    variants.extend([URI + "&encoder=steam", URI + "&chars=alpha&length=8"])
    for uri in variants:
        keyring = FakeKeyring()
        config, error, _ = wizard(directory, keyring, entries=[dict(ENTRY, techInfo=uri)],
                                  answers=[str(directory / "key_backup"), "1"])
        assert isinstance(error, module.ImportFailure), (uri, error)
        assert keyring.writes == [] and config.read_text() == ORIGINAL_CONFIG
        assert SECRET not in str(error) and OTHER not in str(error)
    print("PASS unsupported types/parameters, duplicate parameters, malformed/mismatched secrets")

    for previous in (OTHER, None):
        for failure in (OSError("secret backend detail"), KeyboardInterrupt()):
            keyring = FakeKeyring(previous)
            with patch.object(module.os, "replace", side_effect=failure):
                config, error, _ = wizard(directory, keyring)
            assert error is failure and config.read_text() == ORIGINAL_CONFIG
            assert keyring.get_password("openconnect-lite", "totp/example") == previous
            assert not list(directory.glob(".otp-env-*"))
    keyring = FakeKeyring()
    with patch.object(keyring, "set_password", side_effect=RuntimeError("backend secret")), patch.dict(sys.modules, {"keyring": keyring}):
        try:
            module.activate(SECRET, "example", config)
            assert False, "expected backend failure"
        except module.ImportFailure as error:
            assert "вернуть прежний секрет не удалось" in str(error) and "backend secret" not in str(error)
    assert config.read_text() == ORIGINAL_CONFIG
    print("PASS config failure/interruption rollback and secret-safe rollback failure")

    keyring = FakeKeyring()
    real_replace = module.os.replace

    def interrupted_after_replace(source, destination):
        real_replace(source, destination)
        raise KeyboardInterrupt

    with patch.object(module.os, "replace", side_effect=interrupted_after_replace):
        config, error, _ = wizard(directory, keyring)
    assert isinstance(error, module.ImportFailure) and "уже сохранена" in str(error)
    assert keyring.values["totp/example"] == SECRET and 'LEMANA_VPN_LOCAL_OTP="1"' in config.read_text()
    keyring = FakeKeyring(None)
    with patch.object(keyring, "set_password", return_value=None):
        config, error, _ = wizard(directory, keyring)
    assert isinstance(error, module.ImportFailure) and config.read_text() == ORIGINAL_CONFIG
    keyring = FakeKeyring()
    config.write_text(ORIGINAL_CONFIG.replace('USERNAME="example"', 'USERNAME="another"'))
    with patch.dict(sys.modules, {"keyring": keyring}):
        try:
            module.activate(SECRET, "example", config)
            assert False, "wrong saved account accepted"
        except module.ImportFailure:
            pass
    assert not keyring.writes
    config.write_text(ORIGINAL_CONFIG)
    print("PASS commit-point consistency, silent backend failure and saved-account binding")

    backup = directory / "bad_backup"
    for data in (b"not JSON", b"{}", b"[]", b"[{\"techInfo\":1}]", b"x" * (module.MAX_BACKUP_BYTES + 1)):
        backup.write_bytes(data)
        try:
            module.read_backup(backup)
            assert False, "bad backup accepted"
        except module.ImportFailure:
            pass
    assert module.display_name("hello\x1b[2J\nworld") == "hello[2Jworld"
    assert module.display_name("A" * 110 + SECRET, {SECRET}) == ""
    keyring = FakeKeyring()
    config, error, output = wizard(directory, keyring, entries=[{"name": SECRET.lower(), "techInfo": URI}],
                                   answers=[str(directory / "key_backup"), ""])
    assert SECRET.lower() not in "\n".join(output) and not keyring.writes
    config, error, output = wizard(directory, keyring, entries=[{
        "name": SECRET, "secret": " " + SECRET + " ", "techInfo": URI.replace(SECRET, "%20" + SECRET + "%20")}],
        answers=[str(directory / "key_backup"), ""])
    assert SECRET not in "\n".join(output) and not keyring.writes
    print("PASS malformed/oversized files and terminal-control sanitization")

    # Installed entry point must fail without a controlling terminal, before writes.
    sso = directory / "xdg/openconnect-lite"
    sso.mkdir(parents=True)
    sso.joinpath("config.toml").write_text('[credentials]\nusername="example"\n')
    env = dict(os.environ, HOME=temporary, XDG_CONFIG_HOME=str(sso.parent), XDG_CONFIG_DIRS=str(sso.parent))
    proc = subprocess.run([sys.executable, str(ROOT / "bin/lemana-vpn-import-yandex.py"),
                           "--username=example", "--config", str(config)],
                          stdin=subprocess.DEVNULL, capture_output=True, text=True, start_new_session=True, env=env)
    assert proc.returncode == 1 and "Traceback" not in proc.stderr
    assert config.read_text() == ORIGINAL_CONFIG
    print("PASS no-terminal entry: bounded error, no writes")

    # Drive the actual CLI + helper through an automated PTY and a fake backend.
    fake = directory / "fake"
    fake.mkdir()
    fake.joinpath("security").write_text("#!/bin/sh\nexit 0\n")
    fake.joinpath("security").chmod(0o755)
    state = directory / "keyring.json"
    state.write_text(json.dumps({"example": "unchanged-password"}))
    fake.joinpath("keyring.py").write_text('''import json, os
from pathlib import Path
state = Path(os.environ["TEST_KEYRING"])
def get_password(service, account):
    assert service == "openconnect-lite"
    return json.loads(state.read_text()).get(account)
def set_password(service, account, password):
    assert service == "openconnect-lite"
    data = json.loads(state.read_text()); data[account] = password
    state.write_text(json.dumps(data))
def delete_password(service, account):
    data = json.loads(state.read_text()); data.pop(account, None)
    state.write_text(json.dumps(data))
''')
    directory.joinpath("key_backup").write_text(json.dumps([ENTRY]))
    env = {key: value for key, value in env.items() if not key.startswith(("LEMANA_VPN_", "OPENCONNECT_LITE_"))}
    env.update(PATH=str(fake) + ":" + os.environ["PATH"], PYTHONPATH=str(fake), TEST_KEYRING=str(state),
               LEMANA_VPN_CONFIG_DIR=temporary, LEMANA_VPN_OC_PYTHON=sys.executable,
               OPENCONNECT_LITE_CONFIG_DIR=str(sso), LEMANA_VPN_STATUS_DIR=str(directory / "status"))
    master, slave = pty.openpty()

    def controlling_terminal():
        os.setsid()
        fcntl.ioctl(slave, termios.TIOCSCTTY, 0)

    proc = subprocess.Popen(["/bin/bash", str(ROOT / "bin/vpn-lemanapro.sh"), "--import-yandex-key"],
                            stdin=slave, stdout=slave, stderr=slave, env=env, preexec_fn=controlling_terminal)
    os.close(slave)
    output = b""
    phone_code = pyotp.TOTP(SECRET).now()
    inputs = [("Путь к файлу", str(directory / "key_backup")), ("Номер записи", "1"),
              ("для проверки и сохранения:", phone_code)]
    deadline = time.monotonic() + 15
    try:
        while time.monotonic() < deadline:
            if select.select([master], [], [], max(0, deadline - time.monotonic()))[0]:
                try:
                    chunk = os.read(master, 8192)
                except OSError:
                    break  # PTY EOF on macOS.
                if not chunk:
                    break
                output += chunk
                if inputs and inputs[0][0].encode() in output:
                    os.write(master, (inputs.pop(0)[1] + "\n").encode())
            if proc.poll() is not None:
                break
        assert proc.wait(timeout=2) == 0, output.decode(errors="replace")
    finally:
        os.close(master)
        if proc.poll() is None:
            proc.kill()
            proc.wait()
    assert not inputs and SECRET.encode() not in output and phone_code.encode() not in output
    assert json.loads(state.read_text()) == {"example": "unchanged-password", "totp/example": SECRET}
    status = subprocess.run(["bash", str(ROOT / "bin/vpn-lemanapro.sh"), "--status", "--json"],
                            env=env, text=True, capture_output=True, check=True)
    modules = json.loads(status.stdout)["modules"]
    assert modules["local_otp"] and modules["mode"]["ready"] and not modules["bitwarden"]["enabled"]
    override = subprocess.run(["bash", str(ROOT / "bin/vpn-lemanapro.sh"), "--status", "--json"],
                              env=dict(env, LEMANA_VPN_MODE="manual-keychain", LEMANA_VPN_LOCAL_OTP="0"), text=True, capture_output=True, check=True)
    assert not json.loads(override.stdout)["modules"]["local_otp"]
    config_after = config.read_text()
    stored_after = state.read_text()
    runtime_definitions = (ROOT / "bin/vpn-lemanapro.sh").read_text().split('\nif $STATUS_MODE; then', 1)[0]
    mismatch = subprocess.run(["bash", "-c", runtime_definitions + '\n_can_prompt() { return 0; }; _import_yandex_key'],
                              env=dict(env, LEMANA_VPN_USERNAME="different-user"), text=True, capture_output=True)
    assert mismatch.returncode != 0 and "отличается" in mismatch.stderr
    mismatch = subprocess.run([sys.executable, str(ROOT / "bin/lemana-vpn-import-yandex.py"),
                               "--username=different-user", "--config", str(config)],
                              env=env, text=True, capture_output=True)
    assert mismatch.returncode != 0 and "профиле SSO" in mismatch.stderr
    assert config.read_text() == config_after and state.read_text() == stored_after
    print("PASS real CLI/PTY import → fake Keychain → persisted mode → status; hidden code; override and account guards")

print("yandex import checks ok")