Skip to content

jmwalletd.auth

jmwalletd.auth

JWT token authority for the wallet daemon.

Implements HS256-based JWT tokens compatible with the reference JoinMarket implementation. Two token types are managed:

  • Access token: short-lived (30 min), used in Authorization / x-jm-authorization headers.
  • Refresh token: longer-lived (4 hr), used only to obtain new token pairs.

Signing keys are regenerated on daemon start and on each wallet unlock/create/lock cycle, ensuring tokens from previous sessions are always invalidated.

Attributes

ACCESS_TOKEN_EXPIRY_SECONDS = 1800 module-attribute

LEEWAY_SECONDS = 10 module-attribute

REFRESH_TOKEN_EXPIRY_SECONDS = 14400 module-attribute

Classes

JMTokenAuthority dataclass

Manages JWT signing keys and token issuance/verification.

Compatible with the reference implementation's auth semantics: - Access and refresh tokens use separate signing keys. - Refresh key is rotated on every token refresh. - All keys are regenerated on reset (wallet lock/unlock cycle).

Source code in jmwalletd/src/jmwalletd/auth.py 
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 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
@dataclass
class JMTokenAuthority:
    """Manages JWT signing keys and token issuance/verification.

    Compatible with the reference implementation's auth semantics:
    - Access and refresh tokens use separate signing keys.
    - Refresh key is rotated on every token refresh.
    - All keys are regenerated on reset (wallet lock/unlock cycle).
    """

    _access_key: str = field(default_factory=lambda: secrets.token_hex(32))
    _refresh_key: str = field(default_factory=lambda: secrets.token_hex(32))
    _wallet_name: str = ""

    @property
    def scope(self) -> str:
        """Return the current scope string for token payloads."""
        if not self._wallet_name:
            return "walletrpc"
        b64_name = base64.b64encode(self._wallet_name.encode()).decode()
        return f"walletrpc {b64_name}"

    def reset(self) -> None:
        """Regenerate all signing keys, invalidating all existing tokens."""
        self._access_key = secrets.token_hex(32)
        self._refresh_key = secrets.token_hex(32)
        self._wallet_name = ""

    def issue(self, wallet_name: str) -> TokenPair:
        """Issue a new access + refresh token pair for the given wallet.

        The refresh signing key is rotated on each call, invalidating any
        previously issued refresh token.
        """
        self._wallet_name = wallet_name
        self._refresh_key = secrets.token_hex(32)

        now = time.time()
        scope = self.scope

        access_payload = {"exp": now + ACCESS_TOKEN_EXPIRY_SECONDS, "scope": scope}
        refresh_payload = {"exp": now + REFRESH_TOKEN_EXPIRY_SECONDS, "scope": scope}

        access_token = jwt.encode(access_payload, self._access_key, algorithm="HS256")
        refresh_token = jwt.encode(refresh_payload, self._refresh_key, algorithm="HS256")

        return TokenPair(
            token=access_token,
            refresh_token=refresh_token,
            scope=scope,
        )

    def verify_access(self, token: str, *, verify_exp: bool = True) -> dict[str, str]:
        """Verify an access token and return the decoded payload.

        Args:
            token: The raw JWT string.
            verify_exp: Whether to enforce expiration. Set to False for the
                token-refresh flow (the expired access token is still accepted).

        Raises:
            jwt.InvalidTokenError: On any verification failure.
        """
        options = {}
        if not verify_exp:
            options["verify_exp"] = False

        payload: dict[str, str] = jwt.decode(
            token,
            self._access_key,
            algorithms=["HS256"],
            leeway=LEEWAY_SECONDS,
            options=options,  # type: ignore[arg-type]
        )

        # Validate scope includes our expected scope.
        #
        # Scope is an OAuth2-style space-separated set of tokens. We must
        # compare token-by-token (set membership) and NOT use a naive
        # substring check: a substring check would accept a token issued
        # for wallet "ali" (scope "walletrpc YWxp") when the daemon is
        # currently serving wallet "alice" (scope "walletrpc YWxpY2U="),
        # because "walletrpc YWxp" is a substring of "walletrpc YWxpY2U=".
        token_scope = payload.get("scope", "")
        if not self.scope:
            return payload
        expected_tokens = set(self.scope.split())
        presented_tokens = set(token_scope.split())
        if not expected_tokens.issubset(presented_tokens):
            msg = f"Scope mismatch: expected '{self.scope}' in '{token_scope}'"
            raise jwt.InvalidTokenError(msg)

        return payload

    def verify_refresh(self, token: str) -> dict[str, str]:
        """Verify a refresh token and return the decoded payload.

        Raises:
            jwt.InvalidTokenError: On any verification failure.
        """
        payload: dict[str, str] = jwt.decode(
            token,
            self._refresh_key,
            algorithms=["HS256"],
            leeway=LEEWAY_SECONDS,
        )

        token_scope = payload.get("scope", "")
        if self.scope:
            expected_tokens = set(self.scope.split())
            presented_tokens = set(token_scope.split())
            if not expected_tokens.issubset(presented_tokens):
                msg = f"Scope mismatch: expected '{self.scope}' in '{token_scope}'"
                raise jwt.InvalidTokenError(msg)

        return payload
