"""
Tests fuer Phase 12:

  1) "Host-Key konnte nicht ermittelt werden -- Ziel nicht erreichbar:
     Permission denied for user ... on host ...": die Ermittlung lief ueber
     einen vollstaendigen asyncssh.connect()-Versuch und wartete darauf, dass
     SSHClient.validate_host_public_key() den Fingerprint liefert. Diesen
     Callback ruft asyncssh aber nur auf, wenn known_hosts NICHT None ist --
     mit known_hosts=None ist die Host-Key-Pruefung abgeschaltet. Der
     Fingerprint blieb leer, und der voellig erwartbare Auth-Fehler wurde als
     "Ziel nicht erreichbar" gemeldet. Jetzt: asyncssh.get_server_host_key(),
     also reiner Key-Exchange ohne Anmeldung.
  2) Aus derselben Ursache folgt: das Pinning im REGULAEREN Verbindungspfad
     war wirkungslos -- der Callback wurde dort ebenso nie aufgerufen. Die
     Pruefung findet jetzt vor der Anmeldung statt und wird nach dem
     Sitzungsaufbau noch einmal gegengeprueft.
  3) Der Benutzername gehoert zu den ZUGANGSDATEN, nicht zum Host
     (Migration 0010): ssh_keys.username bzw. rdp_credentials.username/.domain.

Die Tests kommen ohne laufenden SSH-Server aus: asyncssh wird an den beiden
Eintrittspunkten (get_server_host_key/connect) ersetzt, die Datenbank ist eine
echte SQLite-Datei, auf die die echten Migrationen angewendet werden.
"""
from __future__ import annotations

import pathlib
import sqlite3

import pytest

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ed25519

from app.rdp_proxy.guacd_client import GuacamoleProtocolError, build_rdp_params
from app.security.crypto import encrypt_secret
from app.ssh_proxy.proxy import (
    HostKeyDiscoveryError,
    HostKeyMismatchError,
    HostKeyNotPinnedError,
    HostNotConfiguredError,
    connect_to_host,
    discover_and_store_host_key,
    load_host,
    resolve_ssh_username,
)

MIGRATIONS = pathlib.Path(__file__).resolve().parent.parent / "app" / "db" / "migrations"

FINGERPRINT = "SHA256:0123456789abcdefghijklmnopqrstuvwxyzABCDEFG"
PUBLIC_KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAITESTKEYTESTKEYTESTKEYTESTKEYTESTKEY host"


# --------------------------------------------------------------------------
# Hilfsmittel: echte SQLite-DB mit den echten Migrationen + winziger
# aiosqlite-kompatibler Adapter (die Produktionsfunktionen erwarten await).
# --------------------------------------------------------------------------

class _Cursor:
    def __init__(self, cursor: sqlite3.Cursor) -> None:
        self._cursor = cursor
        self.lastrowid = cursor.lastrowid

    async def fetchone(self):
        return self._cursor.fetchone()

    async def fetchall(self):
        return self._cursor.fetchall()


class FakeConnection:
    """Genau so viel aiosqlite, wie app/ssh_proxy/proxy.py benutzt."""

    def __init__(self, path: str) -> None:
        self.raw = sqlite3.connect(path)

    async def execute(self, sql, params=()):
        return _Cursor(self.raw.execute(sql, params))

    async def commit(self):
        self.raw.commit()


def _apply_migrations(db: sqlite3.Connection, upto: str | None = None) -> None:
    for path in sorted(MIGRATIONS.glob("*.sql")):
        db.executescript(path.read_text(encoding="utf-8"))
        if upto and path.name.startswith(upto):
            break
    db.commit()


def _plain_key_pem() -> str:
    return ed25519.Ed25519PrivateKey.generate().private_bytes(
        serialization.Encoding.PEM,
        serialization.PrivateFormat.OpenSSH,
        serialization.NoEncryption(),
    ).decode()


def _make_db(tmp_path, *, host_username=None, key_username="l4u", fingerprint=FINGERPRINT,
             host_key=PUBLIC_KEY, with_key=True) -> str:
    path = str(tmp_path / "jumphost.sqlite3")
    db = sqlite3.connect(path)
    _apply_migrations(db)
    db.execute("INSERT INTO host_groups (id, name, tenant_id) VALUES (1, 'gruppe', 1)")
    db.execute(
        "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, "
        "ssh_host_key_fingerprint, ssh_host_key, ssh_username) "
        "VALUES (1, 1, 'ziel', '185.228.139.3', 'ssh', 22, 'linux', ?, ?, ?)",
        (fingerprint, host_key, host_username),
    )
    if with_key:
        db.execute(
            "INSERT INTO ssh_keys (id, label, private_key_enc, public_key, key_type, tenant_id, username) "
            "VALUES (1, 'testkey', ?, 'ssh-ed25519 AAAA', 'ed25519', 1, ?)",
            (encrypt_secret(_plain_key_pem().encode(), associated_data=b"ssh_private_key"), key_username),
        )
        db.execute("INSERT INTO host_ssh_key_map (host_id, ssh_key_id) VALUES (1, 1)")
    db.commit()
    db.close()
    return path


