"""End-to-end SQLI traffic over a real TLS socket.

``tests/test_tls.py`` covers the handshake. This covers what happens
*after* it: every byte of real SQLI traffic crossing genuine TLS records,
through the same codecs, reader, and cursor machinery the plain-socket
suite exercises.

That distinction matters because ``SSLSocket.recv`` is not
``socket.recv``. It returns at most one TLS record's worth of plaintext
however much you ask for, it can return fewer bytes than are available,
and plaintext buffered inside the SSL object is invisible to the OS. The
Phase 39 buffered reader asks for up to 64 KB per call and loops until
satisfied — that loop is the thing which has to be right, and nothing in
the plain-socket suite puts the same pressure on it.

**Scope.** A TLS-terminating proxy in front of the plain SQLI listener
supplies the TLS half. This tests the driver's TLS path, which is the
half we own. It does *not* test IBM's server-side TLS listener: Informix
15 wants a PKCS#12 keystore whose stash the developer-edition image
rejects (``GSK_ERROR_BAD_KEYFILE_PASSWORD``), and that side is IBM's
code. Anything below ``ssl.wrap_socket`` is identical either way.

Skipped when ``openssl`` isn't on PATH — the proxy needs a certificate.
"""

from __future__ import annotations

import contextlib
import datetime
import decimal
import select
import shutil
import socket
import ssl
import subprocess
import tempfile
import threading
from pathlib import Path

import pytest

import informix_db
from tests.conftest import ConnParams

pytestmark = pytest.mark.integration


# ---------------------------------------------------------------------------
# TLS-terminating proxy
# ---------------------------------------------------------------------------


class _TlsProxy:
    """Accepts TLS, relays plaintext to the real Informix listener."""

    def __init__(self, backend: tuple[str, int]) -> None:
        self.backend = backend
        self.tmpdir = tempfile.mkdtemp(prefix="ifx-tls-test-")
        self.cert = str(Path(self.tmpdir) / "cert.pem")
        key = str(Path(self.tmpdir) / "key.pem")
        subprocess.run(
            ["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
             "-keyout", key, "-out", self.cert, "-days", "1",
             "-subj", "/CN=127.0.0.1"],
            check=True, capture_output=True,
        )
        self._ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self._ctx.load_cert_chain(self.cert, key)
        self._sock = socket.socket()
        self._sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self._sock.bind(("127.0.0.1", 0))
        self._sock.listen(64)
        self.port: int = self._sock.getsockname()[1]
        self._stop = threading.Event()
        threading.Thread(target=self._serve, daemon=True).start()

    def _serve(self) -> None:
        self._sock.settimeout(0.5)
        while not self._stop.is_set():
            try:
                raw, _ = self._sock.accept()
            except (TimeoutError, OSError):
                continue
            threading.Thread(
                target=self._handle, args=(raw,), daemon=True
            ).start()

    def _handle(self, raw: socket.socket) -> None:
        try:
            client = self._ctx.wrap_socket(raw, server_side=True)
        except (ssl.SSLError, OSError):
            with contextlib.suppress(OSError):
                raw.close()
            return
        try:
            upstream = socket.create_connection(self.backend, timeout=20)
        except OSError:
            with contextlib.suppress(OSError):
                client.close()
            return
        try:
            self._pump(client, upstream)
        finally:
            for s in (client, upstream):
                with contextlib.suppress(OSError):
                    s.close()

    @staticmethod
    def _pump(a: socket.socket, b: socket.socket) -> None:
        # Drain the SSL object's own buffer before consulting select():
        # select only sees the OS socket, so already-decrypted bytes
        # sitting inside the SSL object would stall the relay.
        socks = [a, b]
        while True:
            pending = [s for s in socks
                       if isinstance(s, ssl.SSLSocket) and s.pending()]
            ready = pending or select.select(socks, [], [], 1.0)[0]
            for s in ready:
                other = b if s is a else a
                try:
                    data = s.recv(65536)
                except (ssl.SSLError, OSError):
                    return
                if not data:
                    return
                try:
                    other.sendall(data)
                except OSError:
                    return

    def close(self) -> None:
        self._stop.set()
        with contextlib.suppress(OSError):
            self._sock.close()
        shutil.rmtree(self.tmpdir, ignore_errors=True)


