Skip to content

Server Pairing

Module: s2auth.server.pairing

Pairing functionality for the S2 server.

ClientNodeId = UUID module-attribute

PairingAttemptId = UUID module-attribute

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

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

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

UTC = timezone.utc module-attribute

log = logging.getLogger(__name__) module-attribute

AccessError

Bases: S2ConnectError

Permission is denied to the client. This generally results in an HTTP 401 error.

Source code in src/s2auth/common/exceptions.py 
53
54
55
56
class AccessError(S2ConnectError):
    """Permission is denied to the client. This generally results in an HTTP 401 error."""

    error_type = PairingErrorType.Other  # TODO should be something else.

PairingNotCompleteError

Bases: S2ConnectError

Pairing was not completed successfully before initiating a connection.

Source code in src/s2auth/common/exceptions.py 
59
60
61
62
class PairingNotCompleteError(S2ConnectError):
    """Pairing was not completed successfully before initiating a connection."""

    error_type = ConnectionErrorType.NoLongerPaired  # TODO should be something else.

ConnectionDetails

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_pairing.py 
47
48
49
50
51
52
53
54
55
class ConnectionDetails(BaseModel):
    initiateSessionUrl: AnyUrl
    accessToken: AccessToken
    certificateFingerprint: Annotated[
        dict[str, str] | None,
        Field(
            description='A map containing the fingerprints of the CA certificate that is being used by the server for communication. The key of the map is the hashing algorithm used to create the fingerprint, the value is the fingerprint itself. All fingerprints must refer to the same CA certificate. The key "SHA265" must always be provided. The property certificateFingerprint is mandatory when the pairing client will be come the communication server.'
        ),
    ] = None

FinalizePairingPostRequest

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_pairing.py 
 99
100
class FinalizePairingPostRequest(BaseModel):
    success: bool | None = None

HmacChallengeResponse

Bases: RootModel[Base64Bytes]

Source code in src/s2auth/common/model/s2_connect_pairing.py 
71
72
73
74
75
76
77
class HmacChallengeResponse(RootModel[Base64Bytes]):
    root: Annotated[
        Base64Bytes,
        Field(
            description="The Base64 encoded response to the challenge that was calculated by the endpoint."
        ),
    ]

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]+$",
        ),
    ]

RequestConnectionDetailsPostRequest

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_pairing.py 
90
91
class RequestConnectionDetailsPostRequest(BaseModel):
    serverHmacChallengeResponse: HmacChallengeResponse

RequestPairingPostRequest

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_pairing.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class RequestPairingPostRequest(BaseModel):
    clientNodeDescription: NodeDescription
    clientEndpointDescription: EndpointDescription
    nodeId: NodeId | None = None
    nodeIdAlias: NodeIdAlias | None = None
    supportedCommunicationProtocols: list[CommunicationProtocol]
    supportedS2MessageVersions: Annotated[
        list[str],
        Field(
            description="The versions of the S2 JSON message schemas this node implementation currently supports."
        ),
    ]
    supportedHmacHashingAlgorithms: list[HmacHashingAlgorithm]
    clientHmacChallenge: HmacChallenge
    forcePairing: Annotated[
        bool | None,
        Field(
            description="Forces the server to attempt pairing, even though the S2 message versions are not compatible. In this case the nodes won't be able to communicate after pairing, but this could later be solved through a software update on one or both of the nodes."
        ),
    ] = False

RequestPairingPostResponse

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_pairing.py 
150
151
152
153
154
155
156
class RequestPairingPostResponse(BaseModel):
    pairingAttemptId: PairingAttemptId
    serverNodeDescription: NodeDescription
    serverEndpointDescription: EndpointDescription
    selectedHmacHashingAlgorithm: HmacHashingAlgorithm
    clientHmacChallengeResponse: HmacChallengeResponse
    serverHmacChallenge: HmacChallenge

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."
        ),
    ]

Deployment

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py 
44
45
46
class Deployment(str, Enum):
    WAN = "WAN"
    LAN = "LAN"

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")]

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

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"

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

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"

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)

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]

Config

Bases: BaseSettings

Source code in src/s2auth/server/config.py 
 6
 7
 8
 9
10
11
12
13
class Config(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=[".env", ".env.docker"], extra="ignore", env_nested_delimiter="__"
    )
    sqlalchemy_db_uri: SecretStr = SecretStr(
        "postgresql://postgres:postgres@localhost/s2auth"
    )
    domain_name: str = "s2connect.example.com"

HookRegistry

Registry for server hooks that can be overridden by client code.