class FakeHostKey:
    def __init__(self, fingerprint=FINGERPRINT, public=PUBLIC_KEY) -> None:
        self._fingerprint = fingerprint
        self._public = public

    def get_fingerprint(self, kind="sha256"):
        return self._fingerprint

    def export_public_key(self, fmt="openssh"):
        return self._public.encode()


class FakeConnectionResult:
    """Minimale asyncssh-Verbindung: liefert den Host-Key der Sitzung."""

    def __init__(self, host_key=None) -> None:
        self._host_key = host_key or FakeHostKey()
        self.aborted = False

    def get_server_host_key(self):
        return self._host_key

    def abort(self):
        self.aborted = True

    def close(self):
        pass


# --------------------------------------------------------------------------
# 1) Host-Key-Ermittlung
# --------------------------------------------------------------------------

async def test_discovery_braucht_keine_anmeldung(tmp_path, monkeypatch):
    """Reproduktion des gemeldeten Fehlers: die Anmeldung scheitert (der Host
    kennt den Benutzer nicht), der Host-Key ist trotzdem ermittelbar."""
    import asyncssh

    async def _fake_connect(*a, **kw):  # pragma: no cover - darf nicht laufen
        raise AssertionError("Die Ermittlung darf keine Anmeldung mehr versuchen")

    async def _fake_get_host_key(host, port=22):  # bewusst OHNE **kw -- siehe test_phase14.py
        assert (host, port) == ("185.228.139.3", 22)
        return FakeHostKey()

    monkeypatch.setattr(asyncssh, "connect", _fake_connect)
    monkeypatch.setattr(asyncssh, "get_server_host_key", _fake_get_host_key, raising=False)

    conn = FakeConnection(_make_db(tmp_path, fingerprint=None, host_key=None))
    fingerprint = await discover_and_store_host_key(conn, 1, admin_user_id=1)

    assert fingerprint == FINGERPRINT
    row = conn.raw.execute(
        "SELECT ssh_host_key_fingerprint, ssh_host_key FROM hosts WHERE id = 1"
    ).fetchone()
    # Nicht nur der Fingerprint, auch der vollstaendige Schluessel wird
    # gespeichert -- nur damit laesst sich spaeter VOR der Anmeldung pinnen.
    assert row == (FINGERPRINT, PUBLIC_KEY)


async def test_discovery_meldet_echte_unerreichbarkeit(tmp_path, monkeypatch):
    import asyncssh

    async def _fake_get_host_key(host, port=22):  # bewusst OHNE **kw -- siehe test_phase14.py
        raise OSError("Connection refused")

    monkeypatch.setattr(asyncssh, "get_server_host_key", _fake_get_host_key, raising=False)
    conn = FakeConnection(_make_db(tmp_path))
    with pytest.raises(HostKeyDiscoveryError) as excinfo:
        await discover_and_store_host_key(conn, 1, admin_user_id=1)
    assert "Connection refused" in excinfo.value.reason


# --------------------------------------------------------------------------
# 2) Pinning
# --------------------------------------------------------------------------

async def test_verbindung_ohne_hinterlegten_hostkey_wird_abgelehnt(tmp_path, monkeypatch):
    import asyncssh

    async def _fake_connect(*a, **kw):  # pragma: no cover - darf nicht laufen
        raise AssertionError("Ohne gepinnten Host-Key darf nicht verbunden werden")

    monkeypatch.setattr(asyncssh, "connect", _fake_connect)
    conn = FakeConnection(_make_db(tmp_path, fingerprint=None, host_key=None))
    with pytest.raises(HostKeyNotPinnedError):
        await connect_to_host(conn, 1)


async def test_abweichender_hostkey_bricht_vor_der_anmeldung_ab(tmp_path, monkeypatch):
    """Der entscheidende Punkt: kein einziges Byte Zugangsdaten geht raus."""
    import asyncssh

    versuche = []

    async def _fake_connect(*a, **kw):  # pragma: no cover - darf nicht laufen
        versuche.append(kw)
        raise AssertionError("Bei abweichendem Host-Key darf keine Anmeldung erfolgen")

    async def _fake_get_host_key(host, port=22):  # bewusst OHNE **kw -- siehe test_phase14.py
        return FakeHostKey(fingerprint="SHA256:einVoelligAndererSchluessel", public="ssh-ed25519 AAAAfremd x")

    monkeypatch.setattr(asyncssh, "connect", _fake_connect)
    monkeypatch.setattr(asyncssh, "get_server_host_key", _fake_get_host_key, raising=False)

    conn = FakeConnection(_make_db(tmp_path))
    with pytest.raises(HostKeyMismatchError) as excinfo:
        await connect_to_host(conn, 1)
    assert excinfo.value.expected == FINGERPRINT
    assert versuche == []


