Skip to content

Server Connection Initiation

Module: s2auth.server.connection_initiation

Server-side helpers for the S2 Connect connection initiation flow.

The communication client starts a new S2 session by calling /initiateSession with its current access token. The communication server validates the existing pairing, negotiates the communication protocol and S2 message version, and returns a newly generated pending access token. The client must persist that pending token and then confirm it with /confirmAccessToken before the server promotes it to the active access token.

After confirmation, the previous active access token is only retained as a one-time connection token for authenticating the actual S2 communication channel. Each new S2 session repeats this renewal process.

InvalidAccessTokenError

Bases: AccessError

Invalid AccessToken when authenticating a new s2 connection.

Source code in src/s2auth/common/exceptions.py 
65
66
67
68
class InvalidAccessTokenError(AccessError):
    """Invalid AccessToken when authenticating a new s2 connection."""

    error_type = ConnectionErrorType.Other  # TODO should be something else

NoCompatibleCommunitcationProtocol

Bases: S2ConnectErrorWithDetails

No compatible communication protocols are available between the client and server.

Source code in src/s2auth/common/exceptions.py 
89
90
91
92
class NoCompatibleCommunitcationProtocol(S2ConnectErrorWithDetails):
    """No compatible communication protocols are available between the client and server."""

    error_type = ConnectionErrorType.IncompatibleCommunicationProtocols

NoCompatibleS2ConnectVersionError

Bases: S2ConnectErrorWithDetails

No compatible S2Connect versions are available between the client and server.

Source code in src/s2auth/common/exceptions.py 
77
78
79
80
class NoCompatibleS2ConnectVersionError(S2ConnectErrorWithDetails):
    """No compatible S2Connect versions are available between the client and server."""

    error_type = ConnectionErrorType.Other  # TODO should be something else

NoCompatibleS2VersionError

Bases: S2ConnectErrorWithDetails

No compatible S2 versions are available between the client and server.

Source code in src/s2auth/common/exceptions.py 
83
84
85
86
class NoCompatibleS2VersionError(S2ConnectErrorWithDetails):
    """No compatible S2 versions are available between the client and server."""

    error_type = ConnectionErrorType.IncompatibleS2MessageVersions

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.

InvalidServerError

Bases: AccessError

Unknown serverNodeId was specified for this server.

Source code in src/s2auth/common/exceptions.py 
71
72
73
74
class InvalidServerError(AccessError):
    """Unknown serverNodeId was specified for this server."""

    error_type = ConnectionErrorType.Other  # TODO should be something else

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

CommunicationProtocol

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py 
17
18
class CommunicationProtocol(str, Enum):
    WebSocket = "WebSocket"

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

InitiateSessionPostResponse

Bases: BaseModel

Source code in src/s2auth/common/model/s2_connect_session_init.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class InitiateSessionPostResponse(BaseModel):
    selectedCommunicationProtocol: CommunicationProtocol
    selectedS2MessageVersion: Annotated[
        str,
        Field(
            description="The protocol version selected by the server from the list of supported protocol versions provided by the client."
        ),
    ]
    accessToken: AccessToken
    serverNodeDescription: Annotated[
        NodeDescription | None,
        Field(
            description="Optional field to provide (an updated) NodeDescription. When not provided the client will use the stored description."
        ),
    ] = None
    serverEndpointDescription: Annotated[
        EndpointDescription | None,
        Field(
            description="Optional field to provide (an updated) EndpointDescription. When not provided the client will use the stored description."
        ),
    ] = None

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"

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 = ""

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

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

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()

settings()

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

select_protocol(remote_protocols, local_protocols)

Source code in src/s2auth/common/connection_initiation.py 
22
23
24
25
26
def select_protocol(
    remote_protocols: list[CommunicationProtocol],
    local_protocols: list[CommunicationProtocol],
) -> CommunicationProtocol | None:
    return select_compatible_item(remote_protocols, local_protocols)

select_version(remote_versions, local_versions)

Source code in src/s2auth/common/connection_initiation.py 
18
19
def select_version(remote_versions: list[str], local_versions: list[str]) -> str | None:
    return select_compatible_item(remote_versions, local_versions)

initiateConnection(server_node_id, access_token, supported_communication_protocols, supported_s2_versions, selected_s2_connect_version, server_settings=Depends[settings], authentication_ctx=Depends[authentication_context], new_access_token=Depends[generate_access_token], hooks=Depends[hook_registry]) async

Handle POST /initiateSession for a paired communication client.

This implements the server side of the specification's connection initiation steps 4 through 6:

  1. verify that the selected S2 Connect API version is supported;
  2. verify that the client is paired with this server node and presented the currently active access token;
  3. negotiate one common S2 message version and communication protocol;
  4. generate and store a new pending access token; and
  5. return the negotiated values, pending access token, and current server endpoint/node descriptions.

The returned access token is not active yet. The client must persist it and confirm it with confirmAccessToken before the old token is invalidated.

Parameters:

