Phase 14: TLS / SSL transport

Optional TLS via the ``tls`` parameter on connect() and IfxSocket.
Three modes:

  tls=False (default)  — plain TCP, current behavior unchanged
  tls=True             — TLS w/ verification disabled (dev / self-signed)
  tls=ssl.SSLContext   — caller-supplied context (production)

Plus tls_server_hostname for SNI / cert verification.

Architectural choice: Informix uses dedicated TLS-enabled listener
ports (configured in server's sqlhosts), NOT STARTTLS upgrade. The
SSL handshake runs immediately after TCP connect with no protocol-
level negotiation. Wrapping happens inside IfxSocket.__init__ so the
rest of the protocol layer (login PDU, SQ_BIND, fast-path, file
transfer) is fully unaware of whether TLS is in use.

Why tls=True defaults to insecure: most Informix dev installations
use self-signed certs. tls=True produces a context with
check_hostname=False and verify_mode=CERT_NONE. Minimum protocol is
still TLSv1.2 (per ssl.PROTOCOL_TLS_CLIENT). Production users are
expected to pass ssl.SSLContext explicitly.

Tests: 5 unit tests in test_tls.py
* tls=True dev context properties
* default context uses TLSv1.2+
* real handshake against in-process TLS echo server (proves wrap_socket
  works end-to-end)
* custom SSLContext honored verbatim
* tls=True against non-TLS port raises OperationalError clearly

Test certs are generated via openssl CLI subprocess instead of
adding cryptography as a dev dep (saves ~5MB transitive deps for
one phase).

Total: 69 unit + 139 integration = 208 tests.

Architectural milestone: with Phase 14 complete, the driver now
implements EVERYTHING in the SQLI wire-protocol family that a Python
application needs. Remaining backlog (async, pooling) is library-
design work, not protocol work.

docs/DECISION_LOG.md

@@ -1013,6 +1013,68 @@ The only families still unimplemented are: TLS handshake (`STARTTLS`), and the c
 
 ---
 
+## 2026-05-04 — Phase 14: TLS/SSL transport
+
+**Status**: active
+**Decision**: Optional TLS via a new ``tls`` parameter on ``connect()`` and the ``IfxSocket`` constructor. Three modes:
+
+```python
+# 1. Plain TCP (default, current behavior)
+informix_db.connect(host, port=9088, ...)
+
+# 2. TLS with verification DISABLED — dev / self-signed servers
+informix_db.connect(host, port=9089, ..., tls=True)
+
+# 3. Caller-supplied ssl.SSLContext — production
+ctx = ssl.create_default_context(cafile="/path/to/ca.pem")
+informix_db.connect(host, port=9089, ..., tls=ctx)
+```
+
+### Why "no STARTTLS" — Informix uses dedicated TLS ports
+
+Postgres and others negotiate TLS via STARTTLS-style upgrade: connect plain, send a magic byte, receive ack, wrap socket. Informix does not — TLS-enabled listeners run on **separate ports** (the server's ``sqlhosts`` config), and the SSL handshake runs immediately after TCP connect with no protocol-level negotiation. From the driver's perspective this is simpler: just wrap the socket before any SQLI bytes flow.
+
+The wrapping happens inside ``IfxSocket.__init__``, so the rest of the protocol layer (login PDU assembly, SQ_BIND, fast-path, file transfer) is fully unaware of whether TLS is in use. The same code path drives encrypted and plain connections.
+
+### Why ``tls=True`` defaults to *insecure*
+
+Most Informix dev installations use self-signed certs. The first thing users hit when adding TLS is a "self-signed certificate verification failed" error, which then sends them down a path of disabling verification anyway. ``tls=True`` short-circuits that: it produces a context with ``check_hostname=False`` and ``verify_mode=CERT_NONE``. The minimum protocol is still TLSv1.2 (per ``ssl.PROTOCOL_TLS_CLIENT``).
+
+For production, ``tls=ssl.SSLContext(...)`` is the explicit, secure path. The defaults are documented as "dev / loopback only".
+
+### Test strategy: avoid making cryptography a dev dep
+
+Initial draft used the ``cryptography`` package to mint a test certificate — which would have added a 5MB transitive dep just for one phase's tests. Switched to spawning ``openssl`` CLI as a subprocess (skipping the test if not available). Cleaner, no new Python deps.
+
+The integration tests run against a tiny in-process Python TLS echo server using ``ssl.SSLContext(PROTOCOL_TLS_SERVER)``. Each test:
+1. Generates an ephemeral self-signed cert via ``openssl req -x509``
+2. Spins up a one-shot TLS server in a daemon thread
+3. Connects via ``IfxSocket(tls=True)`` (or with a custom context)
+4. Sends/receives an echo round-trip to prove the encrypted channel works
+
+### Tests delivered
+
+5 unit tests in `tests/test_tls.py`:
+- `tls=True` dev context has `check_hostname=False` and `verify_mode=CERT_NONE`
+- Default context uses TLSv1.2+
+- Real handshake against in-process TLS echo server (proves wrap_socket works)
+- Custom `SSLContext` honored verbatim
+- `tls=True` against a non-TLS port raises `OperationalError("TLS handshake failed...")`
+
+Total: **69 unit + 139 integration = 208 tests**.
+
+### What's now done at the protocol level
+
+The driver now implements **everything in the SQLI wire-protocol family that a Python application needs**:
+- Login → cursor → fetch (Phases 0-4)
+- All common types (Phases 5-12)
+- Transactions, file transfer, fast-path, smart-LOBs (Phases 7-13)
+- TLS transport (this phase)
+
+The remaining backlog (async, pooling) is **library-design** work, not protocol work.
+
+---
+
 ## (template — copy below this line for new entries)
 
 ```

src/informix_db/__init__.py

@@ -20,6 +20,7 @@ For the full design, see ``docs/PROTOCOL_NOTES.md`` and
 
 from __future__ import annotations
 
+import ssl
 from importlib.metadata import PackageNotFoundError, version
 
 from .connections import Connection
@@ -93,6 +94,8 @@ def connect(
     client_locale: str = "en_US.8859-1",
     env: dict[str, str] | None = None,
     autocommit: bool = False,
+    tls: bool | ssl.SSLContext = False,
+    tls_server_hostname: str | None = None,
 ) -> Connection:
     """Open a connection to an Informix server.
 
@@ -104,6 +107,19 @@ def connect(
 
     ``database`` may be ``None`` to log in without selecting a database;
     the server still requires a successful login for this to work.
+
+    TLS:
+      ``tls=False`` (default) — plain TCP.
+      ``tls=True`` — TLS with **verification disabled** (dev / self-signed
+        servers; suitable for local development only). Defaults to
+        TLSv1.2+ via ``ssl.PROTOCOL_TLS_CLIENT``.
+      ``tls=ssl.SSLContext(...)`` — bring your own context for production
+        (cert pinning, CA bundles, ALPN, etc.).
+
+    Informix uses dedicated TLS-enabled listener ports (typically
+    configured separately on the server's ``sqlhosts`` file) rather than
+    STARTTLS-style upgrade — point ``port`` at the TLS listener (often
+    ``9089``) when ``tls`` is enabled.
     """
     return Connection(
         host=host,
@@ -112,6 +128,8 @@ def connect(
         password=password,
         database=database,
         server=server,
+        tls=tls,
+        tls_server_hostname=tls_server_hostname,
         connect_timeout=connect_timeout,
         read_timeout=read_timeout,
         keepalive=keepalive,

src/informix_db/_socket.py

@@ -7,15 +7,39 @@ class composes one of these with the protocol layer.
 Mirrors PyMySQL's ``Connection._read_bytes`` / ``_write_bytes`` shape.
 TCP_NODELAY is enabled by default to match IBM JDBC's behavior
 (see ``com.informix.asf.Connection.openSocket``).
+
+Phase 14: optional TLS via ``tls`` parameter — Informix uses dedicated
+TLS-enabled listener ports rather than STARTTLS upgrade, so the
+SSL handshake runs immediately after TCP connect, transparent to
+the rest of the protocol layer.
 """
 
 from __future__ import annotations
 
 import contextlib
 import socket
+import ssl
 
 from .exceptions import InterfaceError, OperationalError
 
+# A ``tls`` parameter to ``IfxSocket`` accepts:
+#   False (default) — plain TCP
+#   True            — TLS with verification disabled (dev / self-signed)
+#   ssl.SSLContext  — caller-supplied context for full control
+TlsOption = bool | ssl.SSLContext
+
+
+def _make_default_dev_context() -> ssl.SSLContext:
+    """A loose context for ``tls=True`` — accepts self-signed certs.
+
+    Use this for dev / loopback Informix instances. Production code
+    should pass a properly-configured ``ssl.SSLContext`` instead.
+    """
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    return ctx
+
 
 class IfxSocket:
     """Owns a connected TCP socket and provides exact-N read/write."""
@@ -27,6 +51,8 @@ class IfxSocket:
         connect_timeout: float | None = None,
         read_timeout: float | None = None,
         keepalive: bool = False,
+        tls: TlsOption = False,
+        tls_server_hostname: str | None = None,
     ):
         self._host = host
         self._port = port
@@ -41,6 +67,26 @@ class IfxSocket:
         sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
         if keepalive:
             sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+
+        # TLS handshake (if enabled). Informix uses dedicated TLS ports —
+        # the server starts in TLS mode rather than negotiating
+        # STARTTLS-style mid-connection. We do the handshake right after
+        # TCP connect and before any SQLI bytes flow.
+        if tls:
+            ctx = tls if isinstance(tls, ssl.SSLContext) else _make_default_dev_context()
+            try:
+                sock = ctx.wrap_socket(
+                    sock,
+                    server_hostname=tls_server_hostname or host,
+                    do_handshake_on_connect=True,
+                )
+            except (ssl.SSLError, OSError) as e:
+                with contextlib.suppress(OSError):
+                    sock.close()
+                raise OperationalError(
+                    f"TLS handshake failed against {host}:{port}: {e}"
+                ) from e
+
         if read_timeout is not None:
             sock.settimeout(read_timeout)
         else:

src/informix_db/connections.py

@@ -13,6 +13,7 @@ from __future__ import annotations
 
 import os
 import socket as socket_mod
+import ssl
 import struct
 import threading
 from io import BytesIO
@@ -86,6 +87,8 @@ class Connection:
         client_locale: str = "en_US.8859-1",
         env: dict[str, str] | None = None,
         autocommit: bool = False,  # honored from Phase 3 onward
+        tls: bool | ssl.SSLContext = False,
+        tls_server_hostname: str | None = None,
     ):
         self._host = host
         self._port = port
@@ -123,6 +126,8 @@ class Connection:
             connect_timeout=connect_timeout,
             read_timeout=read_timeout,
             keepalive=keepalive,
+            tls=tls,
+            tls_server_hostname=tls_server_hostname,
         )
 
         try:

tests/test_tls.py

@@ -0,0 +1,185 @@
+"""Phase 14 unit tests — TLS layer for IfxSocket.
+
+Informix uses dedicated TLS-enabled listener ports rather than
+STARTTLS-style upgrade. The SSL handshake runs immediately after
+TCP connect, transparent to the SQLI protocol layer.
+
+These tests exercise the wrapping logic against a local Python
+TLS echo server — they don't need an actual TLS-enabled Informix
+listener. The integration tests for "plain TCP works" and
+"connecting tls=True to a non-TLS port fails" live alongside the
+other smoke tests.
+"""
+
+from __future__ import annotations
+
+import socket
+import ssl
+import threading
+from pathlib import Path
+
+import pytest
+
+from informix_db._socket import IfxSocket, _make_default_dev_context
+
+# -------- API surface --------
+
+
+def test_default_dev_context_has_no_verification() -> None:
+    """``tls=True`` produces a context safe for self-signed dev servers."""
+    ctx = _make_default_dev_context()
+    assert ctx.check_hostname is False
+    assert ctx.verify_mode == ssl.CERT_NONE
+
+
+def test_dev_context_uses_modern_protocol() -> None:
+    """The default context excludes TLSv1.0/1.1 (Python's SSLContext defaults)."""
+    ctx = _make_default_dev_context()
+    # ssl.PROTOCOL_TLS_CLIENT prevents downgrade attacks; minimum TLS 1.2
+    assert ctx.minimum_version >= ssl.TLSVersion.TLSv1_2
+
+
+# -------- End-to-end wrapping: local TLS echo server --------
+
+
+def _make_self_signed_cert(tmp_path: Path) -> tuple[Path, Path]:
+    """Generate an ephemeral self-signed cert via the ``openssl`` CLI.
+
+    Skips the test if ``openssl`` isn't on PATH. Avoids adding
+    ``cryptography`` as a hard dev dep for one corner-case test.
+    """
+    import shutil
+    import subprocess
+
+    if not shutil.which("openssl"):
+        pytest.skip("openssl CLI not available; needed to generate test cert")
+
+    cert_path = tmp_path / "cert.pem"
+    key_path = tmp_path / "key.pem"
+    subprocess.run(
+        [
+            "openssl", "req", "-x509", "-newkey", "rsa:2048",
+            "-keyout", str(key_path), "-out", str(cert_path),
+            "-days", "1", "-nodes",
+            "-subj", "/CN=localhost",
+            "-addext", "subjectAltName=DNS:localhost",
+        ],
+        check=True, capture_output=True,
+    )
+    return cert_path, key_path
+
+
+def _start_tls_echo_server(
+    cert_path: Path, key_path: Path
+) -> tuple[int, threading.Thread, threading.Event]:
+    """Spin up a one-shot TLS echo server on an ephemeral port.
+
+    Returns ``(port, thread, ready_event)``. The thread accepts ONE
+    connection, echoes back any bytes sent, and exits.
+    """
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    ctx.load_cert_chain(certfile=str(cert_path), keyfile=str(key_path))
+
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.bind(("127.0.0.1", 0))
+    listener.listen(1)
+    port = listener.getsockname()[1]
+    ready = threading.Event()
+
+    def serve() -> None:
+        ready.set()
+        try:
+            client, _ = listener.accept()
+            with ctx.wrap_socket(client, server_side=True) as sslsock:
+                while True:
+                    data = sslsock.recv(4096)
+                    if not data:
+                        break
+                    sslsock.sendall(data)
+        except Exception:
+            pass
+        finally:
+            listener.close()
+
+    t = threading.Thread(target=serve, daemon=True)
+    t.start()
+    return port, t, ready
+
+
+def test_tls_handshake_succeeds_against_real_tls_server(tmp_path: Path) -> None:
+    """``IfxSocket(tls=True)`` completes a TLS handshake with a self-signed server."""
+    cert_path, key_path = _make_self_signed_cert(tmp_path)
+    port, _t, ready = _start_tls_echo_server(cert_path, key_path)
+    ready.wait(timeout=2.0)
+
+    sock = IfxSocket(
+        "127.0.0.1", port,
+        connect_timeout=2.0,
+        read_timeout=2.0,
+        tls=True,
+    )
+    try:
+        # Echo round-trip proves the encrypted channel works
+        sock.write_all(b"phase 14 tls test")
+        echoed = sock.read_exact(len(b"phase 14 tls test"))
+        assert echoed == b"phase 14 tls test"
+    finally:
+        sock.close()
+
+
+def test_tls_handshake_with_custom_context(tmp_path: Path) -> None:
+    """Caller-supplied ``ssl.SSLContext`` is honored verbatim."""
+    cert_path, key_path = _make_self_signed_cert(tmp_path)
+    port, _t, ready = _start_tls_echo_server(cert_path, key_path)
+    ready.wait(timeout=2.0)
+
+    # Caller can supply a context with custom CA verification
+    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    ctx.check_hostname = True
+    ctx.verify_mode = ssl.CERT_REQUIRED
+    ctx.load_verify_locations(cafile=str(cert_path))
+
+    sock = IfxSocket(
+        "127.0.0.1", port,
+        connect_timeout=2.0,
+        read_timeout=2.0,
+        tls=ctx,
+        tls_server_hostname="localhost",
+    )
+    try:
+        sock.write_all(b"verified")
+        assert sock.read_exact(8) == b"verified"
+    finally:
+        sock.close()
+
+
+def test_tls_against_plain_socket_raises_operational_error() -> None:
+    """``tls=True`` against a non-TLS port produces a clear error."""
+    # Use a one-shot plain TCP server that hangs up immediately
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.bind(("127.0.0.1", 0))
+    listener.listen(1)
+    port = listener.getsockname()[1]
+
+    accepted: list[socket.socket] = []
+
+    def serve() -> None:
+        client, _ = listener.accept()
+        accepted.append(client)
+        client.close()  # drop the connection — looks like garbage to TLS
+
+    t = threading.Thread(target=serve, daemon=True)
+    t.start()
+
+    try:
+        from informix_db.exceptions import OperationalError
+        with pytest.raises(OperationalError, match="TLS handshake failed"):
+            IfxSocket(
+                "127.0.0.1", port,
+                connect_timeout=2.0,
+                read_timeout=2.0,
+                tls=True,
+            )
+    finally:
+        listener.close()
+        t.join(timeout=2.0)