async def test_verbindung_nutzt_benutzernamen_der_zugangsdaten(tmp_path, monkeypatch):
    import asyncssh

    aufrufe = {}

    async def _fake_connect(address, **kw):
        aufrufe["address"] = address
        aufrufe.update(kw)
        return FakeConnectionResult()

    async def _fake_get_host_key(host, port=22):  # bewusst OHNE **kw -- siehe test_phase14.py
        return FakeHostKey()

    monkeypatch.setattr(asyncssh, "connect", _fake_connect)
    monkeypatch.setattr(asyncssh, "get_server_host_key", _fake_get_host_key, raising=False)

    conn = FakeConnection(_make_db(tmp_path, host_username="alterWertAmHost", key_username="l4u"))
    await connect_to_host(conn, 1)
    # Der Name aus den Zugangsdaten gewinnt gegen den Altwert am Host.
    assert aufrufe["username"] == "l4u"


async def test_hostkey_wechsel_nach_der_pruefung_beendet_die_sitzung(tmp_path, monkeypatch):
    import asyncssh

    verbindung = FakeConnectionResult(FakeHostKey(fingerprint="SHA256:plotzlichAnders"))

    async def _fake_connect(address, **kw):
        return verbindung

    async def _fake_get_host_key(host, port=22):  # bewusst OHNE **kw -- siehe test_phase14.py
        return FakeHostKey()

    monkeypatch.setattr(asyncssh, "connect", _fake_connect)
    monkeypatch.setattr(asyncssh, "get_server_host_key", _fake_get_host_key, raising=False)

    conn = FakeConnection(_make_db(tmp_path))
    with pytest.raises(HostKeyMismatchError):
        await connect_to_host(conn, 1)
    assert verbindung.aborted is True


async def test_altbestand_ohne_gespeicherten_hostkey_wird_nachgetragen(tmp_path, monkeypatch):
    """Hosts, deren Key vor Migration 0010 ermittelt wurde, kennen nur den
    Fingerprint. Der Schluessel wird beim naechsten Verbindungsaufbau
    nachgetragen -- nachdem er gegen den Fingerprint geprueft wurde."""
    import asyncssh

    async def _fake_connect(address, **kw):
        return FakeConnectionResult()

    async def _fake_get_host_key(host, port=22):  # bewusst OHNE **kw -- siehe test_phase14.py
        return FakeHostKey()

    monkeypatch.setattr(asyncssh, "connect", _fake_connect)
    monkeypatch.setattr(asyncssh, "get_server_host_key", _fake_get_host_key, raising=False)

    conn = FakeConnection(_make_db(tmp_path, host_key=None))
    await connect_to_host(conn, 1)
    (stored,) = conn.raw.execute("SELECT ssh_host_key FROM hosts WHERE id = 1").fetchone()
    assert stored == PUBLIC_KEY


# --------------------------------------------------------------------------
# 3) Benutzername gehoert zu den Zugangsdaten
# --------------------------------------------------------------------------

def test_resolve_ssh_username_vorrang_und_fallback():
    assert resolve_ssh_username({"ssh_username": "alt"}, "neu") == "neu"
    # Altbestand: solange am Schluessel nichts steht, greift der Host-Wert.
    assert resolve_ssh_username({"ssh_username": "alt"}, None) == "alt"
    assert resolve_ssh_username({"ssh_username": "alt"}, "  ") == "alt"


def test_resolve_ssh_username_ohne_jeden_wert_meldet_klartext():
    with pytest.raises(HostNotConfiguredError) as excinfo:
        resolve_ssh_username({"ssh_username": None}, None)
    text = str(excinfo.value)
    assert "Benutzername" in text and "SSH-Key" in text


def _rdp_host(**overrides):
    host = {
        "id": 7, "address": "10.0.0.5", "port": 3389, "rdp_require_nla": True,
        "clipboard_enabled": True, "rdp_ignore_cert": True, "file_transfer_enabled": True,
    }
    host.update(overrides)
    return host


def test_rdp_params_nehmen_benutzernamen_der_zugangsdaten():
    params = build_rdp_params(
        _rdp_host(rdp_username="alt", rdp_domain="ALTEDOMAENE"),
        "geheim", username="Administrator", domain="CONTOSO",
    )
    assert params["username"] == "Administrator"
    assert params["domain"] == "CONTOSO"
    assert params["password"] == "geheim"