Source code in src/s2auth/server/hooks.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class HookRegistry:
    """Registry for server hooks that can be overridden by client code."""

    def __init__(self) -> None:
        # Initialize with default hook implementations
        self._hooks: dict[HookFunction, HookFunction] = {
            get_server_connection_initiation_endpoint: get_server_connection_initiation_endpoint,
            get_server_endpoint_description: get_server_endpoint_description,
            get_server_node_description: get_server_node_description,
            pairing_attempt_request: pairing_attempt_request,
        }

    def register(self, original_hook: HookFunction, custom_hook: HookFunction) -> None:
        """Register a custom hook implementation.

        Args:
            original_hook: The original hook function to override
            custom_hook: The custom hook implementation function

        Raises:
            RuntimeError: If a custom implementation is already registered for this hook
            KeyError: If the original hook is not recognized
        """
        if original_hook not in self._hooks:
            raise KeyError(
                f"Unknown hook function. Available hooks: {list(self._hooks.keys())}"
            )
        if self._hooks[original_hook] is not original_hook:
            raise RuntimeError(
                "Hook already has a custom implementation registered. "
                "Each hook can only be overridden once."
            )
        self._hooks[original_hook] = custom_hook

    def get(self, hook: HookFunction) -> HookFunction:
        """Get a hook implementation (default or custom).

        Args:
            hook: The hook function reference

        Returns:
            The hook implementation function (either default or custom)

        Raises:
            KeyError: If the hook is not registered
        """
        if hook not in self._hooks:
            raise KeyError("Hook is not registered")
        return self._hooks[hook]

register(original_hook, custom_hook)

Register a custom hook implementation.

Parameters:

Name Type Description Default
original_hook HookFunction

The original hook function to override

 required
custom_hook HookFunction

The custom hook implementation function

 required

Raises:

Type Description
RuntimeError

If a custom implementation is already registered for this hook
KeyError

If the original hook is not recognized
Source code in src/s2auth/server/hooks.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def register(self, original_hook: HookFunction, custom_hook: HookFunction) -> None:
    """Register a custom hook implementation.

    Args:
        original_hook: The original hook function to override
        custom_hook: The custom hook implementation function

    Raises:
        RuntimeError: If a custom implementation is already registered for this hook
        KeyError: If the original hook is not recognized
    """
    if original_hook not in self._hooks:
        raise KeyError(
            f"Unknown hook function. Available hooks: {list(self._hooks.keys())}"
        )
    if self._hooks[original_hook] is not original_hook:
        raise RuntimeError(
            "Hook already has a custom implementation registered. "
            "Each hook can only be overridden once."
        )
    self._hooks[original_hook] = custom_hook

get(hook)

Get a hook implementation (default or custom).

Parameters:

Name Type Description Default
hook HookFunction

The hook function reference

 required

Returns:

Type Description
HookFunction

The hook implementation function (either default or custom)

Raises:

Type Description
KeyError

If the hook is not registered
Source code in src/s2auth/server/hooks.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def get(self, hook: HookFunction) -> HookFunction:
    """Get a hook implementation (default or custom).

    Args:
        hook: The hook function reference

    Returns:
        The hook implementation function (either default or custom)

    Raises:
        KeyError: If the hook is not registered
    """
    if hook not in self._hooks:
        raise KeyError("Hook is not registered")
    return self._hooks[hook]

Settings

Bases: BaseSettings

Source code in src/s2auth/server/settings.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=[".env", ".env.docker"], extra="ignore", env_nested_delimiter="__"
    )

    pairing_node_id: Annotated[str, StringConstraints(min_length=8, max_length=12)]
    server_s2_node_id: UUID4
    supported_communication_protocols: list[CommunicationProtocol] = [
        CommunicationProtocol.WebSocket
    ]
    supported_s2_versions: list[str] = ["v1"]  # most recent first
    supported_s2_connect_versions: list[str] = ["v1"]  # most recent first
    cem_s2_node_id: UUID4
    cem_type: str
    cem_model_name: str
    cem_brand: str
    cem_url: AnyUrl | None = None
    cem_deployment_type: Deployment = Deployment.WAN
    # If unset/empty, pairing starts with generated one-time tokens.
    default_pairing_token: str | None = None
    default_pairing_token_created_at: datetime = Field(
        default=SERVER_PROCESS_STARTED_AT,
        exclude=True,
    )
    pairing_token_ttl_seconds: int = Field(default=300, gt=0)
    ssl_certfile: str = ""
    ssl_keyfile: str = ""

ExpiredOneTimePairingTokenError

Bases: Exception

Raised when a one-time pairing token has expired before use.

Source code in src/s2auth/server/token_manager.py 
19
20
class ExpiredOneTimePairingTokenError(Exception):
    """Raised when a one-time pairing token has expired before use."""

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

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

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}")

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

calculate_certificate_fingerprint_from_certificate_file(certificate_file)

