feat: recognize symbolic links

fileglancer/app.py

@@ -1274,27 +1274,28 @@ async def get_file_metadata(path_name: str, subpath: Optional[str] = Query(''),
                 raise HTTPException(status_code=404 if "not found" in error else 500, detail=error)
 
             try:
-                file_info = filestore.get_file_info(subpath, username)
-                logger.trace(f"File info: {file_info}")
-
-                result = {"info": json.loads(file_info.model_dump_json())}
-
-                if file_info.is_dir:
-                    try:
-                        files = list(filestore.yield_file_infos(subpath, username))
-                        result["files"] = [json.loads(f.model_dump_json()) for f in files]
-                    except PermissionError:
-                        logger.error(f"Permission denied when listing files in directory: {subpath}")
-                        result["files"] = []
-                        result["error"] = "Permission denied when listing directory contents"
-                        return JSONResponse(content=result, status_code=403)
-                    except FileNotFoundError:
-                        logger.error(f"Directory not found during listing: {subpath}")
-                        result["files"] = []
-                        result["error"] = "Directory contents not found"
-                        return JSONResponse(content=result, status_code=404)
-
-                return result
+                with db.get_db_session(settings.db_url) as session:
+                    file_info = filestore.get_file_info(subpath, current_user=username, session=session)
+                    logger.trace(f"File info: {file_info}")
+
+                    result = {"info": json.loads(file_info.model_dump_json())}
+
+                    if file_info.is_dir:
+                        try:
+                            files = list(filestore.yield_file_infos(subpath, current_user=username, session=session))
+                            result["files"] = [json.loads(f.model_dump_json()) for f in files]
+                        except PermissionError:
+                            logger.error(f"Permission denied when listing files in directory: {subpath}")
+                            result["files"] = []
+                            result["error"] = "Permission denied when listing directory contents"
+                            return JSONResponse(content=result, status_code=403)
+                        except FileNotFoundError:
+                            logger.error(f"Directory not found during listing: {subpath}")
+                            result["files"] = []
+                            result["error"] = "Directory contents not found"
+                            return JSONResponse(content=result, status_code=404)
+
+                    return result
 
             except RootCheckError as e:
                 # Path attempts to escape root directory - try to find a valid fsp for this absolute path

fileglancer/filestore.py

@@ -13,7 +13,9 @@
 from typing import Optional, Generator
 from loguru import logger
 
+from .database import find_fsp_from_absolute_path
 from .model import FileSharePath
+from .utils import is_likely_binary
 
 # Default buffer size for streaming file contents
 DEFAULT_BUFFER_SIZE = 8192
@@ -44,16 +46,107 @@ class FileInfo(BaseModel):
     last_modified: Optional[float] = None
     hasRead: Optional[bool] = None
     hasWrite: Optional[bool] = None
+    is_symlink: bool = False
+    symlink_target_fsp: Optional[dict] = None  # {"fsp_name": str, "subpath": str}
+
+    @staticmethod
+    def _safe_readlink(path: str, root_path: Optional[str] = None) -> Optional[str]:
+        """
+        Safely read a symlink target.
+
+        This wrapper centralizes symlink reading with defense-in-depth validation.
+        When root_path is provided, verifies the symlink's parent directory is
+        within the allowed root before calling os.readlink(). This check uses
+        the parent directory (not realpath of the symlink itself) because
+        realpath would resolve the symlink to its target, which may legitimately
+        be outside root for cross-share symlinks.
+        """
+        try:
+            if root_path is not None:
+                root_real = os.path.realpath(root_path)
+                # Check the symlink's parent directory is within root
+                # (don't resolve the symlink itself - that would check the target)
+                parent_real = os.path.realpath(os.path.dirname(path))
+                if not (parent_real == root_real or parent_real.startswith(root_real + os.sep)):
+                    logger.warning(f"Refusing to read symlink outside root: {path}")
+                    return None
+            return os.readlink(path)
+        except OSError as e:
+            logger.warning(f"Failed to read symlink target for {path}: {e}")
+            return None
+
+    @classmethod
+    def _get_symlink_target_fsp(cls, absolute_path: str, is_symlink: bool, session,
+                                root_path: Optional[str]) -> Optional[dict]:
+        """
+        Resolve a symlink target to a file share path.
+
+        Returns a dict with fsp_name and subpath if the target is in a known file share,
+        or None if not a symlink, target not found, or target not in any file share.
+        """
+        if not is_symlink or session is None:
+            return None
+
+        # Read the symlink target safely
+        target = cls._safe_readlink(absolute_path, root_path=root_path)
+        if target is None:
+            return None
+
+        # Resolve to absolute path (relative symlinks need dirname context)
+        if not os.path.isabs(target):
+            target = os.path.join(os.path.dirname(absolute_path), target)
+        target = os.path.abspath(target)
+
+        # Try to find which file share contains this target
+        try:
+            match = find_fsp_from_absolute_path(session, target)
+            if match:
+                fsp, subpath = match
+
+                # Reconstruct the canonical target path from the file share root
+                # and the returned subpath so we only operate within managed roots.
+                if subpath:
+                    validated_target = os.path.realpath(os.path.join(fsp.mount_path, subpath))
+                else:
+                    validated_target = os.path.realpath(fsp.mount_path)
+
+                # Check if the symlink target actually exists within the resolved location.
+                # If it doesn't exist, return None (broken symlink)
+                if not os.path.exists(validated_target):
+                    return None
+
+                return {"fsp_name": fsp.name, "subpath": subpath}
+        except (FileNotFoundError, PermissionError, OSError):
+            # Target doesn't exist or isn't accessible
+            pass
+
+        return None
 
     @classmethod
-    def from_stat(cls, path: str, absolute_path: str, stat_result: os.stat_result, current_user: str = None):
-        """Create FileInfo from os.stat_result"""
+    def from_stat(cls, path: str, absolute_path: str,
+                  lstat_result: os.stat_result, stat_result: os.stat_result,
+                  current_user: str = None, session = None,
+                  root_path: Optional[str] = None):
+        """
+        Create FileInfo from pre-computed stat results.
+
+        Args:
+            path: Relative path within the filestore.
+            absolute_path: Absolute filesystem path (used for basename and symlink resolution).
+            lstat_result: Result of os.lstat() on the path (detects symlinks).
+            stat_result: Result of os.stat() or lstat for broken symlinks.
+            current_user: Username for permission checking (optional).
+            session: Database session for symlink resolution (optional).
+            root_path: Filestore root for defense-in-depth validation in symlink reading (optional).
+        """
         if path is None or path == "":
             raise ValueError("Path cannot be None or empty")
+
+        is_symlink = stat.S_ISLNK(lstat_result.st_mode)
         is_dir = stat.S_ISDIR(stat_result.st_mode)
         size = 0 if is_dir else stat_result.st_size
         # Do not expose the name of the root directory
-        name = '' if path=='.' else os.path.basename(absolute_path)
+        name = '' if path == '.' else os.path.basename(absolute_path)
         permissions = stat.filemode(stat_result.st_mode)
         last_modified = stat_result.st_mtime
 
@@ -76,6 +169,9 @@ def from_stat(cls, path: str, absolute_path: str, stat_result: os.stat_result, c
             hasRead = cls._has_read_permission(stat_result, current_user, owner, group)
             hasWrite = cls._has_write_permission(stat_result, current_user, owner, group)
 
+        # Resolve symlink target to file share path if applicable
+        symlink_target_fsp = cls._get_symlink_target_fsp(absolute_path, is_symlink, session, root_path)
+
         return cls(
             name=name,
             path=path,
@@ -87,7 +183,9 @@ def from_stat(cls, path: str, absolute_path: str, stat_result: os.stat_result, c
             group=group,
             last_modified=last_modified,
             hasRead=hasRead,
-            hasWrite=hasWrite
+            hasWrite=hasWrite,
+            is_symlink=is_symlink,
+            symlink_target_fsp=symlink_target_fsp
         )
 
     @staticmethod
@@ -190,19 +288,76 @@ def _check_path_in_root(self, path: Optional[str]) -> str:
         return full_path
 
 
-    def _get_file_info_from_path(self, full_path: str, current_user: str = None) -> FileInfo:
+    def _get_file_info_from_path(self, full_path: str, current_user: str = None, session = None) -> FileInfo:
         """
         Get the FileInfo for a file or directory at the given path.
+
+        full_path comes from either:
+        1. _check_path_in_root() which validates user input against the root
+        2. os.path.join(verified_directory, entry) where entry is from os.listdir()
+
+        In both cases, the path has been validated or constructed from validated
+        components. We pass full_path (not realpath) to from_stat so that lstat()
+        can detect symlinks. Symlink targets may be outside the root (cross-fileshare
+        symlinks), which is valid - we detect and report them without following.
+
+        All filesystem I/O (lstat/stat) is performed here rather than in
+        FileInfo.from_stat so that path validation and I/O are in the same
+        method, which allows static analysis tools (CodeQL) to see that the
+        path is sanitized before use.
         """
-        stat_result = os.stat(full_path)
-        # Use real paths to avoid /var vs /private/var mismatches on macOS.
         root_real = os.path.realpath(self.root_path)
+
+        # Defense-in-depth: normalize full_path with abspath (resolves ".."
+        # without following symlinks) and verify it is within root before any
+        # filesystem operations. We use abspath (not realpath) because symlinks
+        # may legitimately point to targets outside root for cross-share links.
+        full_path = os.path.abspath(full_path)
+
+        def _is_within_root(p: str) -> bool:
+            return p == root_real or p.startswith(root_real + os.sep)
+
+        # Check the normalized path string is under root (catches .. traversal)
+        if not _is_within_root(full_path):
+            raise RootCheckError(
+                f"Path ({full_path}) is outside root directory ({root_real})",
+                full_path,
+            )
+
+        # Check the resolved parent is under root (catches symlink-based traversal
+        # e.g. /root/data/symlink_to_etc/passwd where symlink_to_etc -> /etc)
+        # Skip when full_path is the root itself, since root's parent is above root.
+        if full_path != root_real:
+            parent_real = os.path.realpath(os.path.dirname(full_path))
+            if not _is_within_root(parent_real):
+                raise RootCheckError(
+                    f"Path ({full_path}) resolves outside root directory ({root_real})",
+                    full_path,
+                )
+
         full_real = os.path.realpath(full_path)
         if full_real == root_real:
             rel_path = '.'
         else:
             rel_path = os.path.relpath(full_real, root_real)
-        return FileInfo.from_stat(rel_path, full_path, stat_result, current_user)
+
+        # Perform all filesystem stat calls here, after validation.
+        lstat_result = os.lstat(full_path)
+        is_symlink = stat.S_ISLNK(lstat_result.st_mode)
+        if is_symlink:
+            try:
+                stat_result = os.stat(full_path)
+            except (FileNotFoundError, PermissionError, OSError) as e:
+                logger.warning(f"Broken symlink detected: {full_path}: {e}")
+                stat_result = lstat_result
+        else:
+            stat_result = os.stat(full_path)
+
+        return FileInfo.from_stat(
+            rel_path, full_path, lstat_result, stat_result,
+            current_user=current_user, session=session,
+            root_path=self.root_path,
+        )
 
 
     def get_root_path(self) -> str:
@@ -228,7 +383,7 @@ def get_absolute_path(self, relative_path: Optional[str] = None) -> str:
         return os.path.abspath(os.path.join(self.root_path, relative_path))
 
 
-    def get_file_info(self, path: Optional[str] = None, current_user: str = None) -> FileInfo:
+    def get_file_info(self, path: Optional[str] = None, current_user: str = None, session = None) -> FileInfo:
         """
         Get the FileInfo for a file or directory at the given path.
 
@@ -237,12 +392,51 @@ def get_file_info(self, path: Optional[str] = None, current_user: str = None) ->
                 May be None, in which case the root directory is used.
             current_user (str): The username of the current user for permission checking.
                 May be None, in which case hasRead and hasWrite will be None.
+            session: Database session for symlink resolution.
+                May be None, in which case symlink_target_fsp will be None.
+
+        Raises:
+            RootCheckError: If path attempts to escape root directory
+        """
+        if path is None or path == "":
+            full_path = self.root_path
+        else:
+            full_path = os.path.join(self.root_path, path)
+        return self._get_file_info_from_path(full_path, current_user, session)
+
+
+    def check_is_binary(self, path: Optional[str] = None, sample_size: int = 4096) -> bool:
+        """
+        Check if a file is likely binary by reading a sample of its contents.
+
+        Args:
+            path (str): The relative path to the file to check.
+                May be None, in which case the root is checked (always returns False for directories).
+            sample_size (int): Number of bytes to read for binary detection. Defaults to 4096.
+
+        Returns:
+            bool: True if the file appears to be binary, False otherwise.
+                Returns False for directories.
 
         Raises:
             ValueError: If path attempts to escape root directory
+            FileNotFoundError: If the file does not exist
+            PermissionError: If the file cannot be read
         """
         full_path = self._check_path_in_root(path)
-        return self._get_file_info_from_path(full_path, current_user)
+
+        # Directories are not binary
+        if os.path.isdir(full_path):
+            return False
+
+        try:
+            with open(full_path, 'rb') as f:
+                sample = f.read(sample_size)
+                return is_likely_binary(sample)
+        except Exception as e:
+            # If we can't read the file, assume it's binary to be safe
+            logger.warning(f"Could not read file sample for binary detection: {e}")
+            return True
 
 
     def check_is_binary(self, path: Optional[str] = None, sample_size: int = 4096) -> bool:
@@ -281,7 +475,7 @@ def check_is_binary(self, path: Optional[str] = None, sample_size: int = 4096) -
             return True
 
 
-    def yield_file_infos(self, path: Optional[str] = None, current_user: str = None) -> Generator[FileInfo, None, None]:
+    def yield_file_infos(self, path: Optional[str] = None, current_user: str = None, session = None) -> Generator[FileInfo, None, None]:
         """
         Yield a FileInfo object for each child of the given path.
 
@@ -290,6 +484,8 @@ def yield_file_infos(self, path: Optional[str] = None, current_user: str = None)
                 May be None, in which case the root directory is listed.
             current_user (str): The username of the current user for permission checking.
                 May be None, in which case hasRead and hasWrite will be None.
+            session: Database session for symlink resolution.
+                May be None, in which case symlink_target_fsp will be None for symlinks.
 
         Raises:
             PermissionError: If the path is not accessible due to permissions.
@@ -304,9 +500,10 @@ def yield_file_infos(self, path: Optional[str] = None, current_user: str = None)
         for entry in entries:
             entry_path = os.path.join(full_path, entry)
             try:
-                yield self._get_file_info_from_path(entry_path, current_user)
-            except (FileNotFoundError, PermissionError, OSError) as e:
-                logger.error(f"Error accessing entry: {entry_path}: {e}")
+                yield self._get_file_info_from_path(entry_path, current_user, session)
+            except PermissionError as e:
+                # Skip files we don't have permission to access
+                logger.error(f"Permission denied accessing entry: {entry_path}: {e}")
                 continue