This is an automated email from the ASF dual-hosted git repository. arm pushed a commit to branch safe_paths in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit 0e795273a05049660d793b79a527cc8b0465ac8d Author: Alastair McFarlane <[email protected]> AuthorDate: Tue Mar 31 17:05:39 2026 +0100 Initial safe path work --- atr/models/safe.py | 30 ++++++++++++++++++++++++++++-- atr/paths.py | 44 ++++++++++++++++++++++---------------------- 2 files changed, 50 insertions(+), 24 deletions(-) diff --git a/atr/models/safe.py b/atr/models/safe.py index 61ba28f5..da1093fb 100644 --- a/atr/models/safe.py +++ b/atr/models/safe.py @@ -57,6 +57,9 @@ class SafeType: def __bool__(self) -> bool: return True + def __fspath__(self) -> str: + return self._value + def __eq__(self, other: object) -> bool: if isinstance(other, self.__class__): return self._value == other._value @@ -85,6 +88,30 @@ class SafeType: return {"type": "string"} +class StatePath: + """An absolute path within the managed storage system.""" + + __slots__ = ("_path",) + + def __init__(self, path: pathlib.Path) -> None: + if not path.is_absolute(): + raise ValueError("Path must be absolute") + self._path = path + + def __fspath__(self) -> str: + return str(self._path) + + def __str__(self) -> str: + return str(self._path) + + def __truediv__(self, other: str | pathlib.Path | SafeType) -> pathlib.Path: + return self._path / str(other) if isinstance(other, SafeType) else self._path / other + + @property + def path(self) -> pathlib.Path: + return self._path + + class Alphanumeric(SafeType): @classmethod def _valid_chars(cls) -> frozenset[str]: @@ -93,8 +120,7 @@ class Alphanumeric(SafeType): class CommitteeKey(Alphanumeric): - def _additional_validations(self, value: str): - pass + pass class Numeric(SafeType): diff --git a/atr/paths.py b/atr/paths.py index 6da2558c..6314489e 100644 --- a/atr/paths.py +++ b/atr/paths.py @@ -25,46 +25,46 @@ import atr.models.sql as sql def base_path_for_revision( project_key: safe.ProjectKey, version_key: safe.VersionKey, revision: safe.RevisionNumber ) -> pathlib.Path: - return pathlib.Path(get_unfinished_dir(), str(project_key), str(version_key), str(revision)) + return get_unfinished_dir() / project_key / version_key / revision -def get_archives_dir() -> pathlib.Path: - return pathlib.Path(config.get().ARCHIVES_STORAGE_DIR) +def get_archives_dir() -> safe.StatePath: + return safe.StatePath(pathlib.Path(config.get().ARCHIVES_STORAGE_DIR)) -def get_attestable_dir() -> pathlib.Path: - return pathlib.Path(config.get().ATTESTABLE_STORAGE_DIR) +def get_attestable_dir() -> safe.StatePath: + return safe.StatePath(pathlib.Path(config.get().ATTESTABLE_STORAGE_DIR)) -def get_downloads_dir() -> pathlib.Path: - return pathlib.Path(config.get().DOWNLOADS_STORAGE_DIR) +def get_downloads_dir() -> safe.StatePath: + return safe.StatePath(pathlib.Path(config.get().DOWNLOADS_STORAGE_DIR)) -def get_finished_dir() -> pathlib.Path: - return pathlib.Path(config.get().FINISHED_STORAGE_DIR) +def get_finished_dir() -> safe.StatePath: + return safe.StatePath(pathlib.Path(config.get().FINISHED_STORAGE_DIR)) def get_finished_dir_for(project_key: safe.ProjectKey, version_key: safe.VersionKey) -> pathlib.Path: - return pathlib.Path(config.get().FINISHED_STORAGE_DIR) / str(project_key) / str(version_key) + return get_finished_dir() / project_key / version_key -def get_quarantined_dir() -> pathlib.Path: - return pathlib.Path(config.get().STATE_DIR) / "quarantined" +def get_quarantined_dir() -> safe.StatePath: + return safe.StatePath(pathlib.Path(config.get().STATE_DIR) / "quarantined") -def get_tmp_dir() -> pathlib.Path: +def get_tmp_dir() -> safe.StatePath: # This must be on the same filesystem as the other state subdirectories - return pathlib.Path(config.get().STATE_DIR) / "temporary" + return safe.StatePath(pathlib.Path(config.get().STATE_DIR) / "temporary") -def get_unfinished_dir() -> pathlib.Path: - return pathlib.Path(config.get().UNFINISHED_STORAGE_DIR) +def get_unfinished_dir() -> safe.StatePath: + return safe.StatePath(pathlib.Path(config.get().UNFINISHED_STORAGE_DIR)) def get_unfinished_dir_for( project_key: safe.ProjectKey, version_key: safe.VersionKey, revision: safe.RevisionNumber ) -> pathlib.Path: - return pathlib.Path(config.get().UNFINISHED_STORAGE_DIR) / str(project_key) / str(version_key) / str(revision) + return get_unfinished_dir() / project_key / version_key / revision def get_upload_staging_dir(session_token: str) -> pathlib.Path: @@ -94,7 +94,7 @@ def release_directory_base(release: sql.Release) -> pathlib.Path: project_key = release.project.key version_key = release.version - base_dir: pathlib.Path | None = None + base_dir: safe.StatePath | None = None match phase: case sql.ReleasePhase.RELEASE_CANDIDATE_DRAFT: base_dir = get_unfinished_dir() @@ -120,9 +120,9 @@ def release_directory_revision(release: sql.Release) -> pathlib.Path | None: ): if (path_revision := release.latest_revision_number) is None: return None - path = get_unfinished_dir() / str(path_project) / str(path_version) / path_revision + path = get_unfinished_dir() / path_project / path_version / path_revision case sql.ReleasePhase.RELEASE: - path = get_finished_dir() / str(path_project) / str(path_version) + path = get_finished_dir() / path_project / path_version # Do not add "case _" here return path @@ -137,9 +137,9 @@ def release_directory_version(release: sql.Release) -> pathlib.Path: | sql.ReleasePhase.RELEASE_CANDIDATE | sql.ReleasePhase.RELEASE_PREVIEW ): - path = get_unfinished_dir() / str(path_project) / str(path_version) + path = get_unfinished_dir() / path_project / path_version case sql.ReleasePhase.RELEASE: - path = get_finished_dir() / str(path_project) / str(path_version) + path = get_finished_dir() / path_project / path_version # Do not add "case _" here return path --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