Read a certificate file and return the SHA-256 fingerprint of the leaf certificate.

Source code in src/s2auth/common/hmac.py 
134
135
136
137
138
139
def calculate_certificate_fingerprint_from_certificate_file(
    certificate_file: str | Path,
) -> bytes:
    """Read a certificate file and return the SHA-256 fingerprint of the leaf certificate."""
    cert_bytes = Path(certificate_file).read_bytes()
    return calculate_certificate_fingerprint(_leaf_certificate_bytes(cert_bytes))

create_challenge(length=128)

Create the base64 encoded challenge (sequence of random bytes) to be sent to the other side of the connection. The challenge needs to be passed to the the other side of the connection, who should sign it with a shared pairing token. verify_response can then be used to verify that signature.

Source code in src/s2auth/common/hmac.py 
65
66
67
68
69
70
71
72
def create_challenge(length: int = 128) -> HmacChallenge:
    """
    Create the base64 encoded challenge (sequence of random bytes) to be sent to the other side of the connection.
    The challenge needs to be passed to the the other side of the connection, who should sign it with a shared pairing token.
    verify_response can then be used to verify that signature.
    """
    challenge_value: bytes = secrets.token_bytes(length)
    return HmacChallenge(root=b64encode(challenge_value))

create_response(pairing_token, challenge, deployment, domain_name, fingerprint, algorithm=HmacHashingAlgorithm.SHA256)

Source code in src/s2auth/common/hmac.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def create_response(pairing_token: str,
                    challenge: HmacChallenge,
                    deployment: str | Deployment,
                    domain_name: str | None,
                    fingerprint: bytes | None,
                    algorithm: HmacHashingAlgorithm = HmacHashingAlgorithm.SHA256) -> bytes:
    try:
        digestmod = _get_hashing_algorithm(algorithm)
    except ValueError as e:
        raise VerificationError(str(e)) from e
    if deployment == Deployment.LAN:
        assert fingerprint is not None
        LOGGER.debug(f"Creating LAN HMAC response with fingerprint: fingerprint=CertificateHash(Sha256({list(fingerprint)}))")
        return hmac_response_lan(pairing_token.encode('utf-8'), challenge.root, fingerprint, digestmod)
    else:
        assert domain_name is not None
        LOGGER.debug(f"Creating WAN HMAC response with domain_name: {domain_name}")
        response =  hmac_response_wan(pairing_token.encode('utf-8'), challenge.root, domain_name, digestmod)
        LOGGER.debug(f"WAN HMAC response: {response}")
        return response

create_pairing_code(s2_node_id=None, length=9)

Create pairing code, which is [pairing S2 node ID]-[pairing token] if the S2 node id is set otherwise just the token

Source code in src/s2auth/common/hmac.py 
51
52
53
54
55
56
57
58
59
60
61
62
@register_provider()
def create_pairing_code(s2_node_id: str | None = None, length: int = 9) -> PairingToken:
    """
    Create pairing code, which is [pairing S2 node ID]-[pairing token] if the S2 node id is set otherwise just the token
    """
    if length < 9:
        raise ValueError("The pairing token needs to be at least 9 bytes.")
    token_str = ''.join(random.choice(CHARS) for _ in range(length))

    if s2_node_id:
        return f"{s2_node_id}-{token_str}"
    return token_str

generate_access_token()

Generate a cryptographically secure random access token.

Source code in src/s2auth/common/hmac.py 
75
76
77
78
@register_provider()
def generate_access_token() -> AccessToken:
    """Generate a cryptographically secure random access token."""
    return AccessToken(root=b64encode(secrets.token_bytes(32)))

select_algorithm(node_algorithms)

Source code in src/s2auth/common/hmac.py 
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def select_algorithm(
    node_algorithms: list[HmacHashingAlgorithm],
) -> HmacHashingAlgorithm:
    supported_algorithms = get_supported_algorithms()
    common = set(supported_algorithms) & set(node_algorithms)

    if not common:
        raise IncompatibleHmacHashingAlgorithms(
            f"Node does not support any of our algorithms: {get_supported_algorithms()}"
        )

    # Return the last algorithm from supported_algorithms that's in common
    # (later algorithms are preferred as they are typically stronger)
    return [alg for alg in supported_algorithms if alg in common][-1]

verify_response(pairing_token, challenge, response, deployment, domain_name, fingerprint, algorithm=HmacHashingAlgorithm.SHA256)

Verify that a received challenge response signature for correctness based on pairing token and algorithm.

