Repository navigation
Expand file tree
/
Copy pathcache_keys.py
More file actions
151 lines (125 loc) · 4.63 KB
/
Copy pathcache_keys.py
File metadata and controls
151 lines (125 loc) · 4.63 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
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
"""Cache-key builders and scope helpers for the tokens the SDK mints."""
import hashlib
from typing import Optional
from ..utils import get_unverified_payload
_FIELD_SEPARATOR = "\x1f"
def _normalized_scopes(scope: Optional[str]) -> str:
"""Sort and dedupe scopes so equivalent scope strings produce the same cache key."""
if not scope:
return ""
return " ".join(sorted(set(scope.split())))
def is_covered_by(requested: Optional[str], granted: Optional[str]) -> bool:
"""True when the requested scopes are all covered by the granted scopes."""
granted_set = set(_normalized_scopes(granted).split())
requested_set = set(_normalized_scopes(requested).split())
return requested_set.issubset(granted_set)
def session_fingerprint(access_token: str) -> str:
"""
Derive a session-scoped fingerprint for an access token, to keep concurrent
sessions' cached tokens from colliding. Not a trust or authorization signal:
the token is decoded without signature verification.
"""
try:
payload = get_unverified_payload(access_token)
except ValueError:
return hashlib.sha256(access_token.encode()).hexdigest()
if not isinstance(payload, dict):
return hashlib.sha256(access_token.encode()).hexdigest()
sid = payload.get("sid")
if isinstance(sid, str) and sid:
return sid
jti = payload.get("jti")
if isinstance(jti, str) and jti:
return jti
return hashlib.sha256(access_token.encode()).hexdigest()
def _obo_identity_fields(
namespace: str,
*,
issuer: str,
incoming_client_id: str,
exchange_tenant: str,
exchange_client_id: str,
audience: str,
org_id: Optional[str],
session_key: str,
) -> list[str]:
"""The fields every OBO cache key shares, namespaced by key kind.
Folds in the verified issuer, the incoming token's client, the exchange tenant, and the
exchange client so a token minted under one of them is never read back under another.
"""
return [
namespace,
issuer,
incoming_client_id,
exchange_tenant,
exchange_client_id,
audience,
org_id or "",
session_key,
]
def obo_cache_key(
*,
layout: str,
sub: str,
issuer: str,
incoming_client_id: str,
exchange_tenant: str,
exchange_client_id: str,
audience: str,
org_id: Optional[str],
scopes: Optional[str],
session_key: str,
) -> str:
"""Token cache key for an On Behalf Of exchange.
layout is "strict" or "index" so the two scope-matching modes never read each other's tokens.
The normalized scopes are folded in.
"""
fields = _obo_identity_fields(
"obo_token_" + layout,
issuer=issuer,
incoming_client_id=incoming_client_id,
exchange_tenant=exchange_tenant,
exchange_client_id=exchange_client_id,
audience=audience,
org_id=org_id,
session_key=session_key,
)
fields.append(_normalized_scopes(scopes))
return sub + ":" + hashlib.sha256(_FIELD_SEPARATOR.join(fields).encode()).hexdigest()
def index_cache_key(
*,
sub: str,
issuer: str,
incoming_client_id: str,
exchange_tenant: str,
exchange_client_id: str,
audience: str,
org_id: Optional[str],
session_key: str,
) -> str:
"""Cache key for the OBO token index, scoped to caller, issuer, exchange, audience, org, and session.
The "obo_index" namespace keeps it from colliding with either layout's token cache key.
"""
fields = _obo_identity_fields(
"obo_index",
issuer=issuer,
incoming_client_id=incoming_client_id,
exchange_tenant=exchange_tenant,
exchange_client_id=exchange_client_id,
audience=audience,
org_id=org_id,
session_key=session_key,
)
return sub + ":" + hashlib.sha256(_FIELD_SEPARATOR.join(fields).encode()).hexdigest()
def m2m_cache_key(*, tenant: str, client_id: str, audience: str, scopes: Optional[str]) -> str:
"""Cache key for a client-credentials (M2M) exchange, scoped to tenant, client, audience, and scopes."""
fields = _FIELD_SEPARATOR.join(
["m2m", tenant, client_id, audience, _normalized_scopes(scopes)]
)
return hashlib.sha256(fields.encode()).hexdigest()
def token_vault_cache_key(
*, sub: str, connection: str, tenant: str, client_id: str, login_hint: Optional[str] = None
) -> str:
"""Cache key for a Token Vault exchange, scoped to tenant, client, caller, connection, and login hint."""
fields = _FIELD_SEPARATOR.join(["token_vault", tenant, client_id, sub, connection, login_hint or ""])
return hashlib.sha256(fields.encode()).hexdigest()