c0e16a5840
Add scanner error detection to configuration loading. Raise an error if the configuration YAML-file cannot be loaded. Detect and reject token definitions with write permissions and without incoming-labels or write access for collections that do not define an incoming directory. Rename `get_permissions` to `get_token_permissions`
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
"""Use configuration information to fetch token permissions, ids, and incoming_label """
|
|
|
|
from dump_things_service.abstract_config import Configuration
|
|
from dump_things_service.auth import (
|
|
AuthenticationInfo,
|
|
AuthenticationSource,
|
|
InvalidTokenError,
|
|
)
|
|
from dump_things_service.abstract_config import (
|
|
get_token_permissions,
|
|
get_token_config_for_representation_and_collection,
|
|
)
|
|
|
|
|
|
class ConfigAuthenticationSource(AuthenticationSource):
|
|
def __init__(
|
|
self,
|
|
abstract_configuration: Configuration,
|
|
collection_name: str,
|
|
):
|
|
self.abstract_configuration = abstract_configuration
|
|
self.collection_name = collection_name
|
|
|
|
def authenticate(
|
|
self,
|
|
token_representation: str,
|
|
) -> AuthenticationInfo:
|
|
|
|
result = get_token_config_for_representation_and_collection(
|
|
self.abstract_configuration,
|
|
self.collection_name,
|
|
token_representation,
|
|
)
|
|
|
|
if not result:
|
|
msg = f'Token not valid for collection `{self.collection_name}`'
|
|
raise InvalidTokenError(msg)
|
|
|
|
_, token_config, token_collection_config = result
|
|
return AuthenticationInfo(
|
|
token_permission=get_token_permissions(token_collection_config.mode),
|
|
user_id=token_config.user_id,
|
|
incoming_label=token_collection_config.incoming_label,
|
|
)
|