Name Type Description Default
server_node_id NodeId

Node ID of the communication server the client wants to connect to.

 required
access_token AccessToken

Currently active access token supplied by the communication client.

 required
supported_communication_protocols list[CommunicationProtocol]

Communication protocols supported by the client.

 required
supported_s2_versions list[str]

S2 message versions supported by the client.

 required
selected_s2_connect_version str

S2 Connect API version selected by the client.

 required
server_settings Settings

Server configuration injected by the DI container.

 Depends[settings]
authentication_ctx AuthenticationContext

Authentication state for the paired client.

 Depends[authentication_context]
new_access_token AccessToken

Pending access token generated by the DI provider.

 Depends[generate_access_token]
hooks HookRegistry

Hook registry used to retrieve server description hooks.

 Depends[hook_registry]

Returns:

Type Description
InitiateSessionPostResponse

Response containing the negotiated communication protocol, negotiated S2
InitiateSessionPostResponse

message version, pending access token, and server descriptions.

Raises:

Type Description
NoCompatibleS2ConnectVersionError

If the selected S2 Connect API version is not supported by this server.
PairingNotCompleteError

If the client is not paired or has no active access token.
InvalidAccessTokenError

If the supplied access token is not the active token for the pairing.
InvalidServerError

If the request targets a different server node.
NoCompatibleS2VersionError

If there is no overlap between client and server S2 message versions.
NoCompatibleCommunitcationProtocol

If there is no overlap between client and server communication protocols.
Source code in src/s2auth/server/connection_initiation.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 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
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
@inject
async def initiateConnection(
    server_node_id: NodeId,
    access_token: AccessToken,
    supported_communication_protocols: list[CommunicationProtocol],
    supported_s2_versions: list[str],
    selected_s2_connect_version: str,
    server_settings: Settings = Depends[settings],
    authentication_ctx: AuthenticationContext = Depends[authentication_context],
    new_access_token: AccessToken = Depends[generate_access_token],
    hooks: HookRegistry = Depends[hook_registry],
) -> InitiateSessionPostResponse:
    """Handle ``POST /initiateSession`` for a paired communication client.

    This implements the server side of the specification's connection
    initiation steps 4 through 6:

    1. verify that the selected S2 Connect API version is supported;
    2. verify that the client is paired with this server node and presented the
       currently active access token;
    3. negotiate one common S2 message version and communication protocol;
    4. generate and store a new pending access token; and
    5. return the negotiated values, pending access token, and current server
       endpoint/node descriptions.

    The returned access token is not active yet. The client must persist it and
    confirm it with ``confirmAccessToken`` before the old token is invalidated.

    Args:
        server_node_id: Node ID of the communication server the client wants to
            connect to.
        access_token: Currently active access token supplied by the
            communication client.
        supported_communication_protocols: Communication protocols supported by
            the client.
        supported_s2_versions: S2 message versions supported by the client.
        selected_s2_connect_version: S2 Connect API version selected by the
            client.
        server_settings: Server configuration injected by the DI container.
        authentication_ctx: Authentication state for the paired client.
        new_access_token: Pending access token generated by the DI provider.
        hooks: Hook registry used to retrieve server description hooks.

    Returns:
        Response containing the negotiated communication protocol, negotiated S2
        message version, pending access token, and server descriptions.

    Raises:
        NoCompatibleS2ConnectVersionError: If the selected S2 Connect API
            version is not supported by this server.
        PairingNotCompleteError: If the client is not paired or has no active
            access token.
        InvalidAccessTokenError: If the supplied access token is not the active
            token for the pairing.
        InvalidServerError: If the request targets a different server node.
        NoCompatibleS2VersionError: If there is no overlap between client and
            server S2 message versions.
        NoCompatibleCommunitcationProtocol: If there is no overlap between
            client and server communication protocols.
    """
    if selected_s2_connect_version not in server_settings.supported_s2_connect_versions:
        raise NoCompatibleS2ConnectVersionError(
            f"S2 Connect version {selected_s2_connect_version} is not compatible with any of {server_settings.supported_s2_connect_versions}",
            additional_info=f"Supported s2 connect versions: {server_settings.supported_s2_connect_versions}",
        )

    if (
        authentication_ctx.state != ClientState.PAIRED
        or authentication_ctx.current_access_token is None
    ):
        raise PairingNotCompleteError(
            f"The client state was {authentication_ctx.state} while we expected {ClientState.PAIRED}."
        )

    if authentication_ctx.current_access_token != access_token:
        raise InvalidAccessTokenError("Invalid access token")

    if server_settings.server_s2_node_id != server_node_id.root:
        raise InvalidServerError(
            f"Pairing was attempted with server {server_node_id.root} but we are server node {server_settings.server_s2_node_id}"
        )

    selected_version = select_version(
        remote_versions=supported_s2_versions,
        local_versions=server_settings.supported_s2_versions,
    )

    if selected_version is None:
        raise NoCompatibleS2VersionError(
            f"No compatible versions between {supported_s2_versions} and {server_settings.supported_s2_versions}",
            additional_info=f"Supported s2 versions: {server_settings.supported_s2_versions}",
        )

    selected_protocol = select_protocol(
        remote_protocols=supported_communication_protocols,
        local_protocols=server_settings.supported_communication_protocols,
    )
    if selected_protocol is None:
        raise NoCompatibleCommunitcationProtocol(
            f"No compatible communication protocols between {supported_communication_protocols} and {server_settings.supported_communication_protocols}",
            additional_info=f"Supported communication protocols: {server_settings.supported_communication_protocols}.",
        )
    next_access_token = new_access_token
    authentication_ctx.next_access_token = next_access_token

    endpoint_hook = hooks.get(get_server_endpoint_description)
    node_hook = hooks.get(get_server_node_description)

    server_endpoint_description = await endpoint_hook(authentication_ctx.client_node_id)
    server_node_description = await node_hook(authentication_ctx.client_node_id)

    return InitiateSessionPostResponse(
        selectedCommunicationProtocol=selected_protocol,
        selectedS2MessageVersion=selected_version,
        accessToken=next_access_token,
        serverNodeDescription=server_node_description,
        serverEndpointDescription=server_endpoint_description,
    )

