diff --git a/docs/authentication.md b/docs/authentication.md
index 0ab6625f..80b435d3 100644
--- a/docs/authentication.md
+++ b/docs/authentication.md
@@ -3,7 +3,8 @@
 Kalshi uses RSA-PSS request signing. You'll need:
 
 - A **key ID** (string, identifies the key on Kalshi's side).
-- A **private key** — RSA, **PEM-encoded, PKCS#8, unencrypted**.
+- A **private key** — RSA, PEM-encoded, PKCS#8. May be unencrypted or
+  encrypted; encrypted keys require a passphrase (see below).
 
 Generate the pair in your [Kalshi account settings](https://kalshi.com/account/profile)
 and download the PEM file. The signing scheme used internally is
@@ -175,13 +176,68 @@ load, check:
 - **Must be PKCS#8** (`-----BEGIN PRIVATE KEY-----`). Legacy PKCS#1
   (`-----BEGIN RSA PRIVATE KEY-----`) is supported by the underlying
   `cryptography` library on a best-effort basis.
-- **Must be unencrypted.** Passphrase-protected keys raise `KalshiAuthError`
-  with a hint. Strip the passphrase with `openssl pkey`:
+- **Passphrase-protected keys are supported via `password=`** — see below. If
+  you load an encrypted PEM without supplying a password, `KalshiAuth` raises
+  `KalshiAuthError` with a hint pointing at the `password=` parameter and the
+  `KALSHI_PRIVATE_KEY_PASSPHRASE` env var.
 
-    ```bash
-    openssl pkey -in encrypted.pem -out unencrypted.pem
-    ```
+### Passphrase-protected keys
 
+Encrypted PEMs are a recommended practice for trading keys: you keep the key
+at-rest encrypted on disk or in a secret manager. The SDK accepts a
+`password=` keyword on every loader so you don't have to write a plaintext key
+to disk:
+
+```python
+from kalshi import KalshiAuth, KalshiClient
+
+# Inline literal (don't hard-code in real code — pull from a secret manager).
+auth = KalshiAuth.from_key_path(
+    "your-key-id",
+    "~/.kalshi/encrypted_key.pem",
+    password="hunter2",
+)
+
+# Deferred fetch: pass a zero-arg callable. The SDK invokes it during load,
+# so the secret never sits in the calling frame longer than necessary.
+auth = KalshiAuth.from_pem(
+    "your-key-id",
+    pem_string,
+    password=lambda: secret_manager.get("kalshi/passphrase"),
+)
+
+# Bytes are accepted too.
+auth = KalshiAuth.from_pem("your-key-id", pem, password=b"hunter2")
+```
+
+For env-driven flows, set `KALSHI_PRIVATE_KEY_PASSPHRASE` alongside
+`KALSHI_PRIVATE_KEY` / `KALSHI_PRIVATE_KEY_PATH`:
+
+```bash
+export KALSHI_KEY_ID="..."
+export KALSHI_PRIVATE_KEY_PATH="~/.kalshi/encrypted_key.pem"
+export KALSHI_PRIVATE_KEY_PASSPHRASE="hunter2"
+```
+
+```python
+from kalshi import KalshiClient
+
+with KalshiClient.from_env() as client:
+    ...
+```
+
+An explicit `password=` argument always wins over
+`KALSHI_PRIVATE_KEY_PASSPHRASE` when both are supplied. A wrong passphrase
+raises `KalshiAuthError` ("Invalid PEM private key…"); a missing passphrase
+for an encrypted PEM raises `KalshiAuthError` pointing at the `password=`
+parameter.
+
+If you'd rather store the key unencrypted, you can still strip the passphrase
+with `openssl pkey`:
+
+```bash
+openssl pkey -in encrypted.pem -out unencrypted.pem
+```
 ### Manual signing
 
 `KalshiAuth.sign_request(method, path, timestamp_ms=None)` is part of the
diff --git a/docs/configuration.md b/docs/configuration.md
index 7767ff28..501b5b9d 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -80,9 +80,10 @@ This catches plaintext config slip-ups before any request is sent.
 KalshiConfig(extra_headers={"User-Agent": "acme-trader/2.1 (+ops@acme.co)"})
 ```
 
-`extra_headers` is merged into every request after the SDK's own headers, so
-you can override anything the SDK sets (rarely a good idea outside
-User-Agent).
+`extra_headers` is set at the `httpx.Client` level and is **overridden** by the
+SDK's per-request `KALSHI-ACCESS-*` headers, so it cannot be used to forge or
+replace auth headers. It can only override client-level defaults that the SDK
+doesn't re-set on every request — `User-Agent` is the realistic use case.
 
 ## HTTP/2
 
@@ -90,9 +91,12 @@ User-Agent).
 KalshiConfig(http2=True)
 ```
 
-Requires `pip install 'httpx[http2]'` — the SDK doesn't pin h2 as a hard
-dependency. Once enabled, the same setting flows into the WebSocket client too
-(WebSocket runs on top of an HTTP/1.1 upgrade today; this only affects REST).
+Requires the `h2` package — install via `pip install 'kalshi-sdk[http2]'` (or
+`pip install 'kalshi-sdk[all]'`). `KalshiConfig` validates this at construction
+time and raises `ValueError` immediately if `h2` is missing, rather than
+deferring the failure to the first request. Once enabled, the same setting
+flows into the WebSocket client too (WebSocket runs on top of an HTTP/1.1
+upgrade today; this only affects REST).
 
 ## Connection-pool limits
 
diff --git a/kalshi/_base_client.py b/kalshi/_base_client.py
index 030fdb62..22083757 100644
--- a/kalshi/_base_client.py
+++ b/kalshi/_base_client.py
@@ -6,10 +6,12 @@
 
 from __future__ import annotations
 
+import email.utils
 import logging
 import math
 import random
 import time
+from datetime import UTC, datetime
 from typing import Any
 from urllib.parse import urlparse
 
@@ -90,7 +92,26 @@ def _map_error(response: httpx.Response) -> KalshiError:
                 if retry_after_val < 0 or not math.isfinite(retry_after_val):
                     retry_after_val = None
             except ValueError:
-                retry_after_val = None  # HTTP-date format, fall back to computed backoff
+                # RFC 7231 §7.1.3: Retry-After may also be an HTTP-date.
+                try:
+                    dt = email.utils.parsedate_to_datetime(retry_after)
+                except (TypeError, ValueError):
+                    dt = None
+                if dt is None:
+                    logger.debug(
+                        "Retry-After %r is neither delta-seconds nor HTTP-date; "
+                        "falling back to computed backoff",
+                        retry_after,
+                    )
+                else:
+                    # RFC 5322 dates without a tz are interpreted as UTC by
+                    # parsedate_to_datetime; ensure aware before subtracting.
+                    if dt.tzinfo is None:
+                        dt = dt.replace(tzinfo=UTC)
+                    delta = (dt - datetime.now(tz=UTC)).total_seconds()
+                    # Clamp negatives (date in the past) to 0 — retry immediately.
+                    # The transport caller caps at config.retry_max_delay.
+                    retry_after_val = max(0.0, delta)
         return KalshiRateLimitError(
             message=message, status_code=status, retry_after=retry_after_val
         )
@@ -157,17 +178,33 @@ def request(
         *,
         params: dict[str, Any] | None = None,
         json: dict[str, Any] | None = None,
+        extra_headers: dict[str, str] | None = None,
     ) -> httpx.Response:
-        """Make an authenticated HTTP request with retry logic."""
+        """Make an authenticated HTTP request with retry logic.
+
+        ``extra_headers`` are merged per-request on top of
+        ``config.extra_headers`` (so a per-call ``X-Request-ID`` overrides a
+        client-default one), but the ``KALSHI-ACCESS-*`` auth headers always
+        win — callers cannot forge them via ``extra_headers``.
+        """
+        # P1.6: canonicalize trailing slash BEFORE both signing and httpx call
+        # so the wire path and the signed payload stay byte-identical. Root "/"
+        # is preserved.
+        if path != "/" and path.endswith("/"):
+            path = path.rstrip("/") or "/"
         # Sign with path-only (not full URL). Kalshi expects: /trade-api/v2/endpoint
         sign_path = self._base_path + path
         last_error: KalshiError | None = None
         # #193: wall-clock budget across the whole request including retries.
         start = time.monotonic()
         total_timeout = self._config.total_timeout
+        config_extra = self._config.extra_headers or {}
+        per_call_extra = extra_headers or {}
 
         for attempt in range(self._config.max_retries + 1):
             auth_headers = self._auth.sign_request(method.upper(), sign_path) if self._auth else {}
+            # Order matters: config defaults < per-request overrides < signed auth.
+            merged_headers = {**config_extra, **per_call_extra, **auth_headers}
 
             logger.debug(
                 "Request: %s %s (attempt %d/%d)",
@@ -183,7 +220,7 @@ def request(
                     url=path,
                     params=params,
                     json=json,
-                    headers=auth_headers,
+                    headers=merged_headers,
                 )
             except httpx.PoolTimeout as e:
                 # #204: pool exhaustion never reached the wire — safe to retry
@@ -312,22 +349,32 @@ async def request(
         *,
         params: dict[str, Any] | None = None,
         json: dict[str, Any] | None = None,
+        extra_headers: dict[str, str] | None = None,
     ) -> httpx.Response:
-        """Make an authenticated async HTTP request with retry logic."""
+        """Make an authenticated async HTTP request with retry logic.
+
+        See :meth:`SyncTransport.request` for ``extra_headers`` semantics.
+        """
         import asyncio
 
+        # P1.6: canonicalize trailing slash BEFORE both signing and httpx call.
+        if path != "/" and path.endswith("/"):
+            path = path.rstrip("/") or "/"
         # Sign with path-only (not full URL). Kalshi expects: /trade-api/v2/endpoint
         sign_path = self._base_path + path
         last_error: KalshiError | None = None
         # #193: wall-clock budget across the whole request including retries.
         start = time.monotonic()
         total_timeout = self._config.total_timeout
+        config_extra = self._config.extra_headers or {}
+        per_call_extra = extra_headers or {}
 
         for attempt in range(self._config.max_retries + 1):
             if self._auth:
                 auth_headers = await self._auth.sign_request_async(method.upper(), sign_path)
             else:
                 auth_headers = {}
+            merged_headers = {**config_extra, **per_call_extra, **auth_headers}
 
             logger.debug(
                 "Async request: %s %s (attempt %d/%d)",
@@ -343,7 +390,7 @@ async def request(
                     url=path,
                     params=params,
                     json=json,
-                    headers=auth_headers,
+                    headers=merged_headers,
                 )
             except httpx.PoolTimeout as e:
                 # #204: pool exhaustion never reached the wire — safe to retry
diff --git a/kalshi/auth.py b/kalshi/auth.py
index ca8b24f5..e79df9c6 100644
--- a/kalshi/auth.py
+++ b/kalshi/auth.py
@@ -20,6 +20,7 @@
 import re
 import threading
 import time
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
@@ -72,10 +73,20 @@ def __init__(self, key_id: str, private_key: rsa.RSAPrivateKey) -> None:
         self._sign_executor_lock = threading.Lock()
 
     @classmethod
-    def from_key_path(cls, key_id: str, key_path: str | Path) -> KalshiAuth:
+    def from_key_path(
+        cls,
+        key_id: str,
+        key_path: str | Path,
+        *,
+        password: bytes | str | Callable[[], bytes | str] | None = None,
+    ) -> KalshiAuth:
         """Load auth from a PEM private key file.
 
         Supports ~ expansion (e.g., ~/kalshi.pem).
+
+        ``password`` accepts ``str`` / ``bytes`` / a zero-arg callable that
+        returns either (the callable form lets you defer fetching the secret
+        from a vault until load-time).
         """
         expanded = Path(key_path).expanduser()
         if not expanded.exists():
@@ -84,24 +95,44 @@ def from_key_path(cls, key_id: str, key_path: str | Path) -> KalshiAuth:
             pem_data = expanded.read_bytes()
         except PermissionError as e:
             raise KalshiAuthError(f"Permission denied reading private key: {expanded}") from e
-        return cls.from_pem(key_id, pem_data)
+        return cls.from_pem(key_id, pem_data, password=password)
 
     @classmethod
-    def from_pem(cls, key_id: str, pem_data: str | bytes) -> KalshiAuth:
-        """Load auth from PEM-encoded private key content."""
+    def from_pem(
+        cls,
+        key_id: str,
+        pem_data: str | bytes,
+        *,
+        password: bytes | str | Callable[[], bytes | str] | None = None,
+    ) -> KalshiAuth:
+        """Load auth from PEM-encoded private key content.
+
+        ``password`` accepts ``str`` / ``bytes`` / a zero-arg callable that
+        returns either; pass ``None`` (default) for an unencrypted PEM.
+        """
         if isinstance(pem_data, str):
             pem_data = pem_data.encode("utf-8")
+        pw_bytes: bytes | None = None
+        if password is not None:
+            raw = password() if callable(password) else password
+            pw_bytes = raw.encode("utf-8") if isinstance(raw, str) else raw
         try:
-            private_key = serialization.load_pem_private_key(pem_data, password=None)
+            private_key = serialization.load_pem_private_key(pem_data, password=pw_bytes)
         except TypeError as e:
+            # cryptography raises TypeError when the PEM is encrypted and
+            # ``password=None``, or when an unencrypted PEM is given a password.
             raise KalshiAuthError(
-                "Passphrase-protected private keys are not supported. "
-                "Remove the passphrase with: openssl pkey -in key.pem -out key_nopass.pem"
+                "Passphrase-protected private key requires a password. "
+                "Pass `password=` to KalshiAuth.from_pem / from_key_path / from_env, "
+                "set the KALSHI_PRIVATE_KEY_PASSPHRASE environment variable, or "
+                "strip the passphrase with: openssl pkey -in key.pem -out key_nopass.pem"
             ) from e
         except (ValueError, UnsupportedAlgorithm) as e:
+            # Wrong password and malformed PEM both surface here as ValueError.
             raise KalshiAuthError(
                 f"Invalid PEM private key: {e}. Ensure the key is an RSA private key "
-                "in PKCS8 PEM format (-----BEGIN PRIVATE KEY-----)."
+                "in PKCS8 PEM format (-----BEGIN PRIVATE KEY-----) and that "
+                "the supplied `password=` (if any) is correct."
             ) from e
         if not isinstance(private_key, rsa.RSAPrivateKey):
             raise KalshiAuthError(
@@ -111,12 +142,20 @@ def from_pem(cls, key_id: str, pem_data: str | bytes) -> KalshiAuth:
         return cls(key_id, private_key)
 
     @classmethod
-    def from_env(cls) -> KalshiAuth:
+    def from_env(
+        cls,
+        *,
+        password: bytes | str | Callable[[], bytes | str] | None = None,
+    ) -> KalshiAuth:
         """Load auth from environment variables.
 
         Reads:
             KALSHI_KEY_ID (required)
             KALSHI_PRIVATE_KEY (PEM string) or KALSHI_PRIVATE_KEY_PATH (file path)
+            KALSHI_PRIVATE_KEY_PASSPHRASE (optional; used when the PEM is encrypted)
+
+        The explicit ``password=`` keyword overrides
+        ``KALSHI_PRIVATE_KEY_PASSPHRASE`` when both are supplied.
         """
         key_id = os.environ.get("KALSHI_KEY_ID")
         if not key_id:
@@ -125,13 +164,17 @@ def from_env(cls) -> KalshiAuth:
                 "Set it to your Kalshi API key ID."
             )
 
+        resolved_password = (
+            password if password is not None else os.environ.get("KALSHI_PRIVATE_KEY_PASSPHRASE")
+        )
+
         pem_string = os.environ.get("KALSHI_PRIVATE_KEY")
         if pem_string:
-            return cls.from_pem(key_id, pem_string)
+            return cls.from_pem(key_id, pem_string, password=resolved_password)
 
         key_path = os.environ.get("KALSHI_PRIVATE_KEY_PATH")
         if key_path:
-            return cls.from_key_path(key_id, key_path)
+            return cls.from_key_path(key_id, key_path, password=resolved_password)
 
         raise KalshiAuthError(
             "Neither KALSHI_PRIVATE_KEY nor KALSHI_PRIVATE_KEY_PATH is set. "
@@ -139,7 +182,11 @@ def from_env(cls) -> KalshiAuth:
         )
 
     @classmethod
-    def try_from_env(cls) -> KalshiAuth | None:
+    def try_from_env(
+        cls,
+        *,
+        password: bytes | str | Callable[[], bytes | str] | None = None,
+    ) -> KalshiAuth | None:
         """Load auth from environment variables, returning None if not configured.
 
         Returns None if KALSHI_KEY_ID is not set, or if neither
@@ -148,18 +195,25 @@ def try_from_env(cls) -> KalshiAuth | None:
         Note: Does not raise on *missing* variables, but may still raise
         KalshiAuthError if variables are set with invalid data (e.g.,
         malformed PEM content or nonexistent key file path).
+
+        See :meth:`from_env` for the ``password`` /
+        ``KALSHI_PRIVATE_KEY_PASSPHRASE`` semantics.
         """
         key_id = os.environ.get("KALSHI_KEY_ID")
         if not key_id:
             return None
 
+        resolved_password = (
+            password if password is not None else os.environ.get("KALSHI_PRIVATE_KEY_PASSPHRASE")
+        )
+
         pem_string = os.environ.get("KALSHI_PRIVATE_KEY")
         if pem_string:
-            return cls.from_pem(key_id, pem_string)
+            return cls.from_pem(key_id, pem_string, password=resolved_password)
 
         key_path = os.environ.get("KALSHI_PRIVATE_KEY_PATH")
         if key_path:
-            return cls.from_key_path(key_id, key_path)
+            return cls.from_key_path(key_id, key_path, password=resolved_password)
 
         logger.warning(
             "KALSHI_KEY_ID is set but neither KALSHI_PRIVATE_KEY nor "
diff --git a/kalshi/config.py b/kalshi/config.py
index 4c0853e8..df2b2829 100644
--- a/kalshi/config.py
+++ b/kalshi/config.py
@@ -88,6 +88,14 @@ def __post_init__(self) -> None:
                 f"KalshiConfig.base_url must include the /trade-api/v2 path "
                 f"component, got base_url={self.base_url!r}"
             )
+        if self.http2:
+            import importlib.util
+
+            if importlib.util.find_spec("h2") is None:
+                raise ValueError(
+                    "http2=True requires the 'h2' package — install with "
+                    "`pip install kalshi-sdk[http2]`"
+                )
 
     @staticmethod
     def _validate_url(url: str, field_name: str, *, secure: str, plaintext: str) -> None:
diff --git a/kalshi/testing/__init__.py b/kalshi/testing/__init__.py
index 15f1ae12..2518a33d 100644
--- a/kalshi/testing/__init__.py
+++ b/kalshi/testing/__init__.py
@@ -10,11 +10,20 @@
 
 .. warning::
 
-    Recorded fixtures contain the full response body returned by Kalshi. If you
-    record against an account with real funds you will write balances, positions,
-    order history, and any PII the API returns to disk. **Always ``.gitignore``
-    the fixture directory unless you have manually scrubbed the JSON, and prefer
-    recording against the demo environment whenever possible.**
+    Recorded fixtures contain the full response **body** AND response
+    **headers** returned by Kalshi. If you record against an account with
+    real funds you will write balances, positions, order history, and any
+    PII the API returns to disk; response headers may carry rate-limit
+    counters, request IDs, and server timestamps.
+
+    :class:`RecordingTransport` (and the async variant) scrub a small set
+    of high-risk response headers by default — ``Set-Cookie``,
+    ``Authorization``, and anything matching
+    ``(?i)^x-kalshi-.*-(id|key|account|user).*$`` — but **do not assume the
+    default is sufficient for your environment**. Always ``.gitignore`` the
+    fixture directory unless you have manually scrubbed the JSON, prefer
+    recording against the demo environment whenever possible, and pass
+    ``response_header_filter=`` to extend the deny-list.
 
 Usage::
 
diff --git a/kalshi/testing/_fixtures.py b/kalshi/testing/_fixtures.py
index b4dce842..d7eb0d20 100644
--- a/kalshi/testing/_fixtures.py
+++ b/kalshi/testing/_fixtures.py
@@ -17,6 +17,7 @@
 
 import json
 import re
+from collections.abc import Callable, Iterable
 from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qsl, urlsplit
@@ -25,6 +26,38 @@
 
 _SANITIZE_RE = re.compile(r"[^A-Za-z0-9._-]+")
 
+# P1.5 default response-header scrubbing: drop credentials and
+# routing/billing identifiers that the live API echoes back. Names matched
+# case-insensitively. Anything advertising a session/account/user/key
+# correlation is filtered so recorded fixtures can be checked into VCS
+# without leaking the recording account's identity.
+_DEFAULT_SCRUB_NAMES = frozenset({"set-cookie", "authorization"})
+_DEFAULT_SCRUB_RE = re.compile(r"(?i)^x-kalshi-.*-(id|key|account|user).*$")
+
+ResponseHeaderFilter = Callable[[str, str], bool] | Iterable[str] | None
+
+
+def _default_response_header_drop(name: str, _value: str) -> bool:
+    """Default predicate: True means drop the header from recorded fixtures."""
+    lower = name.lower()
+    if lower in _DEFAULT_SCRUB_NAMES:
+        return True
+    return bool(_DEFAULT_SCRUB_RE.match(name))
+
+
+def _coerce_header_filter(
+    response_header_filter: ResponseHeaderFilter,
+) -> Callable[[str, str], bool]:
+    """Normalize the user-facing filter parameter into a (name, value) -> drop predicate."""
+    if response_header_filter is None:
+        return _default_response_header_drop
+    if callable(response_header_filter):
+        # Callable: user supplies the drop predicate directly.
+        return response_header_filter
+    # Iterable[str]: case-insensitive deny-list of header names.
+    deny: frozenset[str] = frozenset(name.lower() for name in response_header_filter)
+    return lambda name, _value: name.lower() in deny
+
 
 def fingerprint(request: httpx.Request) -> tuple[str, str, tuple[tuple[str, str], ...]]:
     """Return a (method, path, sorted_query) tuple that identifies a request.
@@ -55,7 +88,11 @@ def _request_to_dict(request: httpx.Request) -> dict[str, Any]:
     }
 
 
-def _response_to_dict(response: httpx.Response) -> dict[str, Any]:
+def _response_to_dict(
+    response: httpx.Response,
+    *,
+    drop_header: Callable[[str, str], bool] | None = None,
+) -> dict[str, Any]:
     # Try to store JSON bodies as objects for readability; fall back to text.
     body_bytes = response.content
     body: Any
@@ -65,19 +102,26 @@ def _response_to_dict(response: httpx.Response) -> dict[str, Any]:
     except (ValueError, UnicodeDecodeError):
         body = body_bytes.decode("latin-1")
         body_kind = "text"
+    predicate = drop_header or _default_response_header_drop
+    headers = [(k, v) for k, v in response.headers.items() if not predicate(k, v)]
     return {
         "status_code": response.status_code,
-        "headers": [(k, v) for k, v in response.headers.items()],
+        "headers": headers,
         "body_kind": body_kind,
         "body": body,
     }
 
 
-def record_pair(request: httpx.Request, response: httpx.Response) -> dict[str, Any]:
+def record_pair(
+    request: httpx.Request,
+    response: httpx.Response,
+    *,
+    drop_header: Callable[[str, str], bool] | None = None,
+) -> dict[str, Any]:
     """Build a serializable pair from a real request/response round-trip."""
     return {
         "request": _request_to_dict(request),
-        "response": _response_to_dict(response),
+        "response": _response_to_dict(response, drop_header=drop_header),
     }
 
 
diff --git a/kalshi/testing/_recorder.py b/kalshi/testing/_recorder.py
index 379d121a..668fa2ba 100644
--- a/kalshi/testing/_recorder.py
+++ b/kalshi/testing/_recorder.py
@@ -8,6 +8,8 @@
 import httpx
 
 from kalshi.testing._fixtures import (
+    ResponseHeaderFilter,
+    _coerce_header_filter,
     fingerprint,
     load_pairs,
     record_pair,
@@ -52,6 +54,14 @@ class RecordingTransport(httpx.BaseTransport):
     `httpx.HTTPTransport`) so real network calls go through, while every
     request/response pair is buffered in memory and flushed to disk on
     ``close()`` — see #105 for the prior O(N²) per-request rewrite this fixes.
+
+    ``response_header_filter`` controls which response headers are persisted.
+    The default predicate drops ``Set-Cookie``, ``Authorization``, and any
+    header matching ``(?i)^x-kalshi-.*-(id|key|account|user).*$`` so common
+    credential/identity leaks don't end up in checked-in fixtures. Supply a
+    callable ``(name, value) -> bool`` (return True to drop) or an iterable
+    of header names to deny — pass ``lambda _n, _v: False`` to keep every
+    header verbatim.
     """
 
     def __init__(
@@ -59,10 +69,12 @@ def __init__(
         dir_path: str | Path,
         *,
         real_transport: httpx.BaseTransport | None = None,
+        response_header_filter: ResponseHeaderFilter = None,
     ) -> None:
         self._dir = Path(dir_path)
         self._real = real_transport if real_transport is not None else httpx.HTTPTransport()
         self._buffer = _RecorderBuffer(self._dir)
+        self._drop_header = _coerce_header_filter(response_header_filter)
 
     def handle_request(self, request: httpx.Request) -> httpx.Response:
         response = self._real.handle_request(request)
@@ -71,7 +83,9 @@ def handle_request(self, request: httpx.Request) -> httpx.Response:
         method, path, _query = fingerprint(request)
         # Append to in-memory buffer; flushed to disk on close().
         # Recordings are expected to run sequentially (see module docstring).
-        self._buffer.append(method, path, record_pair(request, response))
+        self._buffer.append(
+            method, path, record_pair(request, response, drop_header=self._drop_header)
+        )
         return response
 
     def close(self) -> None:
@@ -89,18 +103,22 @@ def __init__(
         dir_path: str | Path,
         *,
         real_transport: httpx.AsyncBaseTransport | None = None,
+        response_header_filter: ResponseHeaderFilter = None,
     ) -> None:
         self._dir = Path(dir_path)
         self._real = (
             real_transport if real_transport is not None else httpx.AsyncHTTPTransport()
         )
         self._buffer = _RecorderBuffer(self._dir)
+        self._drop_header = _coerce_header_filter(response_header_filter)
 
     async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
         response = await self._real.handle_async_request(request)
         await response.aread()
         method, path, _query = fingerprint(request)
-        self._buffer.append(method, path, record_pair(request, response))
+        self._buffer.append(
+            method, path, record_pair(request, response, drop_header=self._drop_header)
+        )
         return response
 
     async def aclose(self) -> None:
diff --git a/pyproject.toml b/pyproject.toml
index a0936b38..73ddfc90 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,7 +26,8 @@ dependencies = [
 [project.optional-dependencies]
 pandas = ["pandas>=2,<4"]
 polars = ["polars>=1,<2"]
-all = ["pandas>=2,<4", "polars>=1,<2"]
+http2 = ["h2>=4,<5"]
+all = ["pandas>=2,<4", "polars>=1,<2", "h2>=4,<5"]
 
 [project.urls]
 Homepage = "https://github.com/TexasCoding/kalshi-python-sdk"
diff --git a/tests/test_async_client.py b/tests/test_async_client.py
index 549ce33f..ec36af62 100644
--- a/tests/test_async_client.py
+++ b/tests/test_async_client.py
@@ -233,10 +233,10 @@ async def fake_sleep(d: float) -> None:
 
     @respx.mock
     @pytest.mark.asyncio
-    async def test_429_retry_after_http_date_falls_back_to_backoff(
+    async def test_429_retry_after_http_date_parsed_and_capped(
         self, transport: AsyncTransport, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        """Async: HTTP-date unparseable; transport retries via computed backoff."""
+        """Async: HTTP-date Retry-After parsed via email.utils, capped at retry_max_delay."""
         sleeps: list[float] = []
 
         async def fake_sleep(d: float) -> None:
@@ -257,8 +257,9 @@ async def fake_sleep(d: float) -> None:
         assert resp.status_code == 200
         assert route.call_count == 2
         assert len(sleeps) == 1
-        assert 0.0 <= sleeps[0] <= 0.01, (
-            f"Expected backoff sleep in [0, 0.01], got {sleeps[0]!r}"
+        # Date is far in the future; delta is huge, capped at retry_max_delay=0.1.
+        assert sleeps[0] == 0.1, (
+            f"Expected sleep clamped to retry_max_delay=0.1, got {sleeps[0]!r}"
         )
 
     @respx.mock
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 40f3d310..9b2d7165 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -482,3 +482,176 @@ def test_returns_none_when_key_id_set_but_no_key(
         monkeypatch.delenv("KALSHI_PRIVATE_KEY_PATH", raising=False)
         auth = KalshiAuth.try_from_env()
         assert auth is None
+
+
+class TestPassphraseSupport:
+    """#217: KalshiAuth loaders accept ``password=`` so callers don't have to
+    write plaintext keys to disk just to use this SDK.
+    """
+
+    @staticmethod
+    def _encrypted_pem(key: rsa.RSAPrivateKey, passphrase: bytes) -> bytes:
+        return key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.BestAvailableEncryption(passphrase),
+        )
+
+    def test_from_pem_with_passphrase_succeeds(
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"correct-horse")
+        auth = KalshiAuth.from_pem("key", encrypted, password="correct-horse")
+        assert auth.key_id == "key"
+        # Sanity-check signing still works post-load.
+        headers = auth.sign_request("GET", "/trade-api/v2/markets", timestamp_ms=1000)
+        assert headers["KALSHI-ACCESS-KEY"] == "key"
+
+    def test_from_pem_with_passphrase_bytes_succeeds(
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"correct-horse")
+        auth = KalshiAuth.from_pem("key", encrypted, password=b"correct-horse")
+        assert auth.key_id == "key"
+
+    def test_from_pem_with_wrong_passphrase_raises_KalshiAuthError(  # noqa: N802
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"correct-horse")
+        with pytest.raises(KalshiAuthError, match=r"Invalid PEM private key"):
+            KalshiAuth.from_pem("key", encrypted, password="wrong")
+
+    def test_from_pem_encrypted_without_password_points_at_password_kwarg(
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"pw")
+        with pytest.raises(KalshiAuthError) as exc_info:
+            KalshiAuth.from_pem("key", encrypted)
+        msg = str(exc_info.value)
+        # New message mentions both the password= kwarg and the legacy openssl recipe.
+        assert "password=" in msg
+        assert "KALSHI_PRIVATE_KEY_PASSPHRASE" in msg
+        assert "openssl pkey" in msg
+
+    def test_from_pem_with_callable_passphrase_invoked(
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"sekrit")
+        calls = {"n": 0}
+
+        def supply_pw() -> bytes:
+            calls["n"] += 1
+            return b"sekrit"
+
+        auth = KalshiAuth.from_pem("k", encrypted, password=supply_pw)
+        assert auth.key_id == "k"
+        assert calls["n"] == 1, "callable passphrase must be invoked exactly once"
+
+    def test_from_key_path_with_passphrase_succeeds(
+        self, rsa_private_key: rsa.RSAPrivateKey
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"pw")
+        with tempfile.NamedTemporaryFile(suffix=".pem", delete=False) as f:
+            f.write(encrypted)
+            f.flush()
+            path = f.name
+        try:
+            auth = KalshiAuth.from_key_path("k", path, password="pw")
+            assert auth.key_id == "k"
+        finally:
+            os.unlink(path)
+
+    def test_from_env_picks_up_KALSHI_PRIVATE_KEY_PASSPHRASE(  # noqa: N802
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        rsa_private_key: rsa.RSAPrivateKey,
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"envpw")
+        monkeypatch.setenv("KALSHI_KEY_ID", "env-encrypted")
+        monkeypatch.setenv("KALSHI_PRIVATE_KEY", encrypted.decode("utf-8"))
+        monkeypatch.delenv("KALSHI_PRIVATE_KEY_PATH", raising=False)
+        monkeypatch.setenv("KALSHI_PRIVATE_KEY_PASSPHRASE", "envpw")
+        auth = KalshiAuth.from_env()
+        assert auth.key_id == "env-encrypted"
+
+    def test_from_env_explicit_password_beats_env_passphrase(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        rsa_private_key: rsa.RSAPrivateKey,
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"realpw")
+        monkeypatch.setenv("KALSHI_KEY_ID", "env-encrypted")
+        monkeypatch.setenv("KALSHI_PRIVATE_KEY", encrypted.decode("utf-8"))
+        monkeypatch.delenv("KALSHI_PRIVATE_KEY_PATH", raising=False)
+        monkeypatch.setenv("KALSHI_PRIVATE_KEY_PASSPHRASE", "wrong-pw-in-env")
+        # Explicit kwarg overrides the env var: load must succeed with realpw.
+        auth = KalshiAuth.from_env(password="realpw")
+        assert auth.key_id == "env-encrypted"
+
+    def test_try_from_env_picks_up_KALSHI_PRIVATE_KEY_PASSPHRASE(  # noqa: N802
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        rsa_private_key: rsa.RSAPrivateKey,
+    ) -> None:
+        encrypted = self._encrypted_pem(rsa_private_key, b"envpw")
+        monkeypatch.setenv("KALSHI_KEY_ID", "env-encrypted")
+        monkeypatch.setenv("KALSHI_PRIVATE_KEY", encrypted.decode("utf-8"))
+        monkeypatch.delenv("KALSHI_PRIVATE_KEY_PATH", raising=False)
+        monkeypatch.setenv("KALSHI_PRIVATE_KEY_PASSPHRASE", "envpw")
+        auth = KalshiAuth.try_from_env()
+        assert auth is not None
+        assert auth.key_id == "env-encrypted"
+
+
+class TestTrailingSlashCanonicalization:
+    """P1.6: the transport canonicalizes trailing slashes BEFORE both signing
+    and the httpx call, so a future ``/markets/`` couldn't desync wire-path and
+    signed-path.
+    """
+
+    def test_trailing_slash_path_canonicalized(
+        self,
+        rsa_private_key: rsa.RSAPrivateKey,
+        test_auth: KalshiAuth,
+    ) -> None:
+        import httpx
+        import respx
+
+        from kalshi._base_client import SyncTransport
+        from kalshi.config import KalshiConfig
+
+        config = KalshiConfig(base_url="https://test.kalshi.com/trade-api/v2", timeout=5.0)
+        transport = SyncTransport(test_auth, config)
+        captured: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured.append(request)
+            return httpx.Response(200, json={})
+
+        with respx.mock:
+            # Match the canonicalized (no-trailing-slash) URL — if the transport
+            # forwarded "/markets/" to httpx, respx would 404 this request.
+            respx.get("https://test.kalshi.com/trade-api/v2/markets").mock(
+                side_effect=handler
+            )
+            resp = transport.request("GET", "/markets/")
+        assert resp.status_code == 200
+        sent = captured[0]
+        # Wire URL was canonicalized to the no-slash form.
+        assert sent.url.path == "/trade-api/v2/markets"
+        # Signed path agrees with the wire path: re-derive the expected message
+        # from the timestamp header and check the signature verifies against
+        # the canonical (no-slash) form.
+        ts = sent.headers["KALSHI-ACCESS-TIMESTAMP"]
+        sig = base64.b64decode(sent.headers["KALSHI-ACCESS-SIGNATURE"])
+        canonical = f"{ts}GET/trade-api/v2/markets".encode()
+        rsa_private_key.public_key().verify(
+            sig,
+            canonical,
+            padding.PSS(
+                mgf=padding.MGF1(hashes.SHA256()),
+                salt_length=padding.PSS.DIGEST_LENGTH,
+            ),
+            hashes.SHA256(),
+        )
+        transport.close()
diff --git a/tests/test_client.py b/tests/test_client.py
index d0f7cd31..a9704b32 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -7,6 +7,7 @@
 
 import os
 import tempfile
+from datetime import UTC
 
 import httpx
 import pytest
@@ -146,12 +147,38 @@ def test_429_retry_after_zero_is_kept(self) -> None:
         assert isinstance(err, KalshiRateLimitError)
         assert err.retry_after == 0.0
 
-    def test_429_retry_after_http_date_falls_back(self) -> None:
-        """HTTP-date format isn't parsed; should fall through to backoff."""
+    def test_429_retry_after_http_date_parsed(self) -> None:
+        """RFC 7231: Retry-After may be an HTTP-date; parsed into delta-seconds."""
+        # 5 days in the future, formatted as RFC 1123.
+        import email.utils as _eu
+        from datetime import datetime, timedelta
+
+        future = datetime.now(tz=UTC) + timedelta(days=5)
+        http_date = _eu.format_datetime(future, usegmt=True)
+        resp = httpx.Response(
+            429, json={"message": "slow"}, headers={"Retry-After": http_date}
+        )
+        err = _map_error(resp)
+        assert isinstance(err, KalshiRateLimitError)
+        # ~5 days ≈ 432000s; allow generous slack for the parse round-trip.
+        assert err.retry_after is not None
+        assert 4 * 86400 < err.retry_after < 6 * 86400
+
+    def test_429_retry_after_past_http_date_clamped_to_zero(self) -> None:
+        """A Retry-After date already in the past clamps to 0, not negative."""
         resp = httpx.Response(
             429,
             json={"message": "slow"},
-            headers={"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"},
+            headers={"Retry-After": "Wed, 21 Oct 1999 07:28:00 GMT"},
+        )
+        err = _map_error(resp)
+        assert isinstance(err, KalshiRateLimitError)
+        assert err.retry_after == 0.0
+
+    def test_429_retry_after_garbage_falls_back(self) -> None:
+        """Neither delta-seconds nor a parseable HTTP-date — None, computed backoff."""
+        resp = httpx.Response(
+            429, json={"message": "slow"}, headers={"Retry-After": "not a date"}
         )
         err = _map_error(resp)
         assert isinstance(err, KalshiRateLimitError)
@@ -301,17 +328,25 @@ def test_429_retry_after_caps_at_retry_max_delay(
         )
 
     @respx.mock
-    def test_429_retry_after_http_date_falls_back_to_backoff(
+    def test_retry_after_http_date_parsed_and_capped(
         self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch
     ) -> None:
-        """HTTP-date Retry-After unparseable; transport still retries via backoff."""
+        """HTTP-date Retry-After is parsed into delta-seconds and capped at retry_max_delay."""
+        import email.utils as _eu
+        from datetime import datetime, timedelta
+
         sleeps: list[float] = []
         monkeypatch.setattr("kalshi._base_client.time.sleep", lambda d: sleeps.append(d))
+        # 30 days from now: parsed delta is much larger than retry_max_delay=0.1s,
+        # so the transport must cap. This documents that the date branch flows
+        # through the same min(delta, retry_max_delay) path as numeric input.
+        future = datetime.now(tz=UTC) + timedelta(days=30)
+        http_date = _eu.format_datetime(future, usegmt=True)
         route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock(
             side_effect=[
                 httpx.Response(
                     429,
-                    headers={"Retry-After": "Wed, 21 Oct 2026 07:28:00 GMT"},
+                    headers={"Retry-After": http_date},
                     json={"message": "rl"},
                 ),
                 httpx.Response(200, json={"markets": []}),
@@ -320,12 +355,34 @@ def test_429_retry_after_http_date_falls_back_to_backoff(
         resp = transport.request("GET", "/markets")
         assert resp.status_code == 200
         assert route.call_count == 2
-        # Backoff is Full Jitter: uniform(0, min(base*2**0, max)) = uniform(0, 0.01).
-        assert len(sleeps) == 1
-        assert 0.0 <= sleeps[0] <= 0.01, (
-            f"Expected backoff sleep in [0, 0.01], got {sleeps[0]!r}"
+        assert sleeps == [0.1], (
+            f"Expected one sleep of 0.1s (capped from parsed HTTP-date), got {sleeps!r}"
         )
 
+    @respx.mock
+    def test_429_retry_after_unparseable_falls_back_to_backoff(
+        self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Retry-After that is neither delta-seconds nor an HTTP-date uses computed backoff."""
+        sleeps: list[float] = []
+        monkeypatch.setattr("kalshi._base_client.time.sleep", lambda d: sleeps.append(d))
+        route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock(
+            side_effect=[
+                httpx.Response(
+                    429,
+                    headers={"Retry-After": "totally bogus"},
+                    json={"message": "rl"},
+                ),
+                httpx.Response(200, json={"markets": []}),
+            ]
+        )
+        resp = transport.request("GET", "/markets")
+        assert resp.status_code == 200
+        assert route.call_count == 2
+        # Full Jitter backoff: uniform(0, min(0.01, 0.1)) = uniform(0, 0.01).
+        assert len(sleeps) == 1
+        assert 0.0 <= sleeps[0] <= 0.01
+
     @respx.mock
     def test_get_retries_on_timeout(
         self, transport: SyncTransport, monkeypatch: pytest.MonkeyPatch
@@ -1172,3 +1229,81 @@ def test_from_env_no_credentials_owns_nothing(
         # No auth → ownership flag is moot but should be False (nothing to own).
         assert client._auth_owned is False
         client.close()
+
+
+class TestExtraHeadersPerRequest:
+    """P1.2: SyncTransport.request accepts per-call ``extra_headers`` that
+    layer on top of config-level ``extra_headers`` but never override the
+    signed ``KALSHI-ACCESS-*`` headers.
+    """
+
+    @respx.mock
+    def test_extra_headers_per_request_overrides_config(
+        self, test_auth: KalshiAuth
+    ) -> None:
+        config = KalshiConfig(
+            base_url="https://test.kalshi.com/trade-api/v2",
+            timeout=5.0,
+            extra_headers={"X-Request-ID": "config-default", "X-App": "shared"},
+        )
+        transport = SyncTransport(test_auth, config)
+        captured: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured.append(request)
+            return httpx.Response(200, json={})
+
+        route = respx.get("https://test.kalshi.com/trade-api/v2/markets").mock(
+            side_effect=handler
+        )
+        resp = transport.request(
+            "GET",
+            "/markets",
+            extra_headers={"X-Request-ID": "per-call-override"},
+        )
+        assert resp.status_code == 200
+        assert route.called
+        sent = captured[0]
+        # Per-call beat config default.
+        assert sent.headers["X-Request-ID"] == "per-call-override"
+        # Config default still present where per-call didn't override.
+        assert sent.headers["X-App"] == "shared"
+        transport.close()
+
+    @respx.mock
+    def test_extra_headers_cannot_forge_KALSHI_ACCESS_headers(  # noqa: N802
+        self, test_auth: KalshiAuth
+    ) -> None:
+        """Per-call extra_headers MUST NOT override the SDK's signed auth headers."""
+        config = KalshiConfig(
+            base_url="https://test.kalshi.com/trade-api/v2",
+            timeout=5.0,
+        )
+        transport = SyncTransport(test_auth, config)
+        captured: list[httpx.Request] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured.append(request)
+            return httpx.Response(200, json={})
+
+        respx.get("https://test.kalshi.com/trade-api/v2/markets").mock(side_effect=handler)
+        resp = transport.request(
+            "GET",
+            "/markets",
+            extra_headers={
+                "KALSHI-ACCESS-KEY": "attacker-key",
+                "KALSHI-ACCESS-SIGNATURE": "AAAAAA==",
+                "KALSHI-ACCESS-TIMESTAMP": "1",
+                "X-Trace": "ok",
+            },
+        )
+        assert resp.status_code == 200
+        sent = captured[0]
+        # The signed key (from test_auth fixture) wins, not the forged one.
+        assert sent.headers["KALSHI-ACCESS-KEY"] == "test-key-id"
+        # Signature/timestamp were re-signed by the SDK, not "AAAAAA=="/"1".
+        assert sent.headers["KALSHI-ACCESS-SIGNATURE"] != "AAAAAA=="
+        assert sent.headers["KALSHI-ACCESS-TIMESTAMP"] != "1"
+        # Non-auth headers still pass through.
+        assert sent.headers["X-Trace"] == "ok"
+        transport.close()
diff --git a/tests/test_config.py b/tests/test_config.py
index b52a87c1..20fe4abe 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import httpx
+import pytest
 import respx
 
 from kalshi._base_client import SyncTransport
@@ -261,3 +262,59 @@ def test_ws_json_loads_callable_stored(self) -> None:
         cfg = KalshiConfig(ws_json_loads=json.loads, ws_json_dumps=json.dumps)
         assert cfg.ws_json_loads is json.loads
         assert cfg.ws_json_dumps is json.dumps
+class TestHttp2ImportCheck:
+    """P1.1: ``http2=True`` must fail-fast at construction if ``h2`` is missing,
+    instead of deferring an opaque ImportError to the first request.
+    """
+
+    def test_http2_True_with_h2_missing_raises_at_construction(  # noqa: N802
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        import importlib.util as _ilu
+
+        real_find_spec = _ilu.find_spec
+
+        def fake_find_spec(name: str, *args: object, **kwargs: object) -> object:
+            if name == "h2":
+                return None
+            return real_find_spec(name, *args, **kwargs)  # type: ignore[arg-type]
+
+        monkeypatch.setattr(_ilu, "find_spec", fake_find_spec)
+        with pytest.raises(ValueError, match=r"http2=True requires the 'h2' package"):
+            KalshiConfig(http2=True)
+
+    def test_http2_True_with_h2_installed_succeeds(  # noqa: N802
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        import importlib.machinery
+        import importlib.util as _ilu
+
+        real_find_spec = _ilu.find_spec
+        sentinel = importlib.machinery.ModuleSpec("h2", loader=None)
+
+        def fake_find_spec(name: str, *args: object, **kwargs: object) -> object:
+            if name == "h2":
+                return sentinel
+            return real_find_spec(name, *args, **kwargs)  # type: ignore[arg-type]
+
+        monkeypatch.setattr(_ilu, "find_spec", fake_find_spec)
+        # Should NOT raise.
+        config = KalshiConfig(http2=True)
+        assert config.http2 is True
+
+    def test_http2_False_does_not_check_h2(  # noqa: N802
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """Default http2=False must never touch find_spec — no startup cost."""
+        called: list[str] = []
+        import importlib.util as _ilu
+
+        real_find_spec = _ilu.find_spec
+
+        def tracking_find_spec(name: str, *args: object, **kwargs: object) -> object:
+            called.append(name)
+            return real_find_spec(name, *args, **kwargs)  # type: ignore[arg-type]
+
+        monkeypatch.setattr(_ilu, "find_spec", tracking_find_spec)
+        KalshiConfig()
+        assert "h2" not in called
diff --git a/tests/test_mock_transport.py b/tests/test_mock_transport.py
index 9805c3b3..94e27971 100644
--- a/tests/test_mock_transport.py
+++ b/tests/test_mock_transport.py
@@ -539,3 +539,122 @@ def test_record_non_json_binary_body_persists_as_text(tmp_path: Path) -> None:
     replay = ReplayTransport(fixtures_dir)
     with KalshiClient(transport=replay, max_retries=0) as client, pytest.raises(KalshiServerError):
         client.exchange.status()
+
+
+class TestResponseHeaderFilter:
+    """P1.5: RecordingTransport response_header_filter parameter."""
+
+    def test_default_filter_drops_set_cookie(self, tmp_path: Path) -> None:
+        """Default filter scrubs Set-Cookie / Authorization / X-Kalshi-* identity headers."""
+        stub = _make_stub_transport(
+            {
+                ("GET", "/trade-api/v2/markets"): httpx.Response(
+                    200,
+                    headers=[
+                        ("Content-Type", "application/json"),
+                        ("Set-Cookie", "session=abcd; HttpOnly"),
+                        ("Authorization", "Bearer should-not-be-recorded"),
+                        ("X-Kalshi-User-Id", "uid-42"),
+                        ("X-Kalshi-Trace-Account-Hash", "acc-99"),
+                        ("X-Kalshi-Trace-Latency-Ms", "12"),  # NOT scrubbed by default
+                        ("X-RateLimit-Remaining", "499"),  # NOT scrubbed by default
+                    ],
+                    content=b'{"markets": []}',
+                )
+            }
+        )
+        rec = RecordingTransport(tmp_path, real_transport=stub)
+        try:
+            with KalshiClient(transport=rec) as client:
+                client.markets.list()
+        finally:
+            rec.close()
+        # Read back the persisted fixture and check header set.
+        fixture_file = tmp_path / fixture_filename("GET", "/trade-api/v2/markets")
+        data = json.loads(fixture_file.read_text())
+        # httpx normalises response header names to lowercase.
+        headers = {k.lower(): v for k, v in data[0]["response"]["headers"]}
+        # Defaults dropped (matched case-insensitively by the predicate):
+        assert "set-cookie" not in headers
+        assert "authorization" not in headers
+        assert "x-kalshi-user-id" not in headers
+        assert "x-kalshi-trace-account-hash" not in headers
+        # Defaults preserved:
+        assert headers.get("content-type") == "application/json"
+        assert headers.get("x-ratelimit-remaining") == "499"
+        # X-Kalshi-Request-Id also matches the default regex (the *-id suffix),
+        # so request IDs are scrubbed too. Latency-Ms doesn't match any suffix
+        # token and is preserved.
+        assert headers.get("x-kalshi-request-id") is None
+        assert headers.get("x-kalshi-trace-latency-ms") == "12"
+
+    def test_custom_filter_callable_invoked(self, tmp_path: Path) -> None:
+        """User-supplied predicate sees (name, value) and decides drop/keep."""
+        calls: list[tuple[str, str]] = []
+
+        def drop_secret_headers(name: str, value: str) -> bool:
+            calls.append((name, value))
+            return name.lower() == "x-secret"
+
+        stub = _make_stub_transport(
+            {
+                ("GET", "/trade-api/v2/markets"): httpx.Response(
+                    200,
+                    headers=[
+                        ("Content-Type", "application/json"),
+                        ("X-Secret", "topsecret"),
+                        ("Set-Cookie", "should-be-kept-when-filter-is-overridden"),
+                    ],
+                    content=b'{"markets": []}',
+                )
+            }
+        )
+        rec = RecordingTransport(
+            tmp_path, real_transport=stub, response_header_filter=drop_secret_headers
+        )
+        try:
+            with KalshiClient(transport=rec) as client:
+                client.markets.list()
+        finally:
+            rec.close()
+        # Predicate was called for every response header.
+        # Predicate was called for every response header (httpx lowercases).
+        seen_names = {name.lower() for name, _ in calls}
+        assert "x-secret" in seen_names
+        assert "set-cookie" in seen_names
+        fixture_file = tmp_path / fixture_filename("GET", "/trade-api/v2/markets")
+        data = json.loads(fixture_file.read_text())
+        headers = {k.lower(): v for k, v in data[0]["response"]["headers"]}
+        # Custom filter dropped X-Secret …
+        assert "x-secret" not in headers
+        # … and the default Set-Cookie scrubbing did NOT apply (user took over).
+        assert headers.get("set-cookie") == "should-be-kept-when-filter-is-overridden"
+
+    def test_iterable_deny_list_drops_matching_headers(self, tmp_path: Path) -> None:
+        """Iterable[str] denylist is matched case-insensitively."""
+        stub = _make_stub_transport(
+            {
+                ("GET", "/trade-api/v2/markets"): httpx.Response(
+                    200,
+                    headers=[
+                        ("Content-Type", "application/json"),
+                        ("X-Trace-ID", "abc"),
+                        ("X-Keep-Me", "yes"),
+                    ],
+                    content=b'{"markets": []}',
+                )
+            }
+        )
+        rec = RecordingTransport(
+            tmp_path, real_transport=stub, response_header_filter=["x-trace-id"]
+        )
+        try:
+            with KalshiClient(transport=rec) as client:
+                client.markets.list()
+        finally:
+            rec.close()
+        fixture_file = tmp_path / fixture_filename("GET", "/trade-api/v2/markets")
+        data = json.loads(fixture_file.read_text())
+        headers = {k.lower(): v for k, v in data[0]["response"]["headers"]}
+        assert "x-trace-id" not in headers
+        assert headers.get("x-keep-me") == "yes"