Skip to content

Security API Reference

Authentication Backends

JWTAuth

socketspec.security.auth.JWTAuth

Validate JSON Web Tokens from a header or query parameter.

Source code in src/socketspec/security/auth.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class JWTAuth:
    """Validate JSON Web Tokens from a header or query parameter."""

    def __init__(
        self,
        secret: str,
        algorithm: str = DEFAULT_JWT_ALGORITHM,
        header: str = DEFAULT_JWT_HEADER,
    ) -> None:
        self._secret = secret
        self._algorithm = algorithm
        self._header = header.lower()

    async def authenticate(
        self,
        headers: dict[str, str],
        query_params: dict[str, str],
    ) -> Identity | None:
        """Decode a JWT and return the resulting identity."""
        token = self._extract_token(headers, query_params)
        if token is None:
            return None
        try:
            import jwt
        except ImportError:
            logger.error("PyJWT is required for JWTAuth but is not installed")
            return None

        try:
            payload = jwt.decode(
                token,
                self._secret,
                algorithms=[self._algorithm],
            )
        except jwt.PyJWTError:
            logger.debug("JWT authentication failed")
            return None

        user_id = payload.get("sub")
        if user_id is not None and not isinstance(user_id, str):
            user_id = str(user_id)

        scopes_raw = payload.get("scopes", [])
        scopes = (
            [str(scope) for scope in scopes_raw] if isinstance(scopes_raw, list) else []
        )

        token_expires_at: datetime | None = None
        exp = payload.get("exp")
        if isinstance(exp, (int, float)):
            token_expires_at = datetime.fromtimestamp(exp, tz=timezone.utc)

        claims = {key: value for key, value in payload.items() if key != "scopes"}

        return Identity(
            user_id=user_id,
            scopes=scopes,
            claims=claims,
            raw_token=token,
            token_expires_at=token_expires_at,
        )

    def _extract_token(
        self,
        headers: dict[str, str],
        query_params: dict[str, str],
    ) -> str | None:
        normalized_headers = {key.lower(): value for key, value in headers.items()}
        header_value = normalized_headers.get(self._header)
        if header_value:
            if header_value.lower().startswith(BEARER_PREFIX):
                return header_value[len(BEARER_PREFIX) :].strip()
            return header_value.strip()
        return query_params.get(QUERY_TOKEN_PARAM)

authenticate(headers: dict[str, str], query_params: dict[str, str]) -> Identity | None async

Decode a JWT and return the resulting identity.

Source code in src/socketspec/security/auth.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
async def authenticate(
    self,
    headers: dict[str, str],
    query_params: dict[str, str],
) -> Identity | None:
    """Decode a JWT and return the resulting identity."""
    token = self._extract_token(headers, query_params)
    if token is None:
        return None
    try:
        import jwt
    except ImportError:
        logger.error("PyJWT is required for JWTAuth but is not installed")
        return None

    try:
        payload = jwt.decode(
            token,
            self._secret,
            algorithms=[self._algorithm],
        )
    except jwt.PyJWTError:
        logger.debug("JWT authentication failed")
        return None

    user_id = payload.get("sub")
    if user_id is not None and not isinstance(user_id, str):
        user_id = str(user_id)

    scopes_raw = payload.get("scopes", [])
    scopes = (
        [str(scope) for scope in scopes_raw] if isinstance(scopes_raw, list) else []
    )

    token_expires_at: datetime | None = None
    exp = payload.get("exp")
    if isinstance(exp, (int, float)):
        token_expires_at = datetime.fromtimestamp(exp, tz=timezone.utc)

    claims = {key: value for key, value in payload.items() if key != "scopes"}

    return Identity(
        user_id=user_id,
        scopes=scopes,
        claims=claims,
        raw_token=token,
        token_expires_at=token_expires_at,
    )

JSON Web Token authentication backend.

from socketspec.security.auth import JWTAuth

auth = JWTAuth(secret="your-256-bit-secret", algorithm="HS256")
Parameter Type Default Description
secret str -- The signing secret or public key
algorithm str "HS256" JWT signing algorithm

Token lookup order:

  1. Authorization: Bearer <token> header
  2. ?token=<token> query parameter

Identity mapping:

Identity field JWT claim
user_id sub
role role
scopes scopes
token_expires_at exp

APIKeyAuth

socketspec.security.auth.APIKeyAuth

Validate a static API key from a header or query parameter.

Source code in src/socketspec/security/auth.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
class APIKeyAuth:
    """Validate a static API key from a header or query parameter."""

    def __init__(
        self,
        api_key: str,
        header: str = DEFAULT_API_KEY_HEADER,
    ) -> None:
        self._api_key = api_key
        self._header = header.lower()

    async def authenticate(
        self,
        headers: dict[str, str],
        query_params: dict[str, str],
    ) -> Identity | None:
        """Return an identity when the provided API key matches."""
        normalized_headers = {key.lower(): value for key, value in headers.items()}
        key = normalized_headers.get(self._header) or query_params.get(
            QUERY_API_KEY_PARAM
        )
        if key is not None and hmac.compare_digest(key, self._api_key):
            return Identity()
        return None

