Repository navigation
Expand file tree
/
Copy pathtoken_store.py
More file actions
92 lines (65 loc) · 2.98 KB
/
Copy pathtoken_store.py
File metadata and controls
92 lines (65 loc) · 2.98 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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
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
"""Token storage for SDK-minted tokens (OBO, M2M, Token Vault), distinct from CacheAdapter."""
from abc import ABC, abstractmethod
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Optional, TypedDict
from .encryption import decrypt, encrypt
from .errors import ConfigurationError
_MIN_SECRET_LENGTH = 32
class _TokenSetRequired(TypedDict):
access_token: str
expires_at: int
class TokenSet(_TokenSetRequired, total=False):
"""A minted access token and its absolute expiration, as stored by a TokenStore."""
granted_scopes: str
token_type: str
@dataclass(frozen=True)
class VerifiedToken:
"""Access token with verified claims from verify_access_token, trusted to build the cache key."""
access_token: str
claims: Mapping[str, Any]
class AbstractTokenStore(ABC):
"""Base class for external token stores with built-in JWE encrypt and decrypt helpers."""
def __init__(self, *, secret: str) -> None:
"""
Args:
secret: Encryption secret of at least 32 characters, for example the output of `openssl rand -hex 32`.
Raises:
ConfigurationError: If secret is blank or shorter than 32 characters.
"""
if not secret or len(secret.strip()) < _MIN_SECRET_LENGTH:
raise ConfigurationError(f"secret must be at least {_MIN_SECRET_LENGTH} characters, for example the output of `openssl rand -hex 32`")
self._secret = secret
@abstractmethod
async def get(self, key: str) -> Optional[TokenSet]:
"""Return the stored token or None if absent."""
pass
@abstractmethod
async def set(self, key: str, value: TokenSet) -> None:
"""Store a token under key."""
pass
@abstractmethod
async def delete(self, key: str) -> None:
"""Delete a stored token by key."""
pass
def encrypt(self, key: str, value: TokenSet) -> str:
"""Encrypt a TokenSet to a JWE string, keyed to this specific cache entry."""
return encrypt(dict(value), self._secret, key)
def decrypt(self, key: str, data: str) -> TokenSet:
"""Decrypt a JWE string back to a TokenSet."""
return decrypt(data, self._secret, key)
class TokenIndexMember(TypedDict):
"""One cached token listed in an index, with the scopes it was granted and when it expires."""
token_key: str
granted_scopes: str
expires_at: int
class IndexedTokenStore(AbstractTokenStore):
"""Token store variant that maintains a scope index with atomic member writes to avoid concurrent-add races."""
@abstractmethod
async def add_index_member(self, index_key: str, member: TokenIndexMember) -> None:
"""Atomically add member to the index, replacing any member with the same token_key."""
pass
@abstractmethod
async def list_index_members(self, index_key: str) -> list[TokenIndexMember]:
"""Return the index members, possibly including expired ones, or [] if the index is absent."""
pass