Skip to content

perps: config/init hardening — host-allowlist split, ws_base_url + password kwargs, config-vs-shorthand guard (#406, #412, #410) #416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TexasCoding merged 1 commit into from
Jun 5, 2026
Merged

perps: config/init hardening — host-allowlist split, ws_base_url + password kwargs, config-vs-shorthand guard (#406, #412, #410) #416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 36 additions & 13 deletions kalshi/perps/async_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import os
from collections.abc import Callable
from pathlib import Path
from types import TracebackType
from typing import TypedDict, Unpack
Expand Down Expand Up @@ -33,10 +34,12 @@ class PerpsClientInitKwargs(TypedDict, total=False):
key_id: str | None
private_key: str | bytes | None
private_key_path: str | Path | None
password: bytes | str | Callable[[], bytes | str] | None
auth: KalshiAuth | None
config: PerpsConfig | None
demo: bool
base_url: str | None
ws_base_url: str | None
timeout: float | None
max_retries: int | None
transport: httpx.AsyncBaseTransport | None
Expand All @@ -61,10 +64,12 @@ def __init__(
key_id: str | None = None,
private_key_path: str | Path | None = None,
private_key: str | bytes | None = None,
password: bytes | str | Callable[[], bytes | str] | None = None,
auth: KalshiAuth | None = None,
config: PerpsConfig | None = None,
demo: bool = False,
base_url: str | None = None,
ws_base_url: str | None = None,
timeout: float | None = None,
max_retries: int | None = None,
transport: httpx.AsyncBaseTransport | None = None,
Expand All @@ -79,10 +84,10 @@ def __init__(
self._auth = auth
self._auth_owned = False
elif key_id and private_key_path:
self._auth = KalshiAuth.from_key_path(key_id, private_key_path)
self._auth = KalshiAuth.from_key_path(key_id, private_key_path, password=password)
self._auth_owned = True
elif key_id and private_key:
self._auth = KalshiAuth.from_pem(key_id, private_key)
self._auth = KalshiAuth.from_pem(key_id, private_key, password=password)
self._auth_owned = True
elif key_id or private_key or private_key_path:
# Partial credentials — fail fast instead of silently building an
Expand All @@ -98,21 +103,35 @@ def __init__(
self._auth_owned = False

if config is not None:
# A PerpsConfig fully fixes the REST + WS endpoints. Combining it
# with env-determining shorthand would silently ignore that
# shorthand — a real-money footgun if an explicit demo=True is
# dropped — so reject the combination outright.
if demo or base_url is not None or ws_base_url is not None:
raise ValueError(
"Pass either `config=` OR the demo/base_url/ws_base_url "
"shorthand, not both. A PerpsConfig already fixes the REST "
"and WS endpoints, so demo=True/base_url/ws_base_url would be "
"silently ignored. Drop whichever you don't need."
)
self._config: PerpsConfig = config
else:
if demo and base_url is not None and base_url.rstrip("/") != PERPS_DEMO_BASE_URL:
if demo and (
(base_url is not None and base_url.rstrip("/") != PERPS_DEMO_BASE_URL)
or (ws_base_url is not None and ws_base_url.rstrip("/") != PERPS_DEMO_WS_URL)
):
raise ValueError(
"Conflicting environment: demo=True together with explicit "
f"base_url={base_url!r}. demo=True implies base_url="
f"{PERPS_DEMO_BASE_URL!r}; passing a different REST endpoint "
"would leave ws_base_url pinned to the demo WS feed, producing "
"a split REST/WS environment. Drop demo=True and pass both "
"base_url and ws_base_url via a PerpsConfig, or use "
"PerpsConfig.demo() / PerpsConfig.production()."
"Conflicting environment: demo=True together with an explicit "
f"base_url={base_url!r} / ws_base_url={ws_base_url!r} pointing "
"elsewhere. demo=True implies the demo REST + WS endpoints; "
"pass both base_url and ws_base_url via a PerpsConfig instead, "
"or use PerpsConfig.demo() / PerpsConfig.production()."
)
config_kwargs: dict[str, object] = {}
if base_url:
config_kwargs["base_url"] = base_url
if ws_base_url:
config_kwargs["ws_base_url"] = ws_base_url
if demo:
config_kwargs.setdefault("base_url", PERPS_DEMO_BASE_URL)
config_kwargs.setdefault("ws_base_url", PERPS_DEMO_WS_URL)
Expand Down Expand Up @@ -152,11 +171,15 @@ def from_env(cls, **kwargs: Unpack[PerpsClientInitKwargs]) -> AsyncPerpsClient:
)
env_built_auth: KalshiAuth | None = None
if not caller_supplied_auth:
env_built_auth = try_perps_auth_from_env()
env_built_auth = try_perps_auth_from_env(password=kwargs.get("password"))
if env_built_auth is not None:
kwargs["auth"] = env_built_auth
kwargs.setdefault("demo", os.environ.get("KALSHI_PERPS_DEMO", "").lower() == "true")
kwargs.setdefault("base_url", os.environ.get("KALSHI_PERPS_API_BASE_URL"))
# Only inject env endpoint overrides when no explicit config is given —
# otherwise __init__'s config-vs-shorthand guard would reject them.
if "config" not in kwargs:
kwargs.setdefault("demo", os.environ.get("KALSHI_PERPS_DEMO", "").lower() == "true")
kwargs.setdefault("base_url", os.environ.get("KALSHI_PERPS_API_BASE_URL"))
kwargs.setdefault("ws_base_url", os.environ.get("KALSHI_PERPS_WS_BASE_URL"))
client = cls(**kwargs)
if env_built_auth is not None:
client._auth_owned = True
Expand Down
52 changes: 39 additions & 13 deletions kalshi/perps/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from __future__ import annotations

import os
from collections.abc import Callable
from pathlib import Path
from types import TracebackType
from typing import TypedDict, Unpack
Expand Down Expand Up @@ -37,10 +38,12 @@ class PerpsClientInitKwargs(TypedDict, total=False):
key_id: str | None
private_key: str | bytes | None
private_key_path: str | Path | None
password: bytes | str | Callable[[], bytes | str] | None
auth: KalshiAuth | None
config: PerpsConfig | None
demo: bool
base_url: str | None
ws_base_url: str | None
timeout: float | None
max_retries: int | None
transport: httpx.BaseTransport | None
Expand Down Expand Up @@ -70,10 +73,12 @@ def __init__(
key_id: str | None = None,
private_key_path: str | Path | None = None,
private_key: str | bytes | None = None,
password: bytes | str | Callable[[], bytes | str] | None = None,
auth: KalshiAuth | None = None,
config: PerpsConfig | None = None,
demo: bool = False,
base_url: str | None = None,
ws_base_url: str | None = None,
timeout: float | None = None,
max_retries: int | None = None,
transport: httpx.BaseTransport | None = None,
Expand All @@ -88,10 +93,10 @@ def __init__(
self._auth = auth
self._auth_owned = False
elif key_id and private_key_path:
self._auth = KalshiAuth.from_key_path(key_id, private_key_path)
self._auth = KalshiAuth.from_key_path(key_id, private_key_path, password=password)
self._auth_owned = True
elif key_id and private_key:
self._auth = KalshiAuth.from_pem(key_id, private_key)
self._auth = KalshiAuth.from_pem(key_id, private_key, password=password)
self._auth_owned = True
elif key_id or private_key or private_key_path:
# Partial credentials — fail fast instead of silently building an
Expand All @@ -107,21 +112,35 @@ def __init__(
self._auth_owned = False

if config is not None:
# A PerpsConfig fully fixes the REST + WS endpoints. Combining it
# with env-determining shorthand would silently ignore that
# shorthand — a real-money footgun if an explicit demo=True is
# dropped — so reject the combination outright.
if demo or base_url is not None or ws_base_url is not None:
raise ValueError(
"Pass either `config=` OR the demo/base_url/ws_base_url "
"shorthand, not both. A PerpsConfig already fixes the REST "
"and WS endpoints, so demo=True/base_url/ws_base_url would be "
"silently ignored. Drop whichever you don't need."
)
self._config: PerpsConfig = config
else:
if demo and base_url is not None and base_url.rstrip("/") != PERPS_DEMO_BASE_URL:
if demo and (
(base_url is not None and base_url.rstrip("/") != PERPS_DEMO_BASE_URL)
or (ws_base_url is not None and ws_base_url.rstrip("/") != PERPS_DEMO_WS_URL)
):
raise ValueError(
"Conflicting environment: demo=True together with explicit "
f"base_url={base_url!r}. demo=True implies base_url="
f"{PERPS_DEMO_BASE_URL!r}; passing a different REST endpoint "
"would leave ws_base_url pinned to the demo WS feed, producing "
"a split REST/WS environment. Drop demo=True and pass both "
"base_url and ws_base_url via a PerpsConfig, or use "
"PerpsConfig.demo() / PerpsConfig.production()."
"Conflicting environment: demo=True together with an explicit "
f"base_url={base_url!r} / ws_base_url={ws_base_url!r} pointing "
"elsewhere. demo=True implies the demo REST + WS endpoints; "
"pass both base_url and ws_base_url via a PerpsConfig instead, "
"or use PerpsConfig.demo() / PerpsConfig.production()."
)
config_kwargs: dict[str, object] = {}
if base_url:
config_kwargs["base_url"] = base_url
if ws_base_url:
config_kwargs["ws_base_url"] = ws_base_url
if demo:
config_kwargs.setdefault("base_url", PERPS_DEMO_BASE_URL)
config_kwargs.setdefault("ws_base_url", PERPS_DEMO_WS_URL)
Expand Down Expand Up @@ -153,12 +172,15 @@ def from_env(cls, **kwargs: Unpack[PerpsClientInitKwargs]) -> PerpsClient:
Reads:
KALSHI_PERPS_KEY_ID (optional — omit for unauthenticated access)
KALSHI_PERPS_PRIVATE_KEY (PEM string) or KALSHI_PERPS_PRIVATE_KEY_PATH
KALSHI_PERPS_PRIVATE_KEY_PASSPHRASE (optional, for an encrypted key)
KALSHI_PERPS_API_BASE_URL (optional, overrides base_url)
KALSHI_PERPS_WS_BASE_URL (optional, overrides ws_base_url)
KALSHI_PERPS_DEMO (optional, "true" for the demo environment)

Perps credentials are intentionally separate from the prediction-API
``KALSHI_*`` vars — Kalshi recommends a distinct API key for the perps
exchange. Returns an unauthenticated client if no credentials are set.
An explicit ``password=`` (or ``config=``) overrides the env value.
"""
caller_supplied_auth = (
"auth" in kwargs
Expand All @@ -168,11 +190,15 @@ def from_env(cls, **kwargs: Unpack[PerpsClientInitKwargs]) -> PerpsClient:
)
env_built_auth: KalshiAuth | None = None
if not caller_supplied_auth:
env_built_auth = try_perps_auth_from_env()
env_built_auth = try_perps_auth_from_env(password=kwargs.get("password"))
if env_built_auth is not None:
kwargs["auth"] = env_built_auth
kwargs.setdefault("demo", os.environ.get("KALSHI_PERPS_DEMO", "").lower() == "true")
kwargs.setdefault("base_url", os.environ.get("KALSHI_PERPS_API_BASE_URL"))
# Only inject env endpoint overrides when no explicit config is given —
# otherwise __init__'s config-vs-shorthand guard would reject them.
if "config" not in kwargs:
kwargs.setdefault("demo", os.environ.get("KALSHI_PERPS_DEMO", "").lower() == "true")
kwargs.setdefault("base_url", os.environ.get("KALSHI_PERPS_API_BASE_URL"))
kwargs.setdefault("ws_base_url", os.environ.get("KALSHI_PERPS_WS_BASE_URL"))
client = cls(**kwargs)
if env_built_auth is not None:
client._auth_owned = True
Expand Down
26 changes: 20 additions & 6 deletions kalshi/perps/config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,15 +28,15 @@
PERPS_PRODUCTION_WS_URL = "wss://external-api-margin-ws.kalshi.com/trade-api/ws/v2/margin"
PERPS_DEMO_WS_URL = "wss://external-api-margin-ws.demo.kalshi.co/trade-api/ws/v2/margin"

# REST + WS hosts both validate against this union (the perps WS host differs
# from the perps REST host, unlike the prediction API where both share a host).
# Each surface validates against its OWN host set — the perps WS host differs
# from the perps REST host (unlike the prediction API where both share a host),
# so base_url must use a REST host and ws_base_url a WS host.
_PERPS_REST_HOSTS = frozenset(
{"external-api.kalshi.com", "external-api.demo.kalshi.co"}
)
_PERPS_WS_HOSTS = frozenset(
{"external-api-margin-ws.kalshi.com", "external-api-margin-ws.demo.kalshi.co"}
)
_PERPS_KNOWN_HOSTS = _PERPS_REST_HOSTS | _PERPS_WS_HOSTS
_LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})

# Perps prod/demo host pairs, used by the split REST/WS environment guard.
Expand Down Expand Up @@ -81,18 +81,23 @@ def __post_init__(self) -> None:
allow_unknown = self.allow_unknown_host or (
os.environ.get("KALSHI_PERPS_ALLOW_UNKNOWN_HOST", "").strip() == "1"
)
# Validate each field against its OWN host set — a REST URL must use a
# REST host and a WS URL a WS host (they differ for perps), so a swapped
# host is rejected rather than silently accepted via the union.
PerpsConfig._validate_perps_url(
self.base_url,
"base_url",
secure="https",
plaintext="http",
known_hosts=_PERPS_REST_HOSTS,
allow_unknown_host=allow_unknown,
)
PerpsConfig._validate_perps_url(
self.ws_base_url,
"ws_base_url",
secure="wss",
plaintext="ws",
known_hosts=_PERPS_WS_HOSTS,
allow_unknown_host=allow_unknown,
)
# Enforce the /trade-api/v2 REST path component (identical to the
Expand All @@ -103,6 +108,14 @@ def __post_init__(self) -> None:
f"PerpsConfig.base_url must include the /trade-api/v2 path "
f"component, got base_url={self.base_url!r}"
)
# Enforce the /trade-api/ws/v2/margin WS path component (mirrors the REST
# path check above).
ws_path = urlparse(self.ws_base_url).path
if ws_path != "/trade-api/ws/v2/margin":
raise ValueError(
f"PerpsConfig.ws_base_url must include the /trade-api/ws/v2/margin "
f"path component, got ws_base_url={self.ws_base_url!r}"
)
# Reject a split REST/WS environment (prod REST + demo WS, or vice
# versa) — the same real-money-vs-demo footgun KalshiConfig guards.
base_host = (urlparse(self.base_url).hostname or "").lower()
Expand Down Expand Up @@ -148,6 +161,7 @@ def _validate_perps_url(
*,
secure: str,
plaintext: str,
known_hosts: frozenset[str],
allow_unknown_host: bool,
) -> None:
"""Reject URLs that would expose credentials (bad scheme or plaintext-to-remote)."""
Expand All @@ -169,11 +183,11 @@ def _validate_perps_url(
f"(url={url!r}). Plaintext to a remote host would "
f"expose the KALSHI-ACCESS-KEY header and request signature."
)
if host not in _PERPS_KNOWN_HOSTS and host not in _LOCAL_HOSTS:
if host not in known_hosts and host not in _LOCAL_HOSTS:
if not allow_unknown_host:
raise ValueError(
f"PerpsConfig.{field_name} host {host!r} is not a known "
f"Kalshi perps endpoint. Known hosts: {sorted(_PERPS_KNOWN_HOSTS)}. "
f"Kalshi perps endpoint. Known hosts: {sorted(known_hosts)}. "
"If this is an intentional mock server, proxy, or alternate "
"perps region, opt in with PerpsConfig(allow_unknown_host=True) "
"or set the environment variable KALSHI_PERPS_ALLOW_UNKNOWN_HOST=1."
Expand All @@ -185,7 +199,7 @@ def _validate_perps_url(
"KALSHI_PERPS_ALLOW_UNKNOWN_HOST=1 silenced the default error.)",
field_name,
host,
sorted(_PERPS_KNOWN_HOSTS),
sorted(known_hosts),
)

@classmethod
Expand Down
Loading