Skip to content

Server Context

Module: s2auth.server.context

PairingToken = Annotated[str, StringConstraints(pattern='^[A-Za-z0-9+/]{4,}={0,2}$', min_length=4)] module-attribute

log = logging.getLogger(__name__) module-attribute

ClientNodeId = UUID module-attribute

PairingAttemptId = UUID module-attribute

ContextTypeT = TypeVar('ContextTypeT', bound=BaseModel) module-attribute

s2_client_node_id_var = ContextVar('s2_client_node_id', default=None) module-attribute

pairing_attempt_id_var = ContextVar('pairing_attempt_id', default=None) module-attribute

pairing_token_var = ContextVar('pairing_token', default=None) module-attribute

HmacChallenge

Bases: RootModel[Base64Bytes]

Source code in src/s2auth/common/model/s2_connect_pairing.py 
62
63
64
65
66
67
68
class HmacChallenge(RootModel[Base64Bytes]):
    root: Annotated[
        Base64Bytes,
        Field(
            description="Random generated binary data encoded using Base64, used as the challenge for the HMAC based challenge response process. The challenge should be send to the other node as part of the pairing process. It must have a length of at least 32 bytes."
        ),
    ]

HmacHashingAlgorithm

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_pairing.py 
58
59
class HmacHashingAlgorithm(str, Enum):
    SHA256 = "SHA256"

S2PairingAttemptId

Bases: RootModel[str]

Source code in src/s2auth/common/model/s2_connect_pairing.py 
80
81
82
83
84
85
86
87
class PairingAttemptId(RootModel[str]):
    root: Annotated[
        str,
        Field(
            description="A secret identifier that is generated by the server for each pairing attempt. It is used as authentication of the client and to keep track of the client requests through the several steps of the pairing process. Once the pairing process is finished (with or without success) the identifier can be discarded. The PairingAttemptId is a random string with a minimal length of 32 characters. The server must use a cryptographically secure pseudorandom number generator to generate the string. The recommended method is to generate 24 random bytes and Base64 encode them.",
            min_length=32,
        ),
    ]

NodeIdAlias

Bases: RootModel[str]

Source code in src/s2auth/common/model/s2_connect_pairing.py 
36
37
38
39
40
41
42
43
44
class NodeIdAlias(RootModel[str]):
    root: Annotated[
        str,
        Field(
            description="A identifier of the node which is unique for the context of the Endpoint. It is used as a short identifier (since the user might have to type it in manually) for the node, which can be used to lookup the actual nodeId.",
            examples=["A0"],
            pattern="^[0-9a-zA-Z]+$",
        ),
    ]

AccessToken

Bases: RootModel[Base64Bytes]

Source code in src/s2auth/common/model/s2_connect_common.py 
21
22
23
24
25
26
27
class AccessToken(RootModel[Base64Bytes]):
    root: Annotated[
        Base64Bytes,
        Field(
            description="One-time access token for secure access to the S2 message communication channel. It must be renewed every time a client wants to access the S2 message communication channel by calling the requestToken endpoint.This token is valid for one time login, with a maximum 5 years, and should have a minimum length of 32 bytes."
        ),
    ]

EndpointDescription

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_common.py 
49
50
51
52
class EndpointDescription(BaseModel):
    name: str | None = None
    logoUrl: AnyUrl | None = None
    deployment: Deployment | None = None

NodeDescription

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_common.py 
55
56
57
58
59
60
61
62
class NodeDescription(BaseModel):
    id: NodeId
    brand: str
    logoUrl: AnyUrl | None = None
    type: str
    modelName: str
    userDefinedName: str | None = None
    role: Role

NodeId

Bases: RootModel[UUID]

Source code in src/s2auth/common/model/s2_connect_common.py 
13
14
class NodeId(RootModel[UUID]):
    root: Annotated[UUID, Field(description="Unique identifier of the node")]

ClientState

Bases: str, Enum

Source code in src/s2auth/server/context.py 
53
54
55
56
57
58
class ClientState(str, Enum):
    PAIRING = "Pairing"
    PAIRED = "Paired"
    CONNECTION_INITIATED = "Connection Initiated"
    CONNECTED = "Connected"
    DISCONNECTED = "Disconnected"

PairingState

Bases: str, Enum

Source code in src/s2auth/server/context.py 
61
62
63
64
class PairingState(str, Enum):
    INITIATED = "Initiated"
    COMPLETED = "Completed"
    FAILED = "Failed"

S2InMemoryContextStorage

Bases: InMemoryContextStorage

In-memory context storage with typed context listing support.