Source code in src/s2auth/common/hmac.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def verify_response(
    pairing_token: str,
    challenge: HmacChallenge,
    response: bytes,
    deployment: str | Deployment,
    domain_name: str | None,
    fingerprint: bytes | None,
    algorithm: HmacHashingAlgorithm = HmacHashingAlgorithm.SHA256,
) -> bool:
    """
    Verify that a received challenge response signature for correctness based on pairing token and algorithm.
    """
    LOGGER.debug(f"pairing_token: {pairing_token}")
    LOGGER.debug(f"challenge: {challenge.root}")
    LOGGER.debug(f"challenge (as str): {b64encode(challenge.root)}")
    LOGGER.debug(f"response: {response}")
    LOGGER.debug(f"algorithm: {algorithm}")

    correct_digest = create_response(pairing_token, challenge, deployment, domain_name, fingerprint, algorithm)
    LOGGER.debug(f"expected response: {correct_digest}")
    if not hmac.compare_digest(correct_digest, response):
        raise VerificationError("Signature is invalid.")
    return True

config() async

Source code in src/s2auth/server/config.py 
16
17
18
@register_provider()
async def config() -> Config:
    return Config()

get_server_connection_initiation_endpoint(authentication_context, server_settings=Depends[settings]) async

Default hook implementation to get the server connection initiation endpoint.

This hook is called during the pairing phase to retrieve the server's endpoint so the S2 Client Node can connect to the server side to establish an S2 connection.

Parameters:

Name Type Description Default
authentication_context ReadOnlyAuthenticationContext

Read-only view of the authentication context (contains client_node_id, state, etc.)

 required
server_settings Settings

Server configuration settings

 Depends[settings]
Source code in src/s2auth/server/hooks.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@inject
async def get_server_connection_initiation_endpoint(
    authentication_context: ReadOnlyAuthenticationContext,
    server_settings: Settings = Depends[settings],
) -> AnyUrl | None:
    """Default hook implementation to get the server connection initiation endpoint.

    This hook is called during the pairing phase to retrieve the server's endpoint
    so the S2 Client Node can connect to the server side to establish an S2 connection.

    Args:
        authentication_context: Read-only view of the authentication context (contains client_node_id, state, etc.)
        server_settings: Server configuration settings
    """
    return server_settings.cem_url

get_server_endpoint_description(client_node_id, server_settings=Depends[settings]) async

Default hook implementation for getting the server endpoint description.

This hook is called during the pairing request phase and connection initialization phase to generate the server's S2 endpoint descriptions. Override this hook to customize the server's identity or to refuse pairing by raising an S2ConnectError.

Parameters:

Name Type Description Default
client_node_id NodeId

NodeId of the client.

 required
server_settings Settings

Server configuration settings

 Depends[settings]

Returns:

Type Description
EndpointDescription

EndpointDescription for the server
Source code in src/s2auth/server/hooks.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@inject
async def get_server_endpoint_description(
    client_node_id: NodeId,
    server_settings: Settings = Depends[settings],
) -> EndpointDescription:
    """Default hook implementation for getting the server endpoint description.

    This hook is called during the pairing request phase and connection initialization phase to generate the server's
    S2 endpoint descriptions. Override this hook to customize the server's
    identity or to refuse pairing by raising an S2ConnectError.

    Args:
        client_node_id: NodeId of the client.
        server_settings: Server configuration settings

    Returns:
        EndpointDescription for the server
    """
    # Default implementation: return basic server descriptions from settings
    endpoint_description = EndpointDescription(
        deployment=server_settings.cem_deployment_type
    )
    return endpoint_description

get_server_node_description(client_node_id, server_settings=Depends[settings]) async

Default hook implementation for pairing request.

This hook is called to get the server node description the pairing and connection initiation phase to generate the server's S2 endpoint and node descriptions. Override this hook to customize the server's identity.

Parameters:

Name Type Description Default
client_node_id NodeId

NodeId of the client.

 required
server_settings Settings

Server configuration settings

 Depends[settings]

Returns:

Type Description
NodeDescription

NodeDescription for the server
Source code in src/s2auth/server/hooks.py 
 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
@inject
async def get_server_node_description(
    client_node_id: NodeId,
    server_settings: Settings = Depends[settings],
) -> NodeDescription:
    """Default hook implementation for pairing request.

    This hook is called to get the server node description the pairing and connection initiation phase
    to generate the server's S2 endpoint and node descriptions.
    Override this hook to customize the server's identity.

    Args:
        client_node_id: NodeId of the client.
        server_settings: Server configuration settings

    Returns:
        NodeDescription for the server
    """
    # Default implementation: return basic server descriptions from settings
    node_description = NodeDescription(
        id=NodeId(root=server_settings.cem_s2_node_id),
        brand=server_settings.cem_brand,
        role=Role.CEM,
        type=server_settings.cem_type,
        modelName=server_settings.cem_model_name,
    )
    return node_description

hook_registry()