validate_access_token(next_access_token, authentication_ctx=Depends[authentication_context]) async

Handle POST /confirmAccessToken for a pending access token.

The client calls this after successfully persisting the pending access token returned by initiateSession. When the token matches the pending token for the pairing, it becomes the new active access token. The previous active access token is moved to current_connection_token so it can be used once to authenticate the S2 communication channel.

Parameters:

Name Type Description Default
next_access_token AccessToken

Pending access token supplied by the client in the confirmation request.

 required
authentication_ctx AuthenticationContext

Authentication state for the paired client.

 Depends[authentication_context]

Raises:

Type Description
InvalidAccessTokenError

If the supplied token is not the pending access token for the pairing.
Source code in src/s2auth/server/connection_initiation.py 
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
async def validate_access_token(
    next_access_token: AccessToken,
    authentication_ctx: AuthenticationContext = Depends[authentication_context],
):
    """Handle ``POST /confirmAccessToken`` for a pending access token.

    The client calls this after successfully persisting the pending access token
    returned by ``initiateSession``. When the token matches the pending token
    for the pairing, it becomes the new active access token. The previous active
    access token is moved to ``current_connection_token`` so it can be used once
    to authenticate the S2 communication channel.

    Args:
        next_access_token: Pending access token supplied by the client in the
            confirmation request.
        authentication_ctx: Authentication state for the paired client.

    Raises:
        InvalidAccessTokenError: If the supplied token is not the pending access
            token for the pairing.
    """
    if next_access_token != authentication_ctx.next_access_token:
        raise InvalidAccessTokenError("Next access token is invalid")

    authentication_ctx.current_connection_token = (
        authentication_ctx.current_access_token
    )
    authentication_ctx.current_access_token = authentication_ctx.next_access_token
    authentication_ctx.next_access_token = None
    authentication_ctx.state = ClientState.CONNECTION_INITIATED

validate_s2_connection_token(connection_token, authentication_ctx=Depends[authentication_context]) async

Validate the one-time token used to open an S2 communication channel.

After confirmAccessToken activates a newly persisted access token, the previous active token is retained as a one-time connection token. A WebSocket or other selected communication protocol can use this token for bearer-token authentication. Once accepted, the token is cleared so it cannot be reused for another S2 session.

Parameters:

Name Type Description Default
connection_token AccessToken

One-time token supplied by the communication client when opening the S2 communication channel.

 required
authentication_ctx AuthenticationContext

Authentication state for the paired client.

 Depends[authentication_context]

Returns:

Type Description
bool

True when the connection token is valid and has been invalidated.

Raises:

Type Description
InvalidAccessTokenError

If the supplied token is not the current one-time connection token.
Source code in src/s2auth/server/connection_initiation.py 
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
async def validate_s2_connection_token(
    connection_token: AccessToken,
    authentication_ctx: AuthenticationContext = Depends[authentication_context],
) -> bool:
    """Validate the one-time token used to open an S2 communication channel.

    After ``confirmAccessToken`` activates a newly persisted access token, the
    previous active token is retained as a one-time connection token. A
    WebSocket or other selected communication protocol can use this token for
    bearer-token authentication. Once accepted, the token is cleared so it
    cannot be reused for another S2 session.

    Args:
        connection_token: One-time token supplied by the communication client
            when opening the S2 communication channel.
        authentication_ctx: Authentication state for the paired client.

    Returns:
        ``True`` when the connection token is valid and has been invalidated.

    Raises:
        InvalidAccessTokenError: If the supplied token is not the current
            one-time connection token.
    """
    if connection_token != authentication_ctx.current_connection_token:
        raise InvalidAccessTokenError("Access token is invalid for an S2 connection.")
    authentication_ctx.current_connection_token = None  # invalidate token
    return True