Skip to content

Server Token Manager

Module: s2auth.server.token_manager

One-time pairing token manager for interactive server use.

UTC = timezone.utc module-attribute

_lock = threading.Lock() module-attribute

_pending_token = None module-attribute

_pending_token_expires_at = None module-attribute

log = logging.getLogger(__name__) module-attribute

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

_normalize_token(token)

Treat empty or whitespace-only tokens as unset.

Source code in src/s2auth/server/token_manager.py 
23
24
25
26
27
28
29
30
def _normalize_token(token: str | None) -> str | None:
    """Treat empty or whitespace-only tokens as unset."""
    if token is None:
        return None
    normalized = token.strip()
    if normalized == "":
        return None
    return normalized

set_pending_pairing_token(token, ttl_seconds=None)

Store a one-time pairing token to be consumed by the next new client.

Source code in src/s2auth/server/token_manager.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
def set_pending_pairing_token(token: str, ttl_seconds: int | None = None) -> None:
    """Store a one-time pairing token to be consumed by the next new client."""
    global _pending_token, _pending_token_expires_at
    with _lock:
        _pending_token = token
        if ttl_seconds is None:
            _pending_token_expires_at = None
        else:
            _pending_token_expires_at = datetime.now(UTC) + timedelta(
                seconds=ttl_seconds
            )
    log.info("One-time pairing token: %s, expires at %s", _pending_token, _pending_token_expires_at)
    log.info("Press P + Enter to generate a new one-time pairing token for the next client.")

consume_pending_pairing_token()

Return and clear the pending one-time pairing token, or None if not set.

Source code in src/s2auth/server/token_manager.py 
48
49
50
51
def consume_pending_pairing_token() -> str | None:
    """Return and clear the pending one-time pairing token, or None if not set."""
    token, _, _ = _consume_pending_pairing_token_with_state()
    return token

_consume_pending_pairing_token_with_state()

Consume pending token and return (token, expired_before_use, had_pending_token).

Source code in src/s2auth/server/token_manager.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def _consume_pending_pairing_token_with_state() -> tuple[str | None, bool, bool]:
    """Consume pending token and return (token, expired_before_use, had_pending_token)."""
    global _pending_token, _pending_token_expires_at
    with _lock:
        had_pending_token = _pending_token is not None
        if (
            _pending_token_expires_at is not None
            and datetime.now(UTC) >= _pending_token_expires_at
        ):
            _pending_token = None
            _pending_token_expires_at = None
            return None, True, had_pending_token
        token = _pending_token
        _pending_token = None
        _pending_token_expires_at = None
        return token, False, had_pending_token

prime_default_pairing_token(server_settings)

Promote DEFAULT_PAIRING_TOKEN into the pending one-time token bucket.

This is intended for server startup so token lifetime starts at startup rather than at first pairing request.

Source code in src/s2auth/server/token_manager.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def prime_default_pairing_token(server_settings: "Settings") -> None:
    """Promote DEFAULT_PAIRING_TOKEN into the pending one-time token bucket.

    This is intended for server startup so token lifetime starts at startup
    rather than at first pairing request.
    """
    default_token = _normalize_token(server_settings.default_pairing_token)
    if default_token is None:
        return

    set_pending_pairing_token(
        default_token,
        ttl_seconds=server_settings.pairing_token_ttl_seconds,
    )
    server_settings.default_pairing_token = None

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