Provider for the hook registry singleton.

Source code in src/s2auth/server/hooks.py 
178
179
180
181
@register_provider(singleton=True)
def hook_registry() -> HookRegistry:
    """Provider for the hook registry singleton."""
    return HookRegistry()

pairing_attempt_request(authentication_context, pairing_context, server_settings=Depends[settings]) async

Default hok implementation during pairing attempt requests.

This hook is called during pairing attempt requests to allow custom code to accept/refuse the pairing attempt. If it returns True, the pairing attempt request is allowed. On errors or False it is refused.

Parameters:

Name Type Description Default
authentication_context ReadOnlyAuthenticationContext

Read-only view of the authentication context (contains client_node_id, state, etc.)

 required
pairing_context ReadOnlyPairingAttemptContext

Read-only view of the pairing attempt context (contains pairing_attempt_id, pairing_token, etc.)

 required
server_settings Settings

Server configuration settings

 Depends[settings]

Returns:

Type Description
bool

Boolean whether the pairing is allowed

Raises:

Type Description
S2ConnectError

To refuse the pairing attempt with a specific error message
Source code in src/s2auth/server/hooks.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@inject
async def pairing_attempt_request(
    authentication_context: ReadOnlyAuthenticationContext,
    pairing_context: ReadOnlyPairingAttemptContext,
    server_settings: Settings = Depends[settings],
) -> bool:
    """Default hok implementation during pairing attempt requests.

    This hook is called during pairing attempt requests to allow custom code to accept/refuse the pairing attempt.
    If it returns True, the pairing attempt request is allowed. On errors or False it is refused.

    Args:
        authentication_context: Read-only view of the authentication context (contains client_node_id, state, etc.)
        pairing_context: Read-only view of the pairing attempt context (contains pairing_attempt_id, pairing_token, etc.)
        server_settings: Server configuration settings

    Returns:
        Boolean whether the pairing is allowed

    Raises:
        S2ConnectError: To refuse the pairing attempt with a specific error message
    """
    return True

settings()

Source code in src/s2auth/server/settings.py 
46
47
48
@register_provider(singleton=True)
def settings() -> Settings:
    return Settings()  # pyright:ignore[reportCallIssue]

resolve_pairing_token(server_settings, generated_token)

Resolve the token to use for a new pairing attempt.

Resolution order: 1. Pending one-time token generated during runtime (if not expired) 2. Startup DEFAULT_PAIRING_TOKEN (one-time, only if still within TTL) 3. Generated fallback token

Source code in src/s2auth/server/token_manager.py 
 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
def resolve_pairing_token(
    server_settings: "Settings",
    generated_token: str,
) -> str:
    """Resolve the token to use for a new pairing attempt.

    Resolution order:
    1. Pending one-time token generated during runtime (if not expired)
    2. Startup DEFAULT_PAIRING_TOKEN (one-time, only if still within TTL)
    3. Generated fallback token
    """
    pending_token, pending_token_expired, had_pending_token = (
        _consume_pending_pairing_token_with_state()
    )
    if pending_token is not None:
        return pending_token
    if had_pending_token and pending_token_expired:
        log.warning("One-time pairing token expired before use.")
        raise ExpiredOneTimePairingTokenError(
            "One-time pairing token expired before use."
        )

    default_token = _normalize_token(server_settings.default_pairing_token)
    if default_token is not None:
        expires_at = server_settings.default_pairing_token_created_at + timedelta(
            seconds=server_settings.pairing_token_ttl_seconds
        )
        # Consume default token exactly once, regardless of expiration state.
        server_settings.default_pairing_token = None
        if datetime.now(UTC) < expires_at:
            return default_token
        log.warning("Default pairing token expired before use.")
        raise ExpiredOneTimePairingTokenError(
            "Default pairing token expired before use."
        )

    return generated_token

_pairing_token_expiry(ttl_seconds)

Source code in src/s2auth/server/pairing.py 
70
71
def _pairing_token_expiry(ttl_seconds: int) -> datetime:
    return datetime.now(UTC) + timedelta(seconds=ttl_seconds)

_is_pairing_token_expired(pairing_context)

Source code in src/s2auth/server/pairing.py 
74
75
76
77
78
def _is_pairing_token_expired(pairing_context: PairingAttemptContext) -> bool:
    expires_at = pairing_context.pairing_token_expires_at
    if expires_at is None:
        return False
    return datetime.now(UTC) >= expires_at

_effective_deployment(auth_ctx, server_settings)

Source code in src/s2auth/server/pairing.py 
81
82
83
84
85
86
87
88
def _effective_deployment(
    auth_ctx: AuthenticationContext,
    server_settings: Settings,
) -> Deployment:
    endpoint = auth_ctx.s2_endpoint_description
    if endpoint is not None and endpoint.deployment is not None:
        return endpoint.deployment
    return server_settings.cem_deployment_type