@pytest.fixture(scope="module")
def tls_proxy(conn_params: ConnParams):
    if shutil.which("openssl") is None:
        pytest.skip("openssl not on PATH; needed to generate a test cert")
    proxy = _TlsProxy((conn_params.host, conn_params.port))
    try:
        yield proxy
    finally:
        proxy.close()


def _client_ctx(proxy: _TlsProxy) -> ssl.SSLContext:
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.load_verify_locations(proxy.cert)
    ctx.check_hostname = False  # self-signed CN=127.0.0.1
    return ctx


def _connect(proxy: _TlsProxy, conn_params: ConnParams, **kw):
    return informix_db.connect(
        host="127.0.0.1",
        port=proxy.port,
        user=conn_params.user,
        password=conn_params.password,
        database=conn_params.database,
        server=conn_params.server,
        connect_timeout=20.0,
        read_timeout=45.0,
        tls=_client_ctx(proxy),
        **kw,
    )


# ---------------------------------------------------------------------------
# Traffic
# ---------------------------------------------------------------------------


def test_query_over_tls(tls_proxy, conn_params: ConnParams) -> None:
    with _connect(tls_proxy, conn_params, autocommit=True) as conn:
        cur = conn.cursor()
        cur.execute("SELECT FIRST 3 tabname FROM systables ORDER BY tabid")
        assert len(cur.fetchall()) == 3
        assert conn.server_version, "server_version empty over TLS"


def test_type_round_trip_over_tls(tls_proxy, conn_params: ConnParams) -> None:
    """The types that gave us framing bugs, every byte through TLS."""
    ts = datetime.datetime(2026, 8, 31, 12, 30, 15, 120000)
    row = (
        2001, "PackageRoot", None, "/content/package", 77,
        decimal.Decimal("1234567890123456"), True, ts, "nch",
    )
    with _connect(tls_proxy, conn_params, autocommit=True) as conn:
        cur = conn.cursor()
        cur.execute(
            "CREATE TEMP TABLE t_tls_types ("
            " a INT8 NOT NULL, k LVARCHAR(512), d LVARCHAR(512),"
            " v LVARCHAR(1024), n INT8, dec16 DECIMAL(16), b BOOLEAN,"
            " t DATETIME YEAR TO FRACTION(5), c NCHAR(6))"
        )
        cur.execute(
            "INSERT INTO t_tls_types VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", row
        )
        cur.execute("SELECT a, k, d, v, n, dec16, b, t, c FROM t_tls_types")
        assert cur.fetchone() == row


@pytest.mark.parametrize("size", [1, 4096, 16383, 16384, 16385, 32000])
def test_payload_spans_tls_record_boundary(
    tls_proxy, conn_params: ConnParams, size: int
) -> None:
    """A TLS record holds ~16 KB, so these straddle the boundary where a
    single ``recv`` stops being enough."""
    payload = "x" * size
    with _connect(tls_proxy, conn_params, autocommit=True) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_tls_big (k INT, v LVARCHAR(32000))")
        cur.execute("INSERT INTO t_tls_big VALUES (?, ?)", (1, payload))
        cur.execute("SELECT v, k FROM t_tls_big")
        assert cur.fetchone() == (payload, 1)


