feat(auth+config): defense-in-depth (#249, #266, #261, #250)

docs/environment-variables.md

@@ -6,15 +6,18 @@ these:
 | Variable | Default | Effect |
 |---|---|---|
 | `KALSHI_KEY_ID` | unset | Key ID. If unset, `from_env()` returns an unauthenticated client. |
-| `KALSHI_PRIVATE_KEY` | unset | PEM string. Takes precedence over `KALSHI_PRIVATE_KEY_PATH`. |
-| `KALSHI_PRIVATE_KEY_PATH` | unset | Path to PEM file. `~` is expanded. |
+| `KALSHI_PRIVATE_KEY` | unset | PEM string. **Conflicts** with `KALSHI_PRIVATE_KEY_PATH` — set exactly one. |
+| `KALSHI_PRIVATE_KEY_PATH` | unset | Path to PEM file. `~` is expanded. Conflicts with `KALSHI_PRIVATE_KEY`. |
 | `KALSHI_DEMO` | `false` | Truthy values (`true`, `1`, `yes`, `on`, case-insensitive) flip the client to the demo URLs. |
 | `KALSHI_API_BASE_URL` | unset | Overrides `base_url` entirely. Wins over `KALSHI_DEMO`. |
+| `KALSHI_ALLOW_UNKNOWN_HOST` | unset | Set to `1` to allow `base_url`/`ws_base_url` hosts outside `{api.elections.kalshi.com, demo-api.kalshi.co, localhost}`. Default-fail prevents typos like `kalsi.com` from silently signing to an attacker (#250). |
 
 ## Precedence
 
-1. **Credentials**: `KALSHI_PRIVATE_KEY` (in-memory PEM) beats
-   `KALSHI_PRIVATE_KEY_PATH` (file).
+1. **Credentials**: setting both `KALSHI_PRIVATE_KEY` (in-memory PEM) and
+   `KALSHI_PRIVATE_KEY_PATH` raises `KalshiAuthError`. Previously the inline
+   PEM silently won; under #249 the ambiguity is rejected up-front so
+   key-rotation mishaps surface as a clean error instead of a 401/403 mystery.
 2. **Base URL**: an explicit `KALSHI_API_BASE_URL` wins over the demo/production
    pair selected by `KALSHI_DEMO`.
 3. **Constructor overrides**: anything passed positionally / by kwarg to

kalshi/async_client.py

@@ -5,7 +5,7 @@
 import os
 from pathlib import Path
 from types import TracebackType
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypedDict, Unpack
 
 import httpx
 
@@ -33,6 +33,27 @@
 from kalshi.resources.structured_targets import AsyncStructuredTargetsResource
 from kalshi.resources.subaccounts import AsyncSubaccountsResource
 
+
+class ClientInitKwargs(TypedDict, total=False):
+    """Typed forwarder shape for :meth:`AsyncKalshiClient.from_env`.
+
+    Mirrors the ``__init__`` keyword surface so ``mypy --strict`` catches typos
+    like ``from_env(time_out=10)`` instead of silently swallowing them under
+    ``**kwargs: object``. See #266.
+    """
+
+    key_id: str | None
+    private_key: str | bytes | None
+    private_key_path: str | Path | None
+    auth: KalshiAuth | None
+    config: KalshiConfig | None
+    demo: bool
+    base_url: str | None
+    timeout: float | None
+    max_retries: int | None
+    transport: httpx.AsyncBaseTransport | None
+
+
 if TYPE_CHECKING:
     from kalshi.ws.client import KalshiWebSocket
 
@@ -65,6 +86,11 @@ def __init__(
         # Reject empty strings that look like misconfigured credentials
         if key_id is not None and not key_id.strip():
             raise ValueError("key_id must not be empty. Omit it for unauthenticated access.")
+        # #249: refuse ambiguous credential bundle — caller must pick exactly one
+        # private-key source so a key-rotation mishap can't leave the SDK
+        # silently signing with the wrong key.
+        if private_key_path is not None and private_key is not None:
+            raise ValueError("Provide either private_key_path or private_key, not both.")
         # #210: only shut down auth on close() when WE built it. If the
         # caller passed in their own KalshiAuth, they keep ownership.
         self._auth_owned: bool = auth is None
@@ -166,10 +192,11 @@ async def handle(msg): ...
                 "Provide key_id + private_key_path, or use AsyncKalshiClient.from_env()."
             )
         from kalshi.ws.client import KalshiWebSocket as _KalshiWebSocket
+
         return _KalshiWebSocket(auth=self._auth, config=self._config)
 
     @classmethod
-    def from_env(cls, **kwargs: object) -> AsyncKalshiClient:
+    def from_env(cls, **kwargs: Unpack[ClientInitKwargs]) -> AsyncKalshiClient:
         """Create async client from environment variables.
 
         Reads:
@@ -180,12 +207,16 @@ def from_env(cls, **kwargs: object) -> AsyncKalshiClient:
 
         Returns an unauthenticated client if no credentials are configured.
         """
-        auth = KalshiAuth.try_from_env()
-        demo = os.environ.get("KALSHI_DEMO", "").lower() == "true"
-        base_url = os.environ.get("KALSHI_API_BASE_URL")
-        client = cls(auth=auth, demo=demo, base_url=base_url, **kwargs)  # type: ignore[arg-type]
+        # Caller-supplied kwargs win over env-derived values (legacy behaviour
+        # via ``cls(auth=..., demo=..., base_url=..., **kwargs)`` would have
+        # TypeError'd on duplicates; setdefault preserves env semantics while
+        # giving the static signature a clean shape). See #266.
+        kwargs.setdefault("auth", KalshiAuth.try_from_env())
+        kwargs.setdefault("demo", os.environ.get("KALSHI_DEMO", "").lower() == "true")
+        kwargs.setdefault("base_url", os.environ.get("KALSHI_API_BASE_URL"))
+        client = cls(**kwargs)
         # from_env constructs auth locally — caller never owned it.
-        client._auth_owned = auth is not None
+        client._auth_owned = kwargs["auth"] is not None
         return client
 
     async def close(self) -> None:

kalshi/auth.py

@@ -44,6 +44,10 @@ def _normalize_percent_encoding(path: str) -> str:
         /markets/ABC%2fDEF -> /markets/ABC%2FDEF
         /markets/ABC%2FDEF -> /markets/ABC%2FDEF (no change)
     """
+    # #261: vast majority of signed paths contain no percent-encoding (e.g.
+    # /trade-api/v2/markets); skip the regex compile + scan in the hot path.
+    if "%" not in path:
+        return path
     return re.sub(
         r"%([0-9a-fA-F]{2})",
         lambda m: "%" + m.group(1).upper(),
@@ -163,19 +167,25 @@ def from_env(
         key_id = os.environ.get("KALSHI_KEY_ID")
         if not key_id:
             raise KalshiAuthError(
-                "KALSHI_KEY_ID environment variable is not set. "
-                "Set it to your Kalshi API key ID."
+                "KALSHI_KEY_ID environment variable is not set. 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")
+        key_path = os.environ.get("KALSHI_PRIVATE_KEY_PATH")
+        # #249: refuse ambiguous credential bundle — naming both env vars makes
+        # the misconfiguration obvious in production logs.
+        if pem_string and key_path:
+            raise KalshiAuthError(
+                "Both KALSHI_PRIVATE_KEY and KALSHI_PRIVATE_KEY_PATH are set. "
+                "Provide exactly one — unset whichever you don't intend to use."
+            )
         if 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, password=resolved_password)
 
@@ -211,10 +221,17 @@ def try_from_env(
         )
 
         pem_string = os.environ.get("KALSHI_PRIVATE_KEY")
+        key_path = os.environ.get("KALSHI_PRIVATE_KEY_PATH")
+        # #249: refuse ambiguous credential bundle — naming both env vars makes
+        # the misconfiguration obvious in production logs.
+        if pem_string and key_path:
+            raise KalshiAuthError(
+                "Both KALSHI_PRIVATE_KEY and KALSHI_PRIVATE_KEY_PATH are set. "
+                "Provide exactly one — unset whichever you don't intend to use."
+            )
         if 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, password=resolved_password)
 
@@ -289,8 +306,7 @@ def _get_sign_executor(self) -> ThreadPoolExecutor:
         """
         if self._closed:
             raise RuntimeError(
-                "KalshiAuth has been closed; create a new instance to "
-                "sign further requests."
+                "KalshiAuth has been closed; create a new instance to sign further requests."
             )
         if self._sign_executor is None:
             with self._sign_executor_lock:

kalshi/client.py

@@ -5,6 +5,7 @@
 import os
 from pathlib import Path
 from types import TracebackType
+from typing import TypedDict, Unpack
 
 import httpx
 
@@ -32,6 +33,26 @@
 from kalshi.resources.subaccounts import SubaccountsResource
 
 
+class ClientInitKwargs(TypedDict, total=False):
+    """Typed forwarder shape for :meth:`KalshiClient.from_env`.
+
+    Mirrors the ``__init__`` keyword surface so ``mypy --strict`` catches typos
+    like ``from_env(time_out=10)`` instead of silently swallowing them under
+    ``**kwargs: object``. See #266.
+    """
+
+    key_id: str | None
+    private_key: str | bytes | None
+    private_key_path: str | Path | None
+    auth: KalshiAuth | None
+    config: KalshiConfig | None
+    demo: bool
+    base_url: str | None
+    timeout: float | None
+    max_retries: int | None
+    transport: httpx.BaseTransport | None
+
+
 class KalshiClient:
     """Synchronous client for the Kalshi prediction markets API.
 
@@ -67,6 +88,11 @@ def __init__(
         # Reject empty strings that look like misconfigured credentials
         if key_id is not None and not key_id.strip():
             raise ValueError("key_id must not be empty. Omit it for unauthenticated access.")
+        # #249: refuse ambiguous credential bundle — caller must pick exactly one
+        # private-key source so a key-rotation mishap can't leave the SDK
+        # silently signing with the wrong key.
+        if private_key_path is not None and private_key is not None:
+            raise ValueError("Provide either private_key_path or private_key, not both.")
         # #210: only shut down auth on close() when WE built it. If the
         # caller passed in their own KalshiAuth, they keep ownership.
         self._auth_owned: bool = auth is None
@@ -138,7 +164,7 @@ def is_authenticated(self) -> bool:
         return self._auth is not None
 
     @classmethod
-    def from_env(cls, **kwargs: object) -> KalshiClient:
+    def from_env(cls, **kwargs: Unpack[ClientInitKwargs]) -> KalshiClient:
         """Create client from environment variables.
 
         Reads:
@@ -149,12 +175,16 @@ def from_env(cls, **kwargs: object) -> KalshiClient:
 
         Returns an unauthenticated client if no credentials are configured.
         """
-        auth = KalshiAuth.try_from_env()
-        demo = os.environ.get("KALSHI_DEMO", "").lower() == "true"
-        base_url = os.environ.get("KALSHI_API_BASE_URL")
-        client = cls(auth=auth, demo=demo, base_url=base_url, **kwargs)  # type: ignore[arg-type]
+        # Caller-supplied kwargs win over env-derived values (legacy behaviour
+        # via ``cls(auth=..., demo=..., base_url=..., **kwargs)`` would have
+        # TypeError'd on duplicates; setdefault preserves env semantics while
+        # giving the static signature a clean shape). See #266.
+        kwargs.setdefault("auth", KalshiAuth.try_from_env())
+        kwargs.setdefault("demo", os.environ.get("KALSHI_DEMO", "").lower() == "true")
+        kwargs.setdefault("base_url", os.environ.get("KALSHI_API_BASE_URL"))
+        client = cls(**kwargs)
         # from_env constructs auth locally — caller never owned it.
-        client._auth_owned = auth is not None
+        client._auth_owned = kwargs["auth"] is not None
         return client
 
     def close(self) -> None:

kalshi/config.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import os
 from collections.abc import Callable
 from dataclasses import dataclass, field
 from typing import Any
@@ -20,9 +21,7 @@
 DEFAULT_MAX_RETRIES = 3
 DEFAULT_WS_MAX_RETRIES = 10
 
-_KNOWN_HOSTS = frozenset(
-    {"api.elections.kalshi.com", "demo-api.kalshi.co"}
-)
+_KNOWN_HOSTS = frozenset({"api.elections.kalshi.com", "demo-api.kalshi.co"})
 _LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
 
 logger = logging.getLogger("kalshi")
@@ -71,15 +70,34 @@ class KalshiConfig:
     ws_close_timeout: float = 5.0
     ws_json_loads: Callable[[bytes | str], Any] | None = None
     ws_json_dumps: Callable[[Any], bytes | str] | None = None
+    allow_unknown_host: bool = False
 
     def __post_init__(self) -> None:
         # Strip trailing slash to prevent double-slash in auth signing paths
         if self.base_url.endswith("/"):
             object.__setattr__(self, "base_url", self.base_url.rstrip("/"))
         if self.ws_base_url.endswith("/"):
             object.__setattr__(self, "ws_base_url", self.ws_base_url.rstrip("/"))
-        KalshiConfig._validate_url(self.base_url, "base_url", secure="https", plaintext="http")
-        KalshiConfig._validate_url(self.ws_base_url, "ws_base_url", secure="wss", plaintext="ws")
+        # #250: ``KALSHI_ALLOW_UNKNOWN_HOST=1`` is a process-wide escape hatch
+        # for mock servers and proxies — it relaxes the host check the same
+        # way the per-instance ``allow_unknown_host=True`` flag does.
+        allow_unknown = self.allow_unknown_host or (
+            os.environ.get("KALSHI_ALLOW_UNKNOWN_HOST", "").strip() == "1"
+        )
+        KalshiConfig._validate_url(
+            self.base_url,
+            "base_url",
+            secure="https",
+            plaintext="http",
+            allow_unknown_host=allow_unknown,
+        )
+        KalshiConfig._validate_url(
+            self.ws_base_url,
+            "ws_base_url",
+            secure="wss",
+            plaintext="ws",
+            allow_unknown_host=allow_unknown,
+        )
         # #202: prevent silently calling /trade-api or /v1 by enforcing the
         # v2 path component. Trailing slashes are already stripped above.
         base_path = urlparse(self.base_url).path
@@ -119,7 +137,14 @@ def __post_init__(self) -> None:
                 )
 
     @staticmethod
-    def _validate_url(url: str, field_name: str, *, secure: str, plaintext: str) -> None:
+    def _validate_url(
+        url: str,
+        field_name: str,
+        *,
+        secure: str,
+        plaintext: str,
+        allow_unknown_host: bool,
+    ) -> None:
         """Reject URLs that would expose credentials (bad scheme or plaintext-to-remote)."""
         parsed = urlparse(url)
         scheme = parsed.scheme.lower()
@@ -131,9 +156,7 @@ def _validate_url(url: str, field_name: str, *, secure: str, plaintext: str) ->
                 f"got scheme={scheme!r} (url={url!r})"
             )
         if not host:
-            raise ValueError(
-                f"KalshiConfig.{field_name} is missing a host: {url!r}"
-            )
+            raise ValueError(f"KalshiConfig.{field_name} is missing a host: {url!r}")
         if scheme == plaintext and host not in _LOCAL_HOSTS:
             raise ValueError(
                 f"KalshiConfig.{field_name} must use {secure}:// for non-loopback "
@@ -142,10 +165,28 @@ def _validate_url(url: str, field_name: str, *, secure: str, plaintext: str) ->
                 f"expose the KALSHI-ACCESS-KEY header and request signature."
             )
         if host not in _KNOWN_HOSTS and host not in _LOCAL_HOSTS:
+            # #250: a typo or hostile env value (e.g. KALSHI_API_BASE_URL=
+            # https://api.elections.kalsi.com/trade-api/v2 — note the missing
+            # 'h') used to slip through with only a logger.warning. Production
+            # log filters silently drop the warning while the SDK keeps signing
+            # every request — including the KALSHI-ACCESS-KEY and matching
+            # RSA-PSS signature — to the attacker. Default-fail; mock servers
+            # and proxies opt in via ``allow_unknown_host=True`` or the
+            # process-wide ``KALSHI_ALLOW_UNKNOWN_HOST=1`` env var.
+            if not allow_unknown_host:
+                raise ValueError(
+                    f"KalshiConfig.{field_name} host {host!r} is not a known "
+                    f"Kalshi endpoint. Known hosts: {sorted(_KNOWN_HOSTS)}. "
+                    "If this is an intentional mock server, proxy, or "
+                    "alternate Kalshi region, opt in with "
+                    "KalshiConfig(allow_unknown_host=True) or set the "
+                    "environment variable KALSHI_ALLOW_UNKNOWN_HOST=1."
+                )
             logger.warning(
-                "KalshiConfig.%s host %r is not a known Kalshi "
-                "endpoint (%s). Requests will be signed and sent there "
-                "with your API key — verify this is intentional.",
+                "KalshiConfig.%s host %r is not a known Kalshi endpoint (%s). "
+                "Requests will be signed and sent there with your API key — "
+                "verify this is intentional. (allow_unknown_host=True or "
+                "KALSHI_ALLOW_UNKNOWN_HOST=1 silenced the default error.)",
                 field_name,
                 host,
                 sorted(_KNOWN_HOSTS),