_effective_fingerprint(deployment, cfg, server_settings)

Source code in src/s2auth/server/pairing.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def _effective_fingerprint(
    deployment: Deployment,
    cfg: Config,
    server_settings: Settings,
) -> bytes | None:
    if deployment != Deployment.LAN:
        return None
    if not server_settings.ssl_certfile:
        return None
    return calculate_certificate_fingerprint_from_certificate_file(
        server_settings.ssl_certfile
    )

initiate_pairing(client_node_id, store_pairing_ctx=Depends[store_pairing_attempt_context], server_settings=Depends[settings], pairing_token=Depends[create_pairing_code]) async

Create and store a pairing attempt for a client node.

This function is the supported Python API for starting pairing state. In-process callers should invoke initiate_pairing directly and either provide a pairing token explicitly or rely on the configured token provider.

Source code in src/s2auth/server/pairing.py 
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
134
135
136
@inject
async def initiate_pairing(
    client_node_id: ClientNodeId,
    store_pairing_ctx: Callable[[PairingAttemptContext], Awaitable[None]] = Depends[
        store_pairing_attempt_context
    ],
    server_settings: Settings = Depends[settings],
    pairing_token: PairingToken = Depends[create_pairing_code],
):
    """Create and store a pairing attempt for a client node.

    This function is the supported Python API for starting pairing state.
    In-process callers should invoke ``initiate_pairing`` directly and either
    provide a pairing token explicitly or rely on the configured token provider.
    """
    log.info("Initiating pairing for client %s", client_node_id)
    log.info("Generated pairing token for client %s: %s", client_node_id, pairing_token)
    pairing_attempt_id: PairingAttemptId = uuid4()
    # Encode UUID string as base64 bytes for S2PairingAttemptId (str)
    pairing_attempt_id_b64 = b64encode(str(pairing_attempt_id).encode("utf-8")).decode("utf-8")
    pairing_attempt_id_var.set(S2PairingAttemptId(root=pairing_attempt_id_b64))
    pairing_node_id = server_settings.pairing_node_id
    ctx = PairingAttemptContext(
        pairing_attempt_id=pairing_attempt_id,
        pairing_token=pairing_token,
        pairing_node_id=NodeIdAlias(root=pairing_node_id),
        client_node_id=client_node_id,
        pairing_token_expires_at=_pairing_token_expiry(
            server_settings.pairing_token_ttl_seconds
        ),
    )
    await store_pairing_ctx(ctx)
    return ctx

unpair(auth_ctx=Depends[authentication_context], pairing_context=Depends[pairing_attempt_context_by_client_node_id], storage=Depends[context_storage_singleton]) async

Remove the authentication and pairing contexts for a paired client.

Source code in src/s2auth/server/pairing.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@inject
async def unpair(
    auth_ctx: AuthenticationContext = Depends[authentication_context],
    pairing_context: PairingAttemptContext = Depends[
        pairing_attempt_context_by_client_node_id
    ],
    storage: ContextStorage = Depends[context_storage_singleton],
) -> None:
    """Remove the authentication and pairing contexts for a paired client."""
    if auth_ctx.client_node_id is None:
        raise ValueError("AuthenticationContext must have client_node_id set")
    if not isinstance(storage, S2InMemoryContextStorage):
        raise TypeError("unpair requires S2InMemoryContextStorage.")

    await storage.delete_context(AuthenticationContext, auth_ctx.client_node_id)
    await storage.delete_context(
        PairingAttemptContext, pairing_context.pairing_attempt_id
    )

request_pairing(request, store_authentication_ctx=Depends[store_authentication_context], storage=Depends[context_storage_singleton], hooks=Depends[hook_registry], cfg=Depends[config], server_settings=Depends[settings], generated_pairing_token=Depends[create_pairing_code]) async

Initiate a new pairing attempt.

Parameters:

Name Type Description Default
request RequestPairingPostRequest

The pairing request containing client descriptions

 required
store_authentication_ctx Callable[[AuthenticationContext], Awaitable[None]]

Function to store authentication context

 Depends[store_authentication_context]
hooks HookRegistry

Hook registry for calling server hooks

 Depends[hook_registry]

Returns:

Type Description
RequestPairingPostResponse

