diff --git a/src/middlewared/middlewared/api/base/types/filesystem.py b/src/middlewared/middlewared/api/base/types/filesystem.py index 4660f1d0abeef..94af699689fcc 100644 --- a/src/middlewared/middlewared/api/base/types/filesystem.py +++ b/src/middlewared/middlewared/api/base/types/filesystem.py @@ -1,8 +1,7 @@ -import os from pydantic.functional_validators import AfterValidator from typing_extensions import Annotated -__all__ = ["UnixPerm", "Directory"] +__all__ = ["UnixPerm"] def validate_unix_perm(value: str) -> str: @@ -18,13 +17,3 @@ def validate_unix_perm(value: str) -> str: UnixPerm = Annotated[str, AfterValidator(validate_unix_perm)] - - -def validate_dir_path(value: str) -> str: - if value and not os.path.isdir(value): - raise ValueError('This path is not a directory.') - - return value - - -Directory = Annotated[str, AfterValidator(validate_dir_path)] diff --git a/src/middlewared/middlewared/api/v25_04_0/ftp.py b/src/middlewared/middlewared/api/v25_04_0/ftp.py index 64ecd7c1afd41..fcca2608e68df 100644 --- a/src/middlewared/middlewared/api/v25_04_0/ftp.py +++ b/src/middlewared/middlewared/api/v25_04_0/ftp.py @@ -5,7 +5,6 @@ from middlewared.api.base import ( BaseModel, - Directory, Excluded, excluded_field, ForUpdateMetaclass, @@ -30,7 +29,7 @@ class FtpEntry(BaseModel): timeout: Annotated[int, Field(ge=0, le=10000)] timeout_notransfer: Annotated[int, Field(ge=0, le=10000)] onlyanonymous: bool - anonpath: Directory | None + anonpath: str | None onlylocal: bool banner: str filemask: UnixPerm diff --git a/src/middlewared/middlewared/async_validators.py b/src/middlewared/middlewared/async_validators.py index 1e2650d5ac8a2..68235fb4ffe3b 100644 --- a/src/middlewared/middlewared/async_validators.py +++ b/src/middlewared/middlewared/async_validators.py @@ -4,14 +4,14 @@ from middlewared.validators import IpAddress, check_path_resides_within_volume_sync -async def check_path_resides_within_volume(verrors, middleware, schema_name, path): +async def check_path_resides_within_volume(verrors, middleware, schema_name, path, must_be_dir=False): """ async wrapper around synchronous general-purpose path validation function """ vol_names = [vol["vol_name"] for vol in await middleware.call("datastore.query", "storage.volume")] return await middleware.run_in_thread( check_path_resides_within_volume_sync, - verrors, schema_name, path, vol_names + verrors, schema_name, path, vol_names, must_be_dir ) diff --git a/src/middlewared/middlewared/plugins/ftp.py b/src/middlewared/middlewared/plugins/ftp.py index 3d9e239e82099..809f8719640eb 100644 --- a/src/middlewared/middlewared/plugins/ftp.py +++ b/src/middlewared/middlewared/plugins/ftp.py @@ -157,7 +157,9 @@ async def do_update(self, data): new["anonpath"] = None if new["anonpath"] is not None: - await check_path_resides_within_volume(verrors, self.middleware, "ftp_update.anonpath", new["anonpath"]) + await check_path_resides_within_volume( + verrors, self.middleware, "ftp_update.anonpath", new["anonpath"], must_be_dir=True + ) if new["tls"]: if not new["ssltls_certificate"]: diff --git a/src/middlewared/middlewared/validators.py b/src/middlewared/middlewared/validators.py index 757c23adae6a7..8503e32bf1071 100644 --- a/src/middlewared/middlewared/validators.py +++ b/src/middlewared/middlewared/validators.py @@ -392,7 +392,7 @@ def __call__(self, value): raise ValueError('Invalid URL: no netloc specified') -def check_path_resides_within_volume_sync(verrors, schema_name, path, vol_names): +def check_path_resides_within_volume_sync(verrors, schema_name, path, vol_names, must_be_dir=False): """ This provides basic validation of whether a given `path` is allowed to be exposed to end-users. @@ -405,6 +405,8 @@ def check_path_resides_within_volume_sync(verrors, schema_name, path, vol_names) `vol_names` - list of expected pool names + `must_be_dir` - optional check for directory + It checks the following: * path is within /mnt * path is located within one of the specified `vol_names` @@ -422,6 +424,9 @@ def check_path_resides_within_volume_sync(verrors, schema_name, path, vol_names) rp = Path(os.path.realpath(path)) + if must_be_dir and not rp.is_dir(): + verrors.add(schema_name, "The path must be a directory") + vol_paths = [os.path.join("/mnt", vol_name) for vol_name in vol_names] if not path.startswith("/mnt/") or not any( os.path.commonpath([parent]) == os.path.commonpath([parent, rp]) for parent in vol_paths