feat(auth): bootstrap token entries carry an optional scope
This commit is contained in:
@@ -140,7 +140,9 @@ async def _async_main() -> None:
|
||||
# background task). BOOTSTRAP_TOKENS layers on top so first-run /
|
||||
# examples still work without DB writes.
|
||||
token_store = TokenStore(
|
||||
db, bootstrap=TokenStore.parse_bootstrap(settings.bootstrap_tokens)
|
||||
db,
|
||||
bootstrap=TokenStore.parse_bootstrap(settings.bootstrap_tokens),
|
||||
bootstrap_scopes=TokenStore.parse_bootstrap_scopes(settings.bootstrap_tokens),
|
||||
)
|
||||
|
||||
async with AsyncExitStack() as stack:
|
||||
|
||||
@@ -155,6 +155,7 @@ class TokenStore:
|
||||
db: Database | None = None,
|
||||
*,
|
||||
bootstrap: Mapping[str, str] | None = None,
|
||||
bootstrap_scopes: Mapping[str, str] | None = None,
|
||||
ttl_seconds: float = 30.0,
|
||||
flush_interval: float = 5.0,
|
||||
) -> None:
|
||||
@@ -173,9 +174,13 @@ class TokenStore:
|
||||
raise TokenStoreError(msg)
|
||||
by_value[value] = name
|
||||
self._bootstrap_by_value: dict[str, str] = by_value
|
||||
self._bootstrap_scopes: dict[str, str] = dict.fromkeys(
|
||||
by_value.values(), _BOOTSTRAP_SCOPE
|
||||
)
|
||||
# Bootstrap entries are ``"*"`` unless the env narrows them
|
||||
# (``name:value:scope``) - a webhook token in a URL should not be
|
||||
# an admin token.
|
||||
self._bootstrap_scopes: dict[str, str] = {
|
||||
name: (bootstrap_scopes or {}).get(name, _BOOTSTRAP_SCOPE)
|
||||
for name in by_value.values()
|
||||
}
|
||||
|
||||
self._db = db
|
||||
self._ttl = ttl_seconds
|
||||
@@ -191,22 +196,17 @@ class TokenStore:
|
||||
|
||||
@staticmethod
|
||||
def parse_bootstrap(raw: str) -> dict[str, str]:
|
||||
"""Parse ``name1:value1,name2:value2`` (the ``BOOTSTRAP_TOKENS`` form)."""
|
||||
tokens: dict[str, str] = {}
|
||||
for chunk in raw.split(","):
|
||||
entry = chunk.strip()
|
||||
if not entry:
|
||||
continue
|
||||
name, sep, value = entry.partition(":")
|
||||
if not sep:
|
||||
msg = f"token entry missing ':' separator: {entry!r}"
|
||||
raise TokenStoreError(msg)
|
||||
name, value = name.strip(), value.strip()
|
||||
if name in tokens:
|
||||
msg = f"duplicate token name: {name!r}"
|
||||
raise TokenStoreError(msg)
|
||||
tokens[name] = value
|
||||
return tokens
|
||||
"""Parse ``name1:value1,name2:value2[:scope]`` (``BOOTSTRAP_TOKENS``)."""
|
||||
return {name: value for name, (value, _scope) in _parse_entries(raw).items()}
|
||||
|
||||
@staticmethod
|
||||
def parse_bootstrap_scopes(raw: str) -> dict[str, str]:
|
||||
"""Scopes given as a third field in ``BOOTSTRAP_TOKENS``; the rest get ``*``."""
|
||||
return {
|
||||
name: scope
|
||||
for name, (_value, scope) in _parse_entries(raw).items()
|
||||
if scope is not None
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def from_env(cls, raw: str, db: Database | None = None) -> TokenStore:
|
||||
@@ -378,3 +378,22 @@ __all__ = [
|
||||
"TokenStoreError",
|
||||
"hash_token",
|
||||
]
|
||||
|
||||
|
||||
def _parse_entries(raw: str) -> dict[str, tuple[str, str | None]]:
|
||||
entries: dict[str, tuple[str, str | None]] = {}
|
||||
for chunk in raw.split(","):
|
||||
entry = chunk.strip()
|
||||
if not entry:
|
||||
continue
|
||||
name, sep, rest = entry.partition(":")
|
||||
if not sep:
|
||||
msg = f"token entry missing ':' separator: {entry!r}"
|
||||
raise TokenStoreError(msg)
|
||||
value, sep, scope = rest.partition(":")
|
||||
name, value, scope = name.strip(), value.strip(), scope.strip()
|
||||
if name in entries:
|
||||
msg = f"duplicate token name: {name!r}"
|
||||
raise TokenStoreError(msg)
|
||||
entries[name] = (value, scope if sep and scope else None)
|
||||
return entries
|
||||
|
||||
Reference in New Issue
Block a user