Source code in src/s2auth/server/context.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class S2InMemoryContextStorage(InMemoryContextStorage):
    """In-memory context storage with typed context listing support."""

    async def list_contexts(self, ctx_type: type[ContextTypeT]) -> list[ContextTypeT]:
        """Return snapshots of all stored contexts for the requested type."""
        type_store = self._states.get(ctx_type, {})  # pyright: ignore[reportPrivateUsage]
        return [
            cast(ContextTypeT, context.model_copy(deep=True))
            for context in type_store.values()
        ]

    async def delete_context(
        self, ctx_type: type[ContextTypeT], context_id: UUID
    ) -> None:
        """Delete a stored context for the requested type and ID."""
        lock = await self._get_lock(  # pyright: ignore[reportPrivateUsage]
            ctx_type, context_id
        )
        async with lock:
            type_store = self._states.get(ctx_type, {})  # pyright: ignore[reportPrivateUsage]
            if context_id not in type_store:
                raise KeyError(f"No {ctx_type.__name__} context known for {context_id}")
            del type_store[context_id]

list_contexts(ctx_type) async

Return snapshots of all stored contexts for the requested type.

Source code in src/s2auth/server/context.py 
70
71
72
73
74
75
76
async def list_contexts(self, ctx_type: type[ContextTypeT]) -> list[ContextTypeT]:
    """Return snapshots of all stored contexts for the requested type."""
    type_store = self._states.get(ctx_type, {})  # pyright: ignore[reportPrivateUsage]
    return [
        cast(ContextTypeT, context.model_copy(deep=True))
        for context in type_store.values()
    ]

delete_context(ctx_type, context_id) async

Delete a stored context for the requested type and ID.

Source code in src/s2auth/server/context.py 
78
79
80
81
82
83
84
85
86
87
88
89
async def delete_context(
    self, ctx_type: type[ContextTypeT], context_id: UUID
) -> None:
    """Delete a stored context for the requested type and ID."""
    lock = await self._get_lock(  # pyright: ignore[reportPrivateUsage]
        ctx_type, context_id
    )
    async with lock:
        type_store = self._states.get(ctx_type, {})  # pyright: ignore[reportPrivateUsage]
        if context_id not in type_store:
            raise KeyError(f"No {ctx_type.__name__} context known for {context_id}")
        del type_store[context_id]

AuthenticationContext

Bases: BaseModel

Authentication context data for a client connection.

Note: Modifications to context instances should be done carefully in multi-threaded/async environments. Consider using the storage's locking mechanisms if implementing complex state updates.

Source code in src/s2auth/server/context.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class AuthenticationContext(BaseModel):
    """Authentication context data for a client connection.

    Note: Modifications to context instances should be done carefully in
    multi-threaded/async environments. Consider using the storage's locking
    mechanisms if implementing complex state updates.
    """

    # TODO: implement validators to ensure certain combinations of values are invalid
    # for instance with state PAIRED, we also need to have an access_token

    state: ClientState | None = None
    client_node_id: ClientNodeId | None = None
    s2_node_description: NodeDescription | None = None
    s2_endpoint_description: EndpointDescription | None = None
    current_connection_token: AccessToken | None = None
    current_access_token: AccessToken | None = None
    next_access_token: AccessToken | None = None

PairingAttemptContext

Bases: BaseModel

Context data for a pairing attempt.

Note: Modifications to context instances should be done carefully in multi-threaded/async environments. Consider using the storage's locking mechanisms if implementing complex state updates.

Source code in src/s2auth/server/context.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class PairingAttemptContext(BaseModel):
    """Context data for a pairing attempt.

    Note: Modifications to context instances should be done carefully in
    multi-threaded/async environments. Consider using the storage's locking
    mechanisms if implementing complex state updates.
    """

    state: PairingState | None = None
    client_node_id: ClientNodeId | None = None
    pairing_attempt_id: PairingAttemptId
    pairing_node_id: NodeIdAlias
    pairing_token: PairingToken
    pairing_token_expires_at: datetime | None = None
    algorithm: HmacHashingAlgorithm | None = None
    server_hmac_challenge: HmacChallenge | None = None

ReadOnlyAuthenticationContext

Bases: AuthenticationContext

Read-only view of AuthenticationContext for passing to hooks.

This class prevents accidental modification of context state in hooks. Any attempt to modify attributes will raise a ValidationError.

Source code in src/s2auth/server/context.py 
137
138
139
140
141
142
143
144
class ReadOnlyAuthenticationContext(AuthenticationContext):
    """Read-only view of AuthenticationContext for passing to hooks.

    This class prevents accidental modification of context state in hooks.
    Any attempt to modify attributes will raise a ValidationError.
    """

    model_config = ConfigDict(frozen=True)

ReadOnlyPairingAttemptContext

Bases: PairingAttemptContext

Read-only view of PairingAttemptContext for passing to hooks.

This class prevents accidental modification of context state in hooks. Any attempt to modify attributes will raise a ValidationError.

