Check curated-path and incoming-path when creating collections #253

Merged
cmo merged 3 commits from safe-paths into master 2026-07-15 10:51:36 +00:00
3 changed files with 96 additions and 0 deletions

View file

@ -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

View file

@ -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.",
)

View 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