def test_rdp_params_fallback_auf_altwert_am_host():
    params = build_rdp_params(_rdp_host(rdp_username="alt", rdp_domain="D"), "geheim")
    assert (params["username"], params["domain"]) == ("alt", "D")


def test_rdp_params_ohne_benutzernamen_meldet_zugangsdaten():
    with pytest.raises(GuacamoleProtocolError) as excinfo:
        build_rdp_params(_rdp_host(), "geheim")
    assert "Zugangsdaten" in str(excinfo.value)


def test_rdp_params_brauchen_die_hostspalte_nicht_mehr():
    """Ein Hostdatensatz ohne rdp_username ist kein Fehler mehr -- der Name
    kommt jetzt von woanders."""
    params = build_rdp_params(_rdp_host(), "geheim", username="svc")
    assert params["username"] == "svc"


# --------------------------------------------------------------------------
# 4) Migration 0010: Uebernahme der Altwerte
# --------------------------------------------------------------------------

def _pre_0010_db(tmp_path) -> sqlite3.Connection:
    db = sqlite3.connect(str(tmp_path / "alt.sqlite3"))
    _apply_migrations(db, upto="0009")
    return db


def test_migration_uebernimmt_eindeutige_ssh_benutzernamen(tmp_path):
    db = _pre_0010_db(tmp_path)
    db.execute("INSERT INTO host_groups (id, name, tenant_id) VALUES (1, 'g', 1)")
    for host_id in (1, 2):
        db.execute(
            "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, ssh_username) "
            "VALUES (?, 1, 'h', '10.0.0.1', 'ssh', 22, 'linux', 'l4u')",
            (host_id,),
        )
    db.execute(
        "INSERT INTO ssh_keys (id, label, private_key_enc, public_key, key_type, tenant_id) "
        "VALUES (1, 'k', X'00', 'pub', 'ed25519', 1)"
    )
    db.execute("INSERT INTO host_ssh_key_map (host_id, ssh_key_id) VALUES (1, 1), (2, 1)")
    db.commit()

    db.executescript((MIGRATIONS / "0010_credential_usernames.sql").read_text(encoding="utf-8"))
    (username,) = db.execute("SELECT username FROM ssh_keys WHERE id = 1").fetchone()
    assert username == "l4u"


def test_migration_raet_nicht_bei_mehrdeutigen_benutzernamen(tmp_path):
    """Zwei Hosts, zwei verschiedene Benutzernamen, ein Schluessel: hier waere
    jede automatische Wahl geraten -- also bleibt das Feld leer und der
    Fallback greift weiter."""
    db = _pre_0010_db(tmp_path)
    db.execute("INSERT INTO host_groups (id, name, tenant_id) VALUES (1, 'g', 1)")
    for host_id, name in ((1, "root"), (2, "l4u")):
        db.execute(
            "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, ssh_username) "
            "VALUES (?, 1, 'h', '10.0.0.1', 'ssh', 22, 'linux', ?)",
            (host_id, name),
        )
    db.execute(
        "INSERT INTO ssh_keys (id, label, private_key_enc, public_key, key_type, tenant_id) "
        "VALUES (1, 'k', X'00', 'pub', 'ed25519', 1)"
    )
    db.execute("INSERT INTO host_ssh_key_map (host_id, ssh_key_id) VALUES (1, 1), (2, 1)")
    db.commit()

    db.executescript((MIGRATIONS / "0010_credential_usernames.sql").read_text(encoding="utf-8"))
    (username,) = db.execute("SELECT username FROM ssh_keys WHERE id = 1").fetchone()
    assert username is None


def test_migration_uebernimmt_rdp_benutzer_und_domaene(tmp_path):
    db = _pre_0010_db(tmp_path)
    db.execute("INSERT INTO host_groups (id, name, tenant_id) VALUES (1, 'g', 1)")
    db.execute(
        "INSERT INTO hosts (id, host_group_id, hostname, address, protocol, port, os_type, "
        "rdp_username, rdp_domain) "
        "VALUES (1, 1, 'win', '10.0.0.9', 'rdp', 3389, 'windows', 'Administrator', 'CONTOSO')"
    )
    db.execute("INSERT INTO rdp_credentials (host_id, password_enc) VALUES (1, X'00')")
    db.commit()

    db.executescript((MIGRATIONS / "0010_credential_usernames.sql").read_text(encoding="utf-8"))
    row = db.execute("SELECT username, domain FROM rdp_credentials WHERE host_id = 1").fetchone()
    assert row == ("Administrator", "CONTOSO")


async def test_load_host_liefert_den_hostkey_mit(tmp_path):
    conn = FakeConnection(_make_db(tmp_path))
    host = await load_host(conn, 1)
    assert host["ssh_host_key"] == PUBLIC_KEY