Source code in src/s2auth/server/context.py 
147
148
149
150
151
152
153
154
class ReadOnlyPairingAttemptContext(PairingAttemptContext):
    """Read-only view of PairingAttemptContext for passing to hooks.

    This class prevents accidental modification of context state in hooks.
    Any attempt to modify attributes will raise a ValidationError.
    """

    model_config = ConfigDict(frozen=True)

s2_context_storage_singleton()

Use s2auth context storage for server context providers.

Source code in src/s2auth/server/context.py 
92
93
94
95
96
@override_provider(context_storage_singleton)
def s2_context_storage_singleton() -> ContextStorage:
    """Use s2auth context storage for server context providers."""
    log.debug("Instantiating storage")
    return S2InMemoryContextStorage()

client_node_id()

Returns the s2_client_node_id from contextvars.

Source code in src/s2auth/server/context.py 
157
158
159
160
161
162
163
@register_provider()
def client_node_id() -> ClientNodeId:
    """Returns the s2_client_node_id from contextvars."""
    node_id = s2_client_node_id_var.get()
    if node_id is None:
        raise ValueError("s2_client_node_id not set in context")
    return node_id.root

pairing_attempt_id()

Returns the pairing_attempt_id from contextvars.

Source code in src/s2auth/server/context.py 
166
167
168
169
170
171
172
@register_provider()
def pairing_attempt_id() -> PairingAttemptId:
    """Returns the pairing_attempt_id from contextvars."""
    p_id = pairing_attempt_id_var.get()
    if p_id is None:
        raise ValueError("pairing_attempt_id not set in context")
    return UUID(b64decode(p_id.root).decode("utf-8"))

pairing_token()

Returns the pairing token from contextvars.

Source code in src/s2auth/server/context.py 
175
176
177
178
179
180
181
@register_provider()
def pairing_token() -> PairingToken:
    """Returns the pairing token from contextvars."""
    token = pairing_token_var.get()
    if token is None:
        raise ValueError("pairing_token not set in context")
    return token

authentication_context(client_node_id=Depends[client_node_id], storage=Depends[context_storage_singleton]) async

Retrieves the context for the specified client_node_id.

This is an async generator provider that yields the context while holding its per-ID lock. The lock is held for the entire duration that dependent functions use the context, ensuring thread-safe and async-safe modifications.

Works in both async and threaded environments through wepositive-di storage.

Source code in src/s2auth/server/context.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@register_provider(context_manager=True)
@asynccontextmanager
async def authentication_context(
    client_node_id: ClientNodeId = Depends[client_node_id],
    storage: ContextStorage = Depends[context_storage_singleton],
) -> AsyncGenerator[AuthenticationContext, None]:
    """Retrieves the context for the specified client_node_id.

    This is an async generator provider that yields the context while holding
    its per-ID lock. The lock is held for the entire duration that dependent
    functions use the context, ensuring thread-safe and async-safe modifications.

    Works in both async and threaded environments through wepositive-di storage.
    """
    try:
        async with storage.get_context(AuthenticationContext, client_node_id) as ctx:
            yield ctx
    except KeyError as exc:
        raise KeyError(f"No context known for {client_node_id}") from exc

pairing_attempt_context(pairing_attempt_id=Depends[pairing_attempt_id], storage=Depends[context_storage_singleton]) async

Retrieves the context for the specified pairing_attempt_id.

This is an async generator provider that yields the context while holding its per-ID lock. The lock is held for the entire duration that dependent functions use the context, ensuring thread-safe and async-safe modifications.

Works in both async and threaded environments through wepositive-di storage.

Source code in src/s2auth/server/context.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@register_provider(context_manager=True)
@asynccontextmanager
async def pairing_attempt_context(
    pairing_attempt_id: PairingAttemptId = Depends[pairing_attempt_id],
    storage: ContextStorage = Depends[context_storage_singleton],
) -> AsyncGenerator[PairingAttemptContext, None]:
    """Retrieves the context for the specified pairing_attempt_id.

    This is an async generator provider that yields the context while holding
    its per-ID lock. The lock is held for the entire duration that dependent
    functions use the context, ensuring thread-safe and async-safe modifications.

    Works in both async and threaded environments through wepositive-di storage.
    """
    try:
        async with storage.get_context(
            PairingAttemptContext, pairing_attempt_id
        ) as ctx:
            yield ctx
    except KeyError as exc:
        raise KeyError(f"No context known for {pairing_attempt_id}") from exc

pairing_attempt_context_by_client_node_id(client_node_id=Depends[client_node_id], storage=Depends[context_storage_singleton]) async

Retrieve a pairing attempt context by its client_node_id.

