feat(auth): bootstrap token entries carry an optional scope

This commit is contained in:
hh
2026-08-30 18:25:48 +02:00
parent 064d855382
commit e90963a6c3
3 changed files with 54 additions and 20 deletions
+3 -1
View File
@@ -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:
+38 -19
View File
@@ -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