Attributes
scope: str property

Return the current scope string for token payloads.

Methods:
issue(wallet_name: str) -> TokenPair

Issue a new access + refresh token pair for the given wallet.

The refresh signing key is rotated on each call, invalidating any previously issued refresh token.

Source code in jmwalletd/src/jmwalletd/auth.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def issue(self, wallet_name: str) -> TokenPair:
    """Issue a new access + refresh token pair for the given wallet.

    The refresh signing key is rotated on each call, invalidating any
    previously issued refresh token.
    """
    self._wallet_name = wallet_name
    self._refresh_key = secrets.token_hex(32)

    now = time.time()
    scope = self.scope

    access_payload = {"exp": now + ACCESS_TOKEN_EXPIRY_SECONDS, "scope": scope}
    refresh_payload = {"exp": now + REFRESH_TOKEN_EXPIRY_SECONDS, "scope": scope}

    access_token = jwt.encode(access_payload, self._access_key, algorithm="HS256")
    refresh_token = jwt.encode(refresh_payload, self._refresh_key, algorithm="HS256")

    return TokenPair(
        token=access_token,
        refresh_token=refresh_token,
        scope=scope,
    )
reset() -> None

Regenerate all signing keys, invalidating all existing tokens.

Source code in jmwalletd/src/jmwalletd/auth.py 
60
61
62
63
64
def reset(self) -> None:
    """Regenerate all signing keys, invalidating all existing tokens."""
    self._access_key = secrets.token_hex(32)
    self._refresh_key = secrets.token_hex(32)
    self._wallet_name = ""
verify_access(token: str, *, verify_exp: bool = True) -> dict[str, str]

Verify an access token and return the decoded payload.

Args: token: The raw JWT string. verify_exp: Whether to enforce expiration. Set to False for the token-refresh flow (the expired access token is still accepted).

Raises: jwt.InvalidTokenError: On any verification failure.

Source code in jmwalletd/src/jmwalletd/auth.py 
 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
def verify_access(self, token: str, *, verify_exp: bool = True) -> dict[str, str]:
    """Verify an access token and return the decoded payload.

    Args:
        token: The raw JWT string.
        verify_exp: Whether to enforce expiration. Set to False for the
            token-refresh flow (the expired access token is still accepted).

    Raises:
        jwt.InvalidTokenError: On any verification failure.
    """
    options = {}
    if not verify_exp:
        options["verify_exp"] = False

    payload: dict[str, str] = jwt.decode(
        token,
        self._access_key,
        algorithms=["HS256"],
        leeway=LEEWAY_SECONDS,
        options=options,  # type: ignore[arg-type]
    )

    # Validate scope includes our expected scope.
    #
    # Scope is an OAuth2-style space-separated set of tokens. We must
    # compare token-by-token (set membership) and NOT use a naive
    # substring check: a substring check would accept a token issued
    # for wallet "ali" (scope "walletrpc YWxp") when the daemon is
    # currently serving wallet "alice" (scope "walletrpc YWxpY2U="),
    # because "walletrpc YWxp" is a substring of "walletrpc YWxpY2U=".
    token_scope = payload.get("scope", "")
    if not self.scope:
        return payload
    expected_tokens = set(self.scope.split())
    presented_tokens = set(token_scope.split())
    if not expected_tokens.issubset(presented_tokens):
        msg = f"Scope mismatch: expected '{self.scope}' in '{token_scope}'"
        raise jwt.InvalidTokenError(msg)

    return payload
verify_refresh(token: str) -> dict[str, str]

Verify a refresh token and return the decoded payload.

Raises: jwt.InvalidTokenError: On any verification failure.

Source code in jmwalletd/src/jmwalletd/auth.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def verify_refresh(self, token: str) -> dict[str, str]:
    """Verify a refresh token and return the decoded payload.

    Raises:
        jwt.InvalidTokenError: On any verification failure.
    """
    payload: dict[str, str] = jwt.decode(
        token,
        self._refresh_key,
        algorithms=["HS256"],
        leeway=LEEWAY_SECONDS,
    )

    token_scope = payload.get("scope", "")
    if self.scope:
        expected_tokens = set(self.scope.split())
        presented_tokens = set(token_scope.split())
        if not expected_tokens.issubset(presented_tokens):
            msg = f"Scope mismatch: expected '{self.scope}' in '{token_scope}'"
            raise jwt.InvalidTokenError(msg)

    return payload

TokenPair dataclass

A pair of access + refresh tokens with metadata.

Source code in jmwalletd/src/jmwalletd/auth.py 
27
28
29
30
31
32
33
34
35
@dataclass
class TokenPair:
    """A pair of access + refresh tokens with metadata."""

    token: str
    refresh_token: str
    token_type: str = "bearer"
    expires_in: int = ACCESS_TOKEN_EXPIRY_SECONDS
    scope: str = ""
Attributes
expires_in: int = ACCESS_TOKEN_EXPIRY_SECONDS class-attribute instance-attribute
refresh_token: str instance-attribute
scope: str = '' class-attribute instance-attribute
token: str instance-attribute
token_type: str = 'bearer' class-attribute instance-attribute