Skip to content

Server Hooks

Module: s2auth.server.hooks

Server hooks for customizing S2 authentication behavior.

This module provides hooks that can be overridden to customize the behavior of the S2 authentication server. Each hook is registered in a HookRegistry and can be replaced by client code.

See docs/server/hooks.md for detailed documentation on each hook and how to override them.

HookFunction = Callable[..., Awaitable[Any]] module-attribute

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

Role

Bases: str, Enum

Source code in src/s2auth/common/model/s2_connect_common.py 
39
40
41
class Role(str, Enum):
    CEM = "CEM"
    RM = "RM"

EndpointDescription

Bases: BaseModel

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

NodeDescription

Bases: BaseModel

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

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)

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

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

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

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

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

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

register_hook(original_hook, hook_registry=Depends[hook_registry])

Decorator to register a custom hook implementation.

This decorator allows client code to override default hook implementations. Each hook can only be overridden once.

Parameters:

Name Type Description Default
original_hook HookFunction

The original hook function to override

 required

Returns:

Type Description
Callable[[HookFunction], HookFunction]

Decorator function

Raises:

Type Description
RuntimeError

If the hook is already registered by custom code
KeyError

If the hook is not recognized
Example 
from s2auth.server import hooks
from wepositive_di import Depends, inject

@register_hook(hooks.pairing_attempt_request)
@inject
async def my_pairing_hook(
    authentication_context: AuthenticationContext,
    pairing_context: PairingAttemptContext,
    server_settings: Settings = Depends[settings],
) -> tuple[EndpointDescription, NodeDescription]:
    # Custom implementation
    ...
Source code in src/s2auth/server/hooks.py 
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
@inject
def register_hook(
    original_hook: HookFunction, hook_registry: HookRegistry = Depends[hook_registry]
) -> Callable[[HookFunction], HookFunction]:
    """Decorator to register a custom hook implementation.

    This decorator allows client code to override default hook implementations.
    Each hook can only be overridden once.

    Args:
        original_hook: The original hook function to override

    Returns:
        Decorator function

    Raises:
        RuntimeError: If the hook is already registered by custom code
        KeyError: If the hook is not recognized

    Example:
        ```python
        from s2auth.server import hooks
        from wepositive_di import Depends, inject

        @register_hook(hooks.pairing_attempt_request)
        @inject
        async def my_pairing_hook(
            authentication_context: AuthenticationContext,
            pairing_context: PairingAttemptContext,
            server_settings: Settings = Depends[settings],
        ) -> tuple[EndpointDescription, NodeDescription]:
            # Custom implementation
            ...
        ```
    """

    def decorator(custom_hook: HookFunction) -> HookFunction:
        # Get the singleton registry instance by calling the provider
        hook_registry.register(original_hook, custom_hook)
        return custom_hook

    return decorator