Ryan Malloy
2896f7e93e
Tests and docs only, no behaviour change. TLS was the last untested surface and it came back clean. Previous coverage stopped at the handshake; everything after it ran only over plain sockets. That gap mattered because SSLSocket.recv is not socket.recv: it returns at most one TLS record's worth of plaintext however much you ask for, 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 had never been pressured the same way. tests/test_tls_traffic.py runs real SQLI through a TLS-terminating proxy: the framing-bug types end to end, payloads at 1/4096/16383/16384/ 16385/32000 bytes (straddling the ~16 KB record boundary), 500- and 5000-row bulk fetches, error recovery, concurrent TLS sessions, and three negative cases — TLS client against a plaintext port, plaintext client against a TLS port (must raise, not hang), and a verifying context rejecting a self-signed cert. All clean on all three servers. The buffered reader handles SSLSocket semantics correctly. Scope, stated plainly: the proxy supplies the TLS half, so this covers the driver's TLS path — the half we own. It does NOT cover IBM's server-side TLS listener. Setting one up on the developer-edition image was attempted and abandoned: Informix 15 wants a PKCS#12 keystore (onkstash takes a .p12, not the older CMS .kdb) and the engine kept rejecting the stash with GSK_ERROR_BAD_KEYFILE_PASSWORD even with a keystore GSKit itself could open. That half is IBM's code; everything below ssl.wrap_socket is identical either way. 414/414 integration on 15, 14.10 and 12.10 (was 399). Every surface is now fuzzed: type framing, fetch batching, error recovery, cursor lifecycle, threads, pooling, transactions, async cancellation, executemany partial failure, scrollable cursors, smart LOBs, and TLS.
334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""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,
|
|
)
|