From f63b7aa690197d4ed50a86d6013e6a3109116a0d Mon Sep 17 00:00:00 2001 From: Miguel Angel SOLINAS Date: Wed, 13 Aug 2025 14:44:43 +0200 Subject: [PATCH 1/6] fix(utils): improve path handling for Windows compatibility Signed-off-by: Miguel Angel SOLINAS --- src/anomalib/utils/path.py | 55 ++++++++++++++++++++++++++++++++++---- 1 file changed, 50 insertions(+), 5 deletions(-) diff --git a/src/anomalib/utils/path.py b/src/anomalib/utils/path.py index c7e3880471..78d9707115 100644 --- a/src/anomalib/utils/path.py +++ b/src/anomalib/utils/path.py @@ -34,7 +34,50 @@ import re from pathlib import Path - +from contextlib import suppress +import os, sys, shutil, subprocess +import stat + +def _is_windows_junction(p: Path) -> bool: + """Return True if path is a directory junction (reparse point mount point).""" + try: + st = p.lstat() + return getattr(st, "st_reparse_tag", 0) == stat.IO_REPARSE_TAG_MOUNT_POINT + except (OSError, AttributeError): + return False + +def _safe_remove_path(p: Path) -> None: + """Remove file/dir/symlink/junction at p without following links.""" + if not os.path.lexists(str(p)): + return + with suppress(FileNotFoundError): + if p.is_symlink(): + p.unlink() + elif _is_windows_junction(p): + os.rmdir(p) + elif p.is_dir(): + shutil.rmtree(p) + else: + p.unlink() + +def _make_latest_windows(latest: Path, target: Path) -> None: + # Clean previous latest (symlink/junction/dir/file) + _safe_remove_path(latest) + + tmp = latest.with_name(latest.name + "_tmp") + _safe_remove_path(tmp) + + # Try junction first (no admin needed), fallback to copy + try: + subprocess.run( + ["cmd", "/c", "mklink", "/J", str(tmp), str(target.resolve())], + check=True, + capture_output=True, + ) + except Exception: + shutil.copytree(target, tmp) + + os.replace(tmp, latest) def create_versioned_dir(root_dir: str | Path) -> Path: """Create a new version directory and update the ``latest`` symbolic link. @@ -100,10 +143,12 @@ def create_versioned_dir(root_dir: str | Path) -> Path: # Update the 'latest' symbolic link to point to the new version directory latest_link_path = root_dir / "latest" - if latest_link_path.is_symlink() or latest_link_path.exists(): - latest_link_path.unlink() - latest_link_path.symlink_to(new_version_dir, target_is_directory=True) - + if sys.platform.startswith("win"): + _make_latest_windows(latest_link_path, new_version_dir) + else: + if latest_link_path.is_symlink() or latest_link_path.exists(): + latest_link_path.unlink() + latest_link_path.symlink_to(new_version_dir, target_is_directory=True) return latest_link_path From 436690f8e6941d9137d7655852b512b119a10b08 Mon Sep 17 00:00:00 2001 From: rajeshgangireddy Date: Thu, 4 Sep 2025 14:54:18 +0200 Subject: [PATCH 2/6] =?UTF-8?q?=F0=9F=90=9B=20fix(utils/path):=20make=20ch?= =?UTF-8?q?anges=20to=20fix=20pre-commit=20issues?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/anomalib/utils/path.py | 40 ++++++++++++++++++++++++++------------ 1 file changed, 28 insertions(+), 12 deletions(-) diff --git a/src/anomalib/utils/path.py b/src/anomalib/utils/path.py index 78d9707115..e626bcb0c7 100644 --- a/src/anomalib/utils/path.py +++ b/src/anomalib/utils/path.py @@ -32,20 +32,29 @@ across different working directories. """ +import os import re -from pathlib import Path +import shutil +import subprocess +import sys from contextlib import suppress -import os, sys, shutil, subprocess -import stat +from pathlib import Path + def _is_windows_junction(p: Path) -> bool: - """Return True if path is a directory junction (reparse point mount point).""" + """Return True if path is a directory junction.""" + if not sys.platform.startswith("win"): + return False + try: - st = p.lstat() - return getattr(st, "st_reparse_tag", 0) == stat.IO_REPARSE_TAG_MOUNT_POINT - except (OSError, AttributeError): + # On Windows, check if it's a directory that's not a symlink + # Junctions appear as directories but resolve to different paths + return p.exists() and p.is_dir() and not p.is_symlink() and p.resolve() != p + except (OSError, RuntimeError): + # Handle cases where path operations fail return False + def _safe_remove_path(p: Path) -> None: """Remove file/dir/symlink/junction at p without following links.""" if not os.path.lexists(str(p)): @@ -54,12 +63,14 @@ def _safe_remove_path(p: Path) -> None: if p.is_symlink(): p.unlink() elif _is_windows_junction(p): - os.rmdir(p) + # Use rmdir for Windows junctions + p.rmdir() elif p.is_dir(): shutil.rmtree(p) else: p.unlink() + def _make_latest_windows(latest: Path, target: Path) -> None: # Clean previous latest (symlink/junction/dir/file) _safe_remove_path(latest) @@ -69,15 +80,20 @@ def _make_latest_windows(latest: Path, target: Path) -> None: # Try junction first (no admin needed), fallback to copy try: - subprocess.run( - ["cmd", "/c", "mklink", "/J", str(tmp), str(target.resolve())], + # nosec B603, B607: Windows-specific command for creating junction + subprocess.run( # noqa: S603 + ["cmd", "/c", "mklink", "/J", str(tmp), str(target.resolve())], # noqa: S607 check=True, capture_output=True, ) - except Exception: + except (subprocess.CalledProcessError, OSError, FileNotFoundError): shutil.copytree(target, tmp) + else: + tmp.replace(latest) + return + + tmp.replace(latest) - os.replace(tmp, latest) def create_versioned_dir(root_dir: str | Path) -> Path: """Create a new version directory and update the ``latest`` symbolic link. From 88eb5edeb2a8ba30165c5d716b759bf47af5efe4 Mon Sep 17 00:00:00 2001 From: Ashwin Vaidya Date: Tue, 9 Sep 2025 09:54:09 +0200 Subject: [PATCH 3/6] Apply suggestion from @Copilot Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Signed-off-by: Ashwin Vaidya --- src/anomalib/utils/path.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/anomalib/utils/path.py b/src/anomalib/utils/path.py index e626bcb0c7..4884a02422 100644 --- a/src/anomalib/utils/path.py +++ b/src/anomalib/utils/path.py @@ -92,9 +92,6 @@ def _make_latest_windows(latest: Path, target: Path) -> None: tmp.replace(latest) return - tmp.replace(latest) - - def create_versioned_dir(root_dir: str | Path) -> Path: """Create a new version directory and update the ``latest`` symbolic link. From 3fd1080f6f68a3d9fa348786dbeb013978d99c02 Mon Sep 17 00:00:00 2001 From: rajeshgangireddy Date: Wed, 10 Sep 2025 15:26:34 +0200 Subject: [PATCH 4/6] =?UTF-8?q?=F0=9F=90=9B=20fix(path):=20replace=20subpr?= =?UTF-8?q?ocess=20call=20with=20Path.symlink=5Fto=20for=20creating=20dire?= =?UTF-8?q?ctory=20junctions=20on=20Windows.=20Yet=20to=20be=20tested.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/anomalib/utils/path.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/src/anomalib/utils/path.py b/src/anomalib/utils/path.py index 4884a02422..4f63ce57c0 100644 --- a/src/anomalib/utils/path.py +++ b/src/anomalib/utils/path.py @@ -35,7 +35,6 @@ import os import re import shutil -import subprocess import sys from contextlib import suppress from pathlib import Path @@ -78,20 +77,25 @@ def _make_latest_windows(latest: Path, target: Path) -> None: tmp = latest.with_name(latest.name + "_tmp") _safe_remove_path(tmp) - # Try junction first (no admin needed), fallback to copy + # Try creating a directory junction using native Python API try: - # nosec B603, B607: Windows-specific command for creating junction - subprocess.run( # noqa: S603 - ["cmd", "/c", "mklink", "/J", str(tmp), str(target.resolve())], # noqa: S607 - check=True, - capture_output=True, - ) - except (subprocess.CalledProcessError, OSError, FileNotFoundError): + # Use Path.symlink_to with target_is_directory=True for directory junction on Windows + # This creates a junction point that doesn't require admin privileges + tmp.symlink_to(target.resolve(), target_is_directory=True) + except (OSError, NotImplementedError): + # Fallback to copying if junction creation fails + # This can happen if the filesystem doesn't support junctions or + # if there are permission issues shutil.copytree(target, tmp) else: + # Only reached if symlink creation succeeded tmp.replace(latest) return + # If we get here, we used the fallback copy method + tmp.replace(latest) + + def create_versioned_dir(root_dir: str | Path) -> Path: """Create a new version directory and update the ``latest`` symbolic link. From 57cedef6b1c0a97f955f6d5c5833738a0cb2ff89 Mon Sep 17 00:00:00 2001 From: rajeshgangireddy Date: Mon, 15 Sep 2025 16:12:01 +0200 Subject: [PATCH 5/6] =?UTF-8?q?=F0=9F=9A=80=20feat(path):=20Attempt=20to?= =?UTF-8?q?=20create=20'latest'=20=20shortcuts=20on=20windows=20machines?= =?UTF-8?q?=20better.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/anomalib/utils/path.py | 167 ++++++++++++++++++++++++++++++++----- 1 file changed, 145 insertions(+), 22 deletions(-) diff --git a/src/anomalib/utils/path.py b/src/anomalib/utils/path.py index 4f63ce57c0..ae42dab7df 100644 --- a/src/anomalib/utils/path.py +++ b/src/anomalib/utils/path.py @@ -32,6 +32,7 @@ across different working directories. """ +import logging import os import re import shutil @@ -39,6 +40,8 @@ from contextlib import suppress from pathlib import Path +logger = logging.getLogger(__name__) + def _is_windows_junction(p: Path) -> bool: """Return True if path is a directory junction.""" @@ -70,30 +73,148 @@ def _safe_remove_path(p: Path) -> None: p.unlink() -def _make_latest_windows(latest: Path, target: Path) -> None: - # Clean previous latest (symlink/junction/dir/file) - _safe_remove_path(latest) +def _validate_windows_path(path: Path) -> bool: + """Validate that a path is safe for use in Windows commands. - tmp = latest.with_name(latest.name + "_tmp") - _safe_remove_path(tmp) + Args: + path: Path to validate + + Returns: + True if path is safe, False otherwise + """ + path_str = str(path) + + # Check for shell metacharacters that could be dangerous + dangerous_chars = {"&", "|", ";", "<", ">", "^", '"', "'", "`", "$", "(", ")", "*", "?", "[", "]", "{", "}"} + # Check for command injection patterns + injection_patterns = ["&&", "||", ";", "&", "|"] + + # Perform all validation checks + if ( + any(char in path_str for char in dangerous_chars) + or any(pattern in path_str for pattern in injection_patterns) + or "\x00" in path_str + or len(path_str) > 260 # Windows MAX_PATH + ): + return False + + # Ensure the path exists and is actually a directory (for target) + # or that its parent exists (for tmp) + try: + return path.is_dir() if path.exists() else path.parent.exists() + except (OSError, ValueError): + return False - # Try creating a directory junction using native Python API + +def _create_windows_junction_native(tmp: Path, target: Path) -> bool: + """Try to create a Windows junction using native Python API. + + Args: + tmp: Temporary path for the junction + target: Target directory to link to + + Returns: + True if successful, False otherwise + """ try: - # Use Path.symlink_to with target_is_directory=True for directory junction on Windows - # This creates a junction point that doesn't require admin privileges tmp.symlink_to(target.resolve(), target_is_directory=True) except (OSError, NotImplementedError): - # Fallback to copying if junction creation fails - # This can happen if the filesystem doesn't support junctions or - # if there are permission issues - shutil.copytree(target, tmp) + _safe_remove_path(tmp) + return False else: - # Only reached if symlink creation succeeded + return True + + +def _create_windows_junction_subprocess(tmp: Path, target: Path) -> bool: + """Try to create a Windows junction using subprocess mklink. + + Args: + tmp: Temporary path for the junction + target: Target directory to link to + + Returns: + True if successful, False otherwise + """ + import subprocess + + # Validate paths before subprocess call + if not _validate_windows_path(tmp) or not _validate_windows_path(target): + logger.warning( + "Path validation failed for Windows junction creation. " + "Paths contain potentially unsafe characters or exceed length limits. " + "Falling back to directory with version pointer method. " + "tmp='%s', target='%s'", + tmp, + target, + ) + return False + + # Convert to absolute path strings and re-validate + tmp_str = str(tmp.resolve()) + target_str = str(target.resolve()) + + if not _validate_windows_path(Path(tmp_str)) or not _validate_windows_path(Path(target_str)): + logger.warning( + "Resolved path validation failed for Windows junction creation. " + "Resolved paths contain potentially unsafe characters. " + "Falling back to directory with version pointer method. " + "resolved_tmp='%s', resolved_target='%s'", + tmp_str, + target_str, + ) + return False + + # Execute mklink command + try: + result = subprocess.run( # noqa: S603 + ["cmd", "/c", "mklink", "/J", tmp_str, target_str], # noqa: S607 + capture_output=True, + text=True, + check=False, + timeout=30, + ) + return result.returncode == 0 and tmp.exists() + except (subprocess.SubprocessError, subprocess.TimeoutExpired, OSError): + _safe_remove_path(tmp) + return False + + +def _create_fallback_latest(latest: Path, target: Path) -> None: + """Create fallback 'latest' directory with version pointer file. + + Args: + latest: Path where the latest link should be created + target: Target directory to point to + """ + latest.mkdir(exist_ok=True) + version_file = latest / ".version_pointer" + version_file.write_text(str(target.resolve())) + + +def _make_latest_windows(latest: Path, target: Path) -> None: + """Create a Windows 'latest' link using the best available method. + + Tries in order: native junction, subprocess mklink, fallback directory. + """ + # Clean up any existing latest link + _safe_remove_path(latest) + + # Create temporary path for atomic replacement + tmp = latest.with_name(latest.name + "_tmp") + _safe_remove_path(tmp) + + # Try native Python junction creation + if _create_windows_junction_native(tmp, target): + tmp.replace(latest) + return + + # Try subprocess mklink command + if _create_windows_junction_subprocess(tmp, target): tmp.replace(latest) return - # If we get here, we used the fallback copy method - tmp.replace(latest) + # Final fallback: create directory with version pointer + _create_fallback_latest(latest, target) def create_versioned_dir(root_dir: str | Path) -> Path: @@ -109,8 +230,9 @@ def create_versioned_dir(root_dir: str | Path) -> Path: created if it doesn't exist. Returns: - Path: Path to the ``latest`` symbolic link that points to the newly created - version directory. + Path: Path to the newly created version directory (e.g., ``v1``, ``v2``). + Training should save files to this directory. The ``latest`` symlink + will point to this directory for convenience. Examples: Create first version directory: @@ -118,14 +240,12 @@ def create_versioned_dir(root_dir: str | Path) -> Path: >>> from pathlib import Path >>> version_dir = create_versioned_dir(Path("experiments")) >>> version_dir - PosixPath('experiments/latest') - >>> version_dir.resolve().name # Points to v1 - 'v1' + PosixPath('experiments/v1') Create second version directory: >>> version_dir = create_versioned_dir("experiments") - >>> version_dir.resolve().name # Now points to v2 + >>> version_dir.name 'v2' Note: @@ -166,7 +286,10 @@ def create_versioned_dir(root_dir: str | Path) -> Path: if latest_link_path.is_symlink() or latest_link_path.exists(): latest_link_path.unlink() latest_link_path.symlink_to(new_version_dir, target_is_directory=True) - return latest_link_path + + # Return the versioned directory path, not the latest link + # This ensures training saves to the versioned directory directly + return new_version_dir def convert_to_snake_case(s: str) -> str: From dfaaa975280093f08b3fc251f8bf6ea9aabc3a9a Mon Sep 17 00:00:00 2001 From: rajeshgangireddy Date: Mon, 15 Sep 2025 21:46:14 +0200 Subject: [PATCH 6/6] =?UTF-8?q?=F0=9F=90=9B=20fix(path):=20Refactor=20Wind?= =?UTF-8?q?ows=20junction=20handling=20to=20use=20Path.symlink=5Fto=20and?= =?UTF-8?q?=20improve=20fallback=20mechanism=20for=20'latest'=20link=20cre?= =?UTF-8?q?ation.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/anomalib/utils/path.py | 201 ++++++++++++++----------------------- 1 file changed, 74 insertions(+), 127 deletions(-) diff --git a/src/anomalib/utils/path.py b/src/anomalib/utils/path.py index ae42dab7df..a52bdb201f 100644 --- a/src/anomalib/utils/path.py +++ b/src/anomalib/utils/path.py @@ -43,36 +43,6 @@ logger = logging.getLogger(__name__) -def _is_windows_junction(p: Path) -> bool: - """Return True if path is a directory junction.""" - if not sys.platform.startswith("win"): - return False - - try: - # On Windows, check if it's a directory that's not a symlink - # Junctions appear as directories but resolve to different paths - return p.exists() and p.is_dir() and not p.is_symlink() and p.resolve() != p - except (OSError, RuntimeError): - # Handle cases where path operations fail - return False - - -def _safe_remove_path(p: Path) -> None: - """Remove file/dir/symlink/junction at p without following links.""" - if not os.path.lexists(str(p)): - return - with suppress(FileNotFoundError): - if p.is_symlink(): - p.unlink() - elif _is_windows_junction(p): - # Use rmdir for Windows junctions - p.rmdir() - elif p.is_dir(): - shutil.rmtree(p) - else: - p.unlink() - - def _validate_windows_path(path: Path) -> bool: """Validate that a path is safe for use in Windows commands. @@ -106,115 +76,91 @@ def _validate_windows_path(path: Path) -> bool: return False -def _create_windows_junction_native(tmp: Path, target: Path) -> bool: - """Try to create a Windows junction using native Python API. - - Args: - tmp: Temporary path for the junction - target: Target directory to link to - - Returns: - True if successful, False otherwise - """ - try: - tmp.symlink_to(target.resolve(), target_is_directory=True) - except (OSError, NotImplementedError): - _safe_remove_path(tmp) - return False - else: - return True - - -def _create_windows_junction_subprocess(tmp: Path, target: Path) -> bool: - """Try to create a Windows junction using subprocess mklink. - - Args: - tmp: Temporary path for the junction - target: Target directory to link to - - Returns: - True if successful, False otherwise - """ - import subprocess - - # Validate paths before subprocess call - if not _validate_windows_path(tmp) or not _validate_windows_path(target): - logger.warning( - "Path validation failed for Windows junction creation. " - "Paths contain potentially unsafe characters or exceed length limits. " - "Falling back to directory with version pointer method. " - "tmp='%s', target='%s'", - tmp, - target, - ) - return False - - # Convert to absolute path strings and re-validate - tmp_str = str(tmp.resolve()) - target_str = str(target.resolve()) - - if not _validate_windows_path(Path(tmp_str)) or not _validate_windows_path(Path(target_str)): - logger.warning( - "Resolved path validation failed for Windows junction creation. " - "Resolved paths contain potentially unsafe characters. " - "Falling back to directory with version pointer method. " - "resolved_tmp='%s', resolved_target='%s'", - tmp_str, - target_str, - ) +def _is_windows_junction(p: Path) -> bool: + """Return True if path is a directory junction.""" + if not sys.platform.startswith("win"): return False - # Execute mklink command try: - result = subprocess.run( # noqa: S603 - ["cmd", "/c", "mklink", "/J", tmp_str, target_str], # noqa: S607 - capture_output=True, - text=True, - check=False, - timeout=30, - ) - return result.returncode == 0 and tmp.exists() - except (subprocess.SubprocessError, subprocess.TimeoutExpired, OSError): - _safe_remove_path(tmp) + # On Windows, check if it's a directory that's not a symlink + # Junctions appear as directories but resolve to different paths + return p.exists() and p.is_dir() and not p.is_symlink() and p.resolve() != p + except (OSError, RuntimeError): + # Handle cases where path operations fail return False -def _create_fallback_latest(latest: Path, target: Path) -> None: - """Create fallback 'latest' directory with version pointer file. - - Args: - latest: Path where the latest link should be created - target: Target directory to point to - """ - latest.mkdir(exist_ok=True) - version_file = latest / ".version_pointer" - version_file.write_text(str(target.resolve())) +def _safe_remove_path(p: Path) -> None: + """Remove file/dir/symlink/junction at p without following links.""" + if not os.path.lexists(str(p)): + return + with suppress(FileNotFoundError): + if p.is_symlink(): + p.unlink() + elif _is_windows_junction(p): + # Use rmdir for Windows junctions + p.rmdir() + elif p.is_dir(): + shutil.rmtree(p) + else: + p.unlink() def _make_latest_windows(latest: Path, target: Path) -> None: - """Create a Windows 'latest' link using the best available method. - - Tries in order: native junction, subprocess mklink, fallback directory. - """ - # Clean up any existing latest link + # Clean previous latest (symlink/junction/dir/file) _safe_remove_path(latest) - # Create temporary path for atomic replacement tmp = latest.with_name(latest.name + "_tmp") _safe_remove_path(tmp) - # Try native Python junction creation - if _create_windows_junction_native(tmp, target): - tmp.replace(latest) - return - - # Try subprocess mklink command - if _create_windows_junction_subprocess(tmp, target): + # Try creating a directory junction using native Python API + try: + # Use Path.symlink_to with target_is_directory=True for directory junction on Windows + # This creates a junction point that doesn't require admin privileges + tmp.symlink_to(target.resolve(), target_is_directory=True) + except (OSError, NotImplementedError): + # Try using Windows mklink command via subprocess + try: + import subprocess + + # Note: Using subprocess with mklink is safe here as we control + # the command and arguments. This is a standard Windows command. + if not _validate_windows_path(tmp) or not _validate_windows_path(target): + logger.warning( + "Warning: Unsafe characters detected in paths. Falling back to text pointer file for 'latest'.", + ) + msg = f"Unsafe path detected: {tmp} -> {target}" + raise ValueError(msg) + result = subprocess.run( # noqa: S603 + [ # noqa: S607 + "cmd", + "/c", + "mklink", + "/J", + str(tmp), + str(target.resolve()), + ], + capture_output=True, + text=True, + check=False, + ) + + if result.returncode == 0 and tmp.exists(): + tmp.replace(latest) + return + except (subprocess.SubprocessError, OSError): + # Subprocess failed, fall through to fallback + pass + else: + # Only reached if symlink creation succeeded tmp.replace(latest) return - # Final fallback: create directory with version pointer - _create_fallback_latest(latest, target) + # Final fallback: create a text file indicating the latest version + # This preserves the intended behavior without breaking the system + latest.mkdir(exist_ok=True) + version_file = latest / ".version_pointer" + version_file.write_text(str(target.resolve())) def create_versioned_dir(root_dir: str | Path) -> Path: @@ -230,9 +176,8 @@ def create_versioned_dir(root_dir: str | Path) -> Path: created if it doesn't exist. Returns: - Path: Path to the newly created version directory (e.g., ``v1``, ``v2``). - Training should save files to this directory. The ``latest`` symlink - will point to this directory for convenience. + Path: Path to the ``latest`` symbolic link that points to the newly created + version directory. Examples: Create first version directory: @@ -240,12 +185,14 @@ def create_versioned_dir(root_dir: str | Path) -> Path: >>> from pathlib import Path >>> version_dir = create_versioned_dir(Path("experiments")) >>> version_dir - PosixPath('experiments/v1') + PosixPath('experiments/latest') + >>> version_dir.resolve().name # Points to v1 + 'v1' Create second version directory: >>> version_dir = create_versioned_dir("experiments") - >>> version_dir.name + >>> version_dir.resolve().name # Now points to v2 'v2' Note: