diff --git a/AKModel/utils.py b/AKModel/utils.py index 090c44b951fa3d20c6477371591b53cea798a9f6..a7b5d2130d07ad046c36dc1790a94c40efe6d173 100644 --- a/AKModel/utils.py +++ b/AKModel/utils.py @@ -1,15 +1,19 @@ -import json from pathlib import Path import referencing.retrieval from jsonschema import Draft202012Validator from jsonschema.protocols import Validator -from referencing import Registry, Resource +from referencing import Registry from AKPlanning import settings def _construct_schema_path(uri: str | Path) -> Path: + """Construct a schema URI. + + This function also checks for unallowed directory traversals + out of the 'schema' subfolder. + """ schema_base_path = Path(settings.BASE_DIR).resolve() uri_path = (schema_base_path / uri).resolve() if not uri_path.is_relative_to(schema_base_path / "schemas"): @@ -18,8 +22,8 @@ def _construct_schema_path(uri: str | Path) -> Path: @referencing.retrieval.to_cached_resource() -def retrieve_schema(uri: str) -> str: - # avoid dictionary traversals +def retrieve_schema_from_disk(uri: str) -> str: + """Retrieve schemas from disk by URI.""" uri_path = _construct_schema_path(uri) with uri_path.open("r") as ff: return ff.read() @@ -31,7 +35,7 @@ def construct_schema_validator(schema: str | dict) -> Validator: In particular, all schemas from the 'schemas' directory are loaded into the registry. """ - registry = Registry(retrieve=retrieve_schema) + registry = Registry(retrieve=retrieve_schema_from_disk) if isinstance(schema, str): schema_uri = str(Path("schemas") / schema)