Source code in src/s2auth/server/context.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
@register_provider(context_manager=True)
@asynccontextmanager
async def pairing_attempt_context_by_client_node_id(
    client_node_id: ClientNodeId = Depends[client_node_id],
    storage: ContextStorage = Depends[context_storage_singleton],
) -> AsyncGenerator[PairingAttemptContext, None]:
    """Retrieve a pairing attempt context by its client_node_id."""
    if not isinstance(storage, S2InMemoryContextStorage):
        raise TypeError(
            "pairing_attempt_context_by_client_context_id requires S2InMemoryContextStorage."
        )
    for ctx in await storage.list_contexts(PairingAttemptContext):
        if ctx.client_node_id == client_node_id:
            async with storage.get_context(
                PairingAttemptContext, ctx.pairing_attempt_id
            ) as stored_ctx:
                yield stored_ctx
                return

    raise KeyError(f"No context known for client_node_id {client_node_id}")

authentication_context_by_pairing_attempt_context(pairing_attempt_id=Depends[pairing_attempt_id], storage=Depends[context_storage_singleton]) async

Retrieve authentication context through the current pairing attempt.

Source code in src/s2auth/server/context.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
@register_provider(context_manager=True)
@asynccontextmanager
async def authentication_context_by_pairing_attempt_context(
    pairing_attempt_id: PairingAttemptId = Depends[pairing_attempt_id],
    storage: ContextStorage = Depends[context_storage_singleton],
) -> AsyncGenerator[AuthenticationContext, None]:
    """Retrieve authentication context through the current pairing attempt."""
    try:
        async with storage.get_context(
            PairingAttemptContext, pairing_attempt_id
        ) as pairing_context:
            client_node_id = pairing_context.client_node_id
    except KeyError as exc:
        raise KeyError(f"No context known for {pairing_attempt_id}") from exc

    if client_node_id is None:
        raise ValueError("PairingAttemptContext must have client_node_id set")

    try:
        async with storage.get_context(AuthenticationContext, client_node_id) as ctx:
            yield ctx
    except KeyError as exc:
        raise KeyError(f"No context known for {client_node_id}") from exc

store_authentication_context(storage=Depends[context_storage_singleton]) async

Provider that returns a function to store authentication contexts.

Returns a callable that can be used to safely store AuthenticationContext objects in the context storage. Thread-safe and async-safe.

Usage

@inject async def my_function( store_ctx: Callable[[AuthenticationContext], Awaitable[None]] = Depends[store_authentication_context] ): ctx = AuthenticationContext(client_node_id=some_uuid) await store_ctx(ctx)

Source code in src/s2auth/server/context.py 
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
@register_provider()
async def store_authentication_context(
    storage: ContextStorage = Depends[context_storage_singleton],
) -> Callable[[AuthenticationContext], Awaitable[None]]:
    """Provider that returns a function to store authentication contexts.

    Returns a callable that can be used to safely store AuthenticationContext objects
    in the context storage. Thread-safe and async-safe.

    Usage:
        @inject
        async def my_function(
            store_ctx: Callable[[AuthenticationContext], Awaitable[None]] = Depends[store_authentication_context]
        ):
            ctx = AuthenticationContext(client_node_id=some_uuid)
            await store_ctx(ctx)
    """

    async def store_context(context: AuthenticationContext) -> None:
        if context.client_node_id is None:
            raise ValueError("AuthenticationContext must have client_node_id set")
        await storage.store_context(
            AuthenticationContext, context.client_node_id, context
        )

    return store_context

store_pairing_attempt_context(storage=Depends[context_storage_singleton]) async

Provider that returns a function to store pairing attempt contexts.

Returns a callable that can be used to safely store PairingAttemptContext objects in the context storage. Thread-safe and async-safe.

Usage

@inject async def my_function( store_ctx: Callable[[PairingAttemptContext], Awaitable[None]] = Depends[store_pairing_attempt_context] ): ctx = PairingAttemptContext(pairing_attempt_id=some_uuid, ...) await store_ctx(ctx)

Source code in src/s2auth/server/context.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
@register_provider()
async def store_pairing_attempt_context(
    storage: ContextStorage = Depends[context_storage_singleton],
) -> Callable[[PairingAttemptContext], Awaitable[None]]:
    """Provider that returns a function to store pairing attempt contexts.

    Returns a callable that can be used to safely store PairingAttemptContext objects
    in the context storage. Thread-safe and async-safe.

    Usage:
        @inject
        async def my_function(
            store_ctx: Callable[[PairingAttemptContext], Awaitable[None]] = Depends[store_pairing_attempt_context]
        ):
            ctx = PairingAttemptContext(pairing_attempt_id=some_uuid, ...)
            await store_ctx(ctx)
    """

    async def store_context(context: PairingAttemptContext) -> None:
        await storage.store_context(
            PairingAttemptContext, context.pairing_attempt_id, context
        )

    return store_context