@pytest.mark.parametrize("n", [500, 5000])
def test_bulk_fetch_over_tls(
    tls_proxy, conn_params: ConnParams, n: int
) -> None:
    """Total bytes far beyond both one TLS record and the reader's 64 KB
    recv budget, so the top-up loop runs many times."""
    with _connect(tls_proxy, conn_params, autocommit=True) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_tls_bulk (k INT, v VARCHAR(240))")
        cur.executemany(
            "INSERT INTO t_tls_bulk VALUES (?, ?)",
            [(i, f"row{i}-" + "y" * 200) for i in range(n)],
        )
        cur.execute("SELECT k, v FROM t_tls_bulk ORDER BY k")
        rows = cur.fetchall()
        assert len(rows) == n
        assert rows[0][0] == 0
        assert rows[-1][0] == n - 1


def test_error_recovery_over_tls(tls_proxy, conn_params: ConnParams) -> None:
    with _connect(tls_proxy, conn_params, autocommit=True) as conn:
        cur = conn.cursor()
        cur.execute("CREATE TEMP TABLE t_tls_dup (k INT PRIMARY KEY)")
        cur.execute("INSERT INTO t_tls_dup VALUES (1)")
        for _ in range(4):
            with pytest.raises(informix_db.Error):
                cur.execute("INSERT INTO t_tls_dup VALUES (1)")
            with pytest.raises(informix_db.Error):
                cur.execute("SELECT * FROM t_tls_no_such_table_xyz")
            cur.execute("SELECT k FROM t_tls_dup")
            assert cur.fetchall() == [(1,)]


def test_concurrent_tls_connections(tls_proxy, conn_params: ConnParams) -> None:
    """Separate TLS sessions must not cross data."""
    failures: list[str] = []
    lock = threading.Lock()

    def worker(tid: int) -> None:
        tag = f"t{tid}-{'z' * 12}"
        try:
            with _connect(tls_proxy, conn_params, autocommit=True) as conn:
                cur = conn.cursor()
                for r in range(6):
                    token = tid * 1000 + r
                    cur.execute(
                        "SELECT FIRST 1 ?::INT, ?::VARCHAR(24) FROM systables",
                        (token, tag),
                    )
                    if cur.fetchone() != (token, tag):
                        with lock:
                            failures.append(f"thread {tid}: crossed data")
                        return
        except Exception as exc:
            with lock:
                failures.append(f"thread {tid}: {type(exc).__name__}: {exc}")

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert not failures, failures


# ---------------------------------------------------------------------------
# Negative cases — misuse must fail cleanly, never hang or downgrade
# ---------------------------------------------------------------------------


def test_tls_client_against_plaintext_port_fails(
    tls_proxy, conn_params: ConnParams
) -> None:
    with pytest.raises(informix_db.Error):
        informix_db.connect(
            host=conn_params.host, port=conn_params.port,
            user=conn_params.user, password=conn_params.password,
            database=conn_params.database, server=conn_params.server,
            connect_timeout=10.0, read_timeout=10.0,
            tls=_client_ctx(tls_proxy),
        )


def test_plaintext_client_against_tls_port_fails(
    tls_proxy, conn_params: ConnParams
) -> None:
    """Must raise rather than hang — a stalled handshake is the failure
    mode that looks like a dead application."""
    with pytest.raises(informix_db.Error):
        informix_db.connect(
            host="127.0.0.1", port=tls_proxy.port,
            user=conn_params.user, password=conn_params.password,
            database=conn_params.database, server=conn_params.server,
            connect_timeout=10.0, read_timeout=10.0,
        )


def test_verification_rejects_self_signed(
    tls_proxy, conn_params: ConnParams
) -> None:
    """`tls=True` disables verification by design; a caller-supplied
    verifying context must still reject an untrusted cert."""
    strict = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    strict.check_hostname = True
    strict.verify_mode = ssl.CERT_REQUIRED
    with pytest.raises(informix_db.Error):
        informix_db.connect(
            host="127.0.0.1", port=tls_proxy.port,
            user=conn_params.user, password=conn_params.password,
            database=conn_params.database, server=conn_params.server,
            connect_timeout=10.0, read_timeout=10.0, tls=strict,
        )