The pairing response with server descriptions and challenge
Source code in src/s2auth/server/pairing.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
@inject
async def request_pairing(
    request: RequestPairingPostRequest,
    store_authentication_ctx: Callable[
        [AuthenticationContext], Awaitable[None]
    ] = Depends[store_authentication_context],
    storage: ContextStorage = Depends[context_storage_singleton],
    hooks: HookRegistry = Depends[hook_registry],
    cfg: Config = Depends[config],
    server_settings: Settings = Depends[settings],
    generated_pairing_token: PairingToken = Depends[create_pairing_code],
) -> RequestPairingPostResponse:
    """Initiate a new pairing attempt.

    Args:
        request: The pairing request containing client descriptions
        store_authentication_ctx: Function to store authentication context
        hooks: Hook registry for calling server hooks

    Returns:
        The pairing response with server descriptions and challenge
    """

    client_node_id = request.clientNodeDescription.id.root

    if not isinstance(storage, S2InMemoryContextStorage):
        raise TypeError(
            "request_pairing requires S2InMemoryContextStorage to retrieve pairing contexts by client_node_id."
        )

    pairing_attempt_id: PairingAttemptId | None = None
    for ctx in await storage.list_contexts(PairingAttemptContext):
        if ctx.client_node_id == client_node_id:
            if _is_pairing_token_expired(ctx):
                log.info(
                    "Pairing token expired for client %s. Re-initializing pairing context.",
                    client_node_id,
                )
                await storage.delete_context(
                    PairingAttemptContext, ctx.pairing_attempt_id
                )
                continue
            pairing_attempt_id = ctx.pairing_attempt_id
            break

    if pairing_attempt_id is None:
        log.info(
            "No pairing context known for client %s. Initializing one from requestPairing.",
            client_node_id,
        )
        try:
            pairing_token = resolve_pairing_token(
                server_settings=server_settings,
                generated_token=generated_pairing_token,
            )
        except ExpiredOneTimePairingTokenError as exc:
            raise AccessError(
                "Pairing token has expired. Generate a new one-time pairing token and retry."
            ) from exc
        initiated_ctx = await initiate_pairing(
            client_node_id=client_node_id,
            pairing_token=pairing_token,
        )
        pairing_attempt_id = initiated_ctx.pairing_attempt_id

    if pairing_attempt_id is None:
        raise RuntimeError("Failed to initialize pairing attempt context")

    async with storage.get_context(
        PairingAttemptContext, pairing_attempt_id
    ) as pairing_context:
        auth_ctx = AuthenticationContext(
            client_node_id=client_node_id,
            state=ClientState.PAIRING,
            s2_endpoint_description=request.clientEndpointDescription,
            s2_node_description=request.clientNodeDescription,
        )
        await store_authentication_ctx(auth_ctx)
        s2_client_node_id_var.set(NodeId(root=client_node_id))

        pairing_context.client_node_id = client_node_id

        algorithm = select_algorithm(request.supportedHmacHashingAlgorithms)
        pairing_context.algorithm = algorithm
        deployment = _effective_deployment(auth_ctx, server_settings)
        fingerprint = _effective_fingerprint(deployment, cfg, server_settings)

        client_response = create_response(
            pairing_token=pairing_context.pairing_token,
            challenge=request.clientHmacChallenge,
            deployment=deployment,
            domain_name=cfg.domain_name,
            fingerprint=fingerprint,
            algorithm=algorithm,
        )
        pairing_context.state = PairingState.INITIATED
        server_challenge = create_challenge()
        pairing_context.server_hmac_challenge = server_challenge

        pairing_hook = hooks.get(pairing_attempt_request)
        pairing_allowed = await pairing_hook(
            ReadOnlyAuthenticationContext.model_validate(auth_ctx.model_dump()),
            ReadOnlyPairingAttemptContext.model_validate(pairing_context.model_dump()),
        )
        if not pairing_allowed:
            raise AccessError(
                f"Client node {auth_ctx.client_node_id} is not allowed to connect."
            )

        endpoint_hook = hooks.get(get_server_endpoint_description)
        node_hook = hooks.get(get_server_node_description)
        server_endpoint_description = await endpoint_hook(auth_ctx.client_node_id)
        server_node_description = await node_hook(auth_ctx.client_node_id)

        return RequestPairingPostResponse(
            selectedHmacHashingAlgorithm=algorithm,
            serverNodeDescription=server_node_description,
            serverEndpointDescription=server_endpoint_description,
            clientHmacChallengeResponse=HmacChallengeResponse(
                root=b64encode(client_response)
            ),
            serverHmacChallenge=server_challenge,
            pairingAttemptId=S2PairingAttemptId(
                root=b64encode(str(pairing_context.pairing_attempt_id).encode("utf-8")).decode("utf-8")
            ),
        )

handle_client_response(request, pairing_context=Depends[pairing_attempt_context], auth_ctx=Depends[authentication_context_by_pairing_attempt_context], hooks=Depends[hook_registry], new_access_token=Depends[generate_access_token], cfg=Depends[config], server_settings=Depends[settings]) async