authenticate(headers: dict[str, str], query_params: dict[str, str]) -> Identity | None async

Return an identity when the provided API key matches.

Source code in src/socketspec/security/auth.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
async def authenticate(
    self,
    headers: dict[str, str],
    query_params: dict[str, str],
) -> Identity | None:
    """Return an identity when the provided API key matches."""
    normalized_headers = {key.lower(): value for key, value in headers.items()}
    key = normalized_headers.get(self._header) or query_params.get(
        QUERY_API_KEY_PARAM
    )
    if key is not None and hmac.compare_digest(key, self._api_key):
        return Identity()
    return None

API key authentication backend. Key comparison uses hmac.compare_digest for constant-time comparison.

from socketspec.security.auth import APIKeyAuth

# Single key
auth = APIKeyAuth(api_key="your-api-key")

# Multiple keys
auth = APIKeyAuth(valid_keys={"key-alpha", "key-beta"})
Parameter Type Default Description
api_key str \| None None Single valid API key
valid_keys set[str] \| None None Set of valid API keys (use instead of api_key for multiple keys)

AuthBackend Protocol

Implement this protocol to create custom authentication backends:

from socketspec.security.auth import AuthBackend, Identity

class CustomAuth(AuthBackend):
    async def authenticate(
        self,
        headers: dict[str, str],
        query_params: dict[str, str],
    ) -> Identity | None:
        """Return an Identity on success, or None to reject the connection."""
        ...

Identity

socketspec.security.auth.Identity dataclass

Authentication result attached to a connection.

Source code in src/socketspec/connection.py 
47
48
49
50
51
52
53
54
55
@dataclass
class Identity:
    """Authentication result attached to a connection."""

    user_id: str | None = None
    scopes: list[str] = field(default_factory=list)
    claims: dict[str, Any] = field(default_factory=dict)
    raw_token: str | None = None
    token_expires_at: datetime | None = None

Represents the authenticated identity of a connection.

@dataclass
class Identity:
    user_id: str | None = None
    role: str | None = None
    scopes: list[str] = field(default_factory=list)
    token_expires_at: datetime | None = None
    extra: dict = field(default_factory=dict)

For unauthenticated connections (no auth= configured), all fields retain their default values.

Rate Limiting

RateLimit

socketspec.security.ratelimit.RateLimit dataclass

Configuration for per-connection event rate limiting.

Source code in src/socketspec/security/ratelimit.py 
33
34
35
36
37
38
@dataclass
class RateLimit:
    """Configuration for per-connection event rate limiting."""

    events: int = DEFAULT_RATE_LIMIT_EVENTS
    per_seconds: int = DEFAULT_RATE_LIMIT_WINDOW_SECONDS

Token bucket rate limiter applied per connection.

from socketspec.security.ratelimit import RateLimit

rate_limit = RateLimit(rate=10, capacity=20)
Parameter Type Default Description
rate float -- Tokens replenished per second
capacity int -- Maximum bucket size (burst capacity)

When the bucket is empty, the client receives a RATE_LIMIT_ERROR envelope. The connection remains open.

Origin Validation

OriginValidator

Origin validation is configured through SocketApp constructor parameters:

socket = SocketApp(
    allowed_origins=["https://myapp.com", "https://staging.myapp.com"],
)
Value Behavior
["*"] (default) All origins are permitted
["https://myapp.com"] Only exact matches are permitted
[] All origins are rejected

Origins are compared using exact string matching. Note that https://myapp.com and https://myapp.com:443 are treated as different origins. Trailing slashes are significant.

Rejected connections are closed with WebSocket close code 4003.

Session Configuration

SessionConfig

socketspec.session.SessionConfig dataclass

TTL and heartbeat settings for WebSocket sessions.

Source code in src/socketspec/session.py 
41
42
43
44
45
46
47
48
49
50
@dataclass
class SessionConfig:
    """TTL and heartbeat settings for WebSocket sessions."""

    max_duration: int = DEFAULT_MAX_DURATION_SECONDS
    idle_timeout: int = DEFAULT_IDLE_TIMEOUT_SECONDS
    heartbeat_interval: int = DEFAULT_HEARTBEAT_INTERVAL_SECONDS
    heartbeat_timeout: int = DEFAULT_HEARTBEAT_TIMEOUT_SECONDS
    token_refresh_window: int = DEFAULT_TOKEN_REFRESH_WINDOW_SECONDS
    redis_key_ttl: int = DEFAULT_REDIS_KEY_TTL_SECONDS

Controls connection lifecycle timers.

from socketspec.session import SessionConfig

config = SessionConfig(
    heartbeat_interval=25,
    heartbeat_timeout=10,
    idle_timeout=300,
    max_duration=3600,
    token_refresh_window=60,
)
Parameter Type Default Description
heartbeat_interval int 25 Seconds between __ping__ probes
heartbeat_timeout int 10 Seconds to wait for __pong__ before closing
idle_timeout int 0 Disconnect after this many seconds of inactivity (0 = disabled)
max_duration int 0 Absolute session length limit in seconds (0 = disabled)
token_refresh_window int 60 Send __auth_expiring__ this many seconds before JWT expires