Michael Hanke
c332abbd1b
Some checks failed
Codespell / Check for spelling errors (push) Successful in 42s
Codespell / Check for spelling errors (pull_request) Successful in 43s
Ruff / Code linting (push) Failing after 52s
Ruff / Code linting (pull_request) Failing after 54s
Test execution / Test-all (push) Successful in 1m27s
Type annotation (PR) / check (pull_request) Failing after 1m30s
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
import logging
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from dump_things_service import HTTP_401_UNAUTHORIZED
|
|
from dump_things_service.abstract_config import (
|
|
Configuration,
|
|
hash_token_representation,
|
|
)
|
|
from dump_things_service.instance_state import InstanceState
|
|
|
|
logger = logging.getLogger('dump_things_service')
|
|
|
|
|
|
def authenticate_admin(
|
|
instance_state: InstanceState,
|
|
abstract_config: Configuration,
|
|
api_key: str,
|
|
):
|
|
if api_key:
|
|
hashed_token_representation = hash_token_representation(api_key)
|
|
if hashed_token_representation == instance_state.bootstrap_token:
|
|
logger.info('authenticate_admin: using bootstrap token')
|
|
return
|
|
|
|
for token_name, token_config in abstract_config.admin_tokens.items():
|
|
if token_config.representation == hashed_token_representation:
|
|
logger.info(
|
|
"authenticate_admin: using token '%s'",
|
|
token_name,
|
|
)
|
|
return
|
|
|
|
raise HTTPException(
|
|
status_code=HTTP_401_UNAUTHORIZED,
|
|
detail='Invalid admin token',
|
|
)
|