Handle the client's response and return the server's connection details.

Parameters:

Name Type Description Default
request RequestConnectionDetailsPostRequest

The request from the client for connection details

 required
pairing_context PairingAttemptContext

The pairing attempt context

 Depends[pairing_attempt_context]
auth_ctx AuthenticationContext

The authentication context with its connection and endpoint details

 Depends[authentication_context_by_pairing_attempt_context]
hooks HookRegistry

Hook registry for calling server hooks

 Depends[hook_registry]

Returns:

Type Description
ConnectionDetails

The ConnectionDetails for the client to setup the s2 connection.
Source code in src/s2auth/server/pairing.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
@inject
async def handle_client_response(
    request: RequestConnectionDetailsPostRequest,
    pairing_context: PairingAttemptContext = Depends[pairing_attempt_context],
    auth_ctx: AuthenticationContext = Depends[
        authentication_context_by_pairing_attempt_context
    ],
    hooks: HookRegistry = Depends[hook_registry],
    new_access_token: AccessToken = Depends[generate_access_token],
    cfg: Config = Depends[config],
    server_settings: Settings = Depends[settings],
) -> ConnectionDetails:
    """Handle the client's response and return the server's connection details.

    Args:
        request: The request from the client for connection details
        pairing_context: The pairing attempt context
        auth_ctx: The authentication context with its connection and endpoint details
        hooks: Hook registry for calling server hooks

    Returns:
        The ConnectionDetails for the client to setup the s2 connection.


    """
    challenge_response = request.serverHmacChallengeResponse.root
    assert pairing_context.algorithm is not None, "No algorithm selected."
    assert pairing_context.server_hmac_challenge is not None, "No known hmac challenge."
    if _is_pairing_token_expired(pairing_context):
        pairing_context.state = PairingState.FAILED
        raise AccessError("Pairing token has expired.")
    deployment = _effective_deployment(auth_ctx, server_settings)
    fingerprint = _effective_fingerprint(deployment, cfg, server_settings)
    verify_response(
        pairing_token=pairing_context.pairing_token,
        algorithm=pairing_context.algorithm,
        challenge=pairing_context.server_hmac_challenge,
        response=challenge_response,
        deployment=deployment,
        domain_name=cfg.domain_name,
        fingerprint=fingerprint,
    )

    endpoint_hook = hooks.get(get_server_connection_initiation_endpoint)
    server_endpoint = await endpoint_hook(
        ReadOnlyAuthenticationContext.model_validate(auth_ctx.model_dump()),
    )
    access_token = new_access_token
    auth_ctx.current_access_token = access_token
    auth_ctx.next_access_token = None
    pairing_context.state = PairingState.COMPLETED
    return ConnectionDetails(
        initiateSessionUrl=server_endpoint, accessToken=access_token
    )

finalize_pairing(request, pairing_context=Depends[pairing_attempt_context], auth_ctx=Depends[authentication_context_by_pairing_attempt_context]) async

Finalize a completed pairing attempt.

The client calls finalizePairing after it has successfully stored the connection details returned by requestConnectionDetails. Only then is the authentication context marked as paired.

Parameters:

Name Type Description Default
request FinalizePairingPostRequest

Finalization request with the client-reported success flag.

 required
pairing_context PairingAttemptContext

Pairing attempt context loaded from context storage.

 Depends[pairing_attempt_context]
auth_ctx AuthenticationContext

Authentication context loaded from context storage.

 Depends[authentication_context_by_pairing_attempt_context]

Raises:

Type Description
PairingNotCompleteError

If the pairing attempt has not reached the completed state.
Source code in src/s2auth/server/pairing.py 
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
@inject
async def finalize_pairing(
    request: FinalizePairingPostRequest,
    pairing_context: PairingAttemptContext = Depends[pairing_attempt_context],
    auth_ctx: AuthenticationContext = Depends[
        authentication_context_by_pairing_attempt_context
    ],
) -> None:
    """Finalize a completed pairing attempt.

    The client calls ``finalizePairing`` after it has successfully stored the
    connection details returned by ``requestConnectionDetails``. Only then is the
    authentication context marked as paired.

    Args:
        request: Finalization request with the client-reported success flag.
        pairing_context: Pairing attempt context loaded from context storage.
        auth_ctx: Authentication context loaded from context storage.

    Raises:
        PairingNotCompleteError: If the pairing attempt has not reached the
            completed state.
    """
    if not request.success:
        pairing_context.state = PairingState.FAILED
        return

    if pairing_context.state != PairingState.COMPLETED:
        raise PairingNotCompleteError(
            f"The pairing state was {pairing_context.state} while we expected {PairingState.COMPLETED}."
        )

    auth_ctx.state = ClientState.PAIRED