Check curated-path and incoming-path when creating collections #253
3 changed files with 96 additions and 0 deletions
|
|
@ -1,3 +1,11 @@
|
||||||
|
# 6.3.3 (XXXX-XX-XX)
|
||||||
|
|
||||||
|
# Changes
|
||||||
|
|
||||||
|
- Ensure that curated path and incoming path definitions do not point
|
||||||
|
outside of the storage-root tree.
|
||||||
|
|
||||||
|
|
||||||
# 6.3.2 (2026-07-07)
|
# 6.3.2 (2026-07-07)
|
||||||
|
|
||||||
## Changes
|
## Changes
|
||||||
|
|
|
||||||
|
|
@ -136,6 +136,9 @@ async def create_or_replace_collection(
|
||||||
# Check for incoming directory if any of the tokens allows writing
|
# Check for incoming directory if any of the tokens allows writing
|
||||||
validate_incoming_paths(abstract_config, body)
|
validate_incoming_paths(abstract_config, body)
|
||||||
|
|
||||||
|
# Check for paths in the storage route
|
||||||
|
validate_path_destinations(instance_state, body)
|
||||||
|
|
||||||
# If the configuration already exist, we have to delete it here and
|
# If the configuration already exist, we have to delete it here and
|
||||||
# manifest the reduced configuration. This ensures that the new collection
|
# manifest the reduced configuration. This ensures that the new collection
|
||||||
# is fully manifested later
|
# is fully manifested later
|
||||||
|
|
@ -279,3 +282,33 @@ def validate_incoming_paths(
|
||||||
status_code=HTTP_406_NOT_ACCEPTABLE,
|
status_code=HTTP_406_NOT_ACCEPTABLE,
|
||||||
detail=detail,
|
detail=detail,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def validate_path_destinations(
|
||||||
|
instance_state: InstanceState,
|
||||||
|
collection_request: CollectionRequest,
|
||||||
|
):
|
||||||
|
validate_path_destination(
|
||||||
|
instance_state=instance_state,
|
||||||
|
path=collection_request.curated,
|
||||||
|
name='curated',
|
||||||
|
)
|
||||||
|
if collection_request.incoming is not None:
|
||||||
|
validate_path_destination(
|
||||||
|
instance_state=instance_state,
|
||||||
|
path=collection_request.incoming,
|
||||||
|
name='incoming',
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def validate_path_destination(
|
||||||
|
instance_state: InstanceState,
|
||||||
|
path: PurePosixPath,
|
||||||
|
name: str,
|
||||||
|
):
|
||||||
|
abs_path = (instance_state.store_path / Path(path)).resolve().absolute()
|
||||||
|
if not abs_path.is_relative_to(instance_state.store_path):
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=HTTP_406_NOT_ACCEPTABLE,
|
||||||
|
detail=f"{name}-path '{path}' is outside of the store path.",
|
||||||
|
)
|
||||||
|
|
|
||||||
55
dump_things_service/tests/test_paths.py
Normal file
55
dump_things_service/tests/test_paths.py
Normal file
|
|
@ -0,0 +1,55 @@
|
||||||
|
from pathlib import (
|
||||||
|
Path,
|
||||||
|
PurePosixPath,
|
||||||
|
)
|
||||||
|
|
||||||
|
from dump_things_service import HTTP_406_NOT_ACCEPTABLE
|
||||||
|
from dump_things_service.collection_endpoints import CollectionRequest
|
||||||
|
|
||||||
|
# Path to a local simple test schema
|
||||||
|
test_schema_location = str((Path(__file__).parent / 'testschema.yaml').absolute())
|
||||||
|
|
||||||
|
test_curated_path_collection_name = 'test_curated_path_collection'
|
||||||
|
test_curated_path_collection_request = CollectionRequest(
|
||||||
|
name=test_curated_path_collection_name,
|
||||||
|
default_token='test_default_token',
|
||||||
|
curated=PurePosixPath('..'),
|
||||||
|
schema=test_schema_location,
|
||||||
|
)
|
||||||
|
|
||||||
|
test_incoming_path_collection_name = 'test_incoming_path_collection'
|
||||||
|
test_incoming_path_collection_request = CollectionRequest(
|
||||||
|
name=test_incoming_path_collection_name,
|
||||||
|
default_token='test_default_token',
|
||||||
|
curated=PurePosixPath('curated/'),
|
||||||
|
schema=test_schema_location,
|
||||||
|
incoming=PurePosixPath('..'),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_curated_path(fastapi_client_simple):
|
||||||
|
test_client, _, admin_token = fastapi_client_simple
|
||||||
|
|
||||||
|
# Try to add a collection with an illegal curated path
|
||||||
|
response = test_client.post(
|
||||||
|
'/collections',
|
||||||
|
headers={'x-dumpthings-token': admin_token},
|
||||||
|
json=test_curated_path_collection_request.model_dump(
|
||||||
|
mode='json', by_alias=True
|
||||||
|
),
|
||||||
|
)
|
||||||
|
assert response.status_code == HTTP_406_NOT_ACCEPTABLE
|
||||||
|
|
||||||
|
|
||||||
|
def test_incoming_path(fastapi_client_simple):
|
||||||
|
test_client, _, admin_token = fastapi_client_simple
|
||||||
|
|
||||||
|
# Try to add a collection with an illegal curated path
|
||||||
|
response = test_client.post(
|
||||||
|
'/collections',
|
||||||
|
headers={'x-dumpthings-token': admin_token},
|
||||||
|
json=test_incoming_path_collection_request.model_dump(
|
||||||
|
mode='json', by_alias=True
|
||||||
|
),
|
||||||
|
)
|
||||||
|
assert response.status_code == HTTP_406_NOT_ACCEPTABLE
|
||||||
Loading…
Reference in a new issue