wip feature

Author: spoorcc

.pre-commit-config.yaml

@@ -15,14 +15,14 @@ repos:
     -   id: isort
         name: Sort import
         entry: dfetch
-        args: ['filter', 'isort']
+        args: ['filter','--not-dfetched', 'isort']
         language: system
         types: [file, python]
 
     -   id: black
         name: Black (auto-format)
         entry: dfetch
-        args: ['filter', 'black']
+        args: ['filter', '--not-dfetched', 'black']
         language: system
         types: [file, python]
 
@@ -102,7 +102,7 @@ repos:
         name: codespell
         description: Checks for common misspellings in text files.
         entry: dfetch
-        args: ['filter', 'codespell']
+        args: ['filter', '--not-dfetched','codespell']
         language: python
         # exclude: ^doc/_ext/sphinxcontrib_asciinema/_static/asciinema-player_3.12.1.js
         types: [text]

dfetch/commands/filter.py

@@ -7,6 +7,7 @@
 import argparse
 import os
 import sys
+from enum import Enum
 from pathlib import Path
 from typing import Optional
 
@@ -20,6 +21,14 @@
 logger = get_logger(__name__)
 
 
+class FilterType(Enum):
+    """Types of filtering."""
+
+    BLOCK_ONLY_PATH_TRAVERSAL = 0
+    BLOCK_IF_INSIDE = 1
+    BLOCK_IF_OUTSIDE = 2
+
+
 class Filter(dfetch.commands.command.Command):
     """Filter files based on flags and pass on any command.
 
@@ -33,11 +42,19 @@ def create_menu(subparsers: dfetch.commands.command.SubparserActionType) -> None
         """Add the parser menu for this action."""
         parser = dfetch.commands.command.Command.parser(subparsers, Filter)
         parser.add_argument(
-            "--in-manifest",
-            "-i",
+            "--dfetched",
+            "-D",
+            action="store_true",
+            default=True,
+            help="Keep files that came here by dfetching them.",
+        )
+
+        parser.add_argument(
+            "--not-dfetched",
+            "-N",
             action="store_true",
             default=False,
-            help="Keep files that came here through the manifest.",
+            help="Keep files that did not came here by dfetching them.",
         )
 
         parser.add_argument(
@@ -60,83 +77,128 @@ def __call__(self, args: argparse.Namespace) -> None:
         """Perform the filter."""
         if not args.verbose:
             dfetch.log.set_level("ERROR")
-        manifest = dfetch.manifest.manifest.get_manifest()
 
-        pwd = Path.cwd()
+        argument_list = self._get_arguments(args)
+
+        manifest = dfetch.manifest.manifest.get_manifest()
         topdir = Path(manifest.path).parent
-        with in_directory(topdir):
 
-            project_paths = {
+        resolved_args = self._resolve_args(argument_list, topdir)
+
+        with in_directory(topdir):
+            abs_project_paths = {
                 Path(project.destination).resolve() for project in manifest.projects
             }
 
-            input_list = self._determine_input_list(args)
-            block_inside, block_outside = self._filter_files(
-                pwd, topdir, project_paths, input_list
-            )
-
-        blocklist = block_outside if args.in_manifest else block_inside
+        if args.dfetched and not args.not_dfetched:
+            block_type = FilterType.BLOCK_IF_OUTSIDE
+        elif args.not_dfetched:
+            block_type = FilterType.BLOCK_IF_INSIDE
+        else:
+            block_type = FilterType.BLOCK_ONLY_PATH_TRAVERSAL
 
-        filtered_args = [arg for arg in input_list if arg not in blocklist]
+        filtered_args = self._filter_args(
+            topdir, resolved_args, abs_project_paths, block_type
+        )
 
         if args.cmd:
             run_on_cmdline_uncaptured(logger, [args.cmd] + filtered_args)
         else:
             print(os.linesep.join(filtered_args))
 
-    def _determine_input_list(self, args: argparse.Namespace) -> list[str]:
-        """Determine list of inputs to process."""
-        input_list: list[str] = list(str(arg) for arg in args.args)
-        if not sys.stdin.isatty():
-            input_list += list(str(arg).strip() for arg in sys.stdin.readlines())
+    def _filter_args(
+        self,
+        topdir: Path,
+        resolved_args: dict[str, Optional[Path]],
+        abs_project_paths: set[Path],
+        block: FilterType,
+    ) -> list[str]:
+        blocklist = self._filter_files(
+            topdir,
+            abs_project_paths,
+            {path for path in resolved_args.values() if path},
+            block,
+        )
 
-        # If no input from stdin or args loop over all files
-        if not input_list:
-            input_list = list(
-                str(file) for file in Path(".").rglob("*") if file.is_file()
+        filtered_args = [
+            arg for arg in resolved_args.keys() if resolved_args[arg] not in blocklist
+        ]
+
+        return filtered_args
+
+    def _resolve_args(
+        self, argument_list: list[str], topdir: Path
+    ) -> dict[str, Optional[Path]]:
+        resolved_args: dict[str, Optional[Path]] = {}
+        if argument_list:
+            for argument in argument_list:
+                path_obj = Path(argument.strip())
+                resolved_args[argument] = (
+                    path_obj.resolve() if path_obj.exists() else None
+                )
+        else:
+            if not argument_list:
+                resolved_args = {
+                    str(file): file.resolve()
+                    for file in topdir.rglob("*")
+                    if ".git" not in file.parts
+                }
+
+        return resolved_args
+
+    def _get_arguments(self, args: argparse.Namespace) -> list[str]:
+        argument_list: list[str] = list(str(arg) for arg in args.args)
+        if not sys.stdin.isatty():
+            argument_list.extend(
+                non_empty_line for line in sys.stdin if (non_empty_line := line.strip())
             )
 
-        return input_list
+        return argument_list
 
     def _filter_files(
-        self, pwd: Path, topdir: Path, project_paths: set[Path], input_list: list[str]
-    ) -> tuple[list[str], list[str]]:
-        """Filter files in input_set in files in one of the project_paths or not."""
-        block_inside: list[str] = []
-        block_outside: list[str] = []
-
-        for path_or_arg in input_list:
-            arg_abs_path = Path(pwd / path_or_arg.strip()).resolve()
-            if not arg_abs_path.exists():
-                logger.print_info_line(path_or_arg.strip(), "not a file / dir")
-                continue
+        self,
+        topdir: Path,
+        paths: set[Path],
+        input_paths: set[Path],
+        block: FilterType = FilterType.BLOCK_IF_OUTSIDE,
+    ) -> list[Path]:
+        """Filter files in input_set in files in one of the paths or not."""
+        blocklist: list[Path] = []
+
+        for abs_path in input_paths:
             try:
-                arg_abs_path.relative_to(topdir)
+                abs_path.relative_to(topdir)
             except ValueError:
-                logger.print_info_line(path_or_arg.strip(), "outside project")
-                block_inside.append(path_or_arg)
-                block_outside.append(path_or_arg)
+                logger.print_info_line(str(abs_path), "outside project")
+                blocklist.append(abs_path)
+                continue
+
+            if block == FilterType.BLOCK_ONLY_PATH_TRAVERSAL:
                 continue
 
-            containing_dir = self._file_in_project(arg_abs_path, project_paths)
+            containing_dir = self._is_file_contained_in_any_path(abs_path, paths)
 
             if containing_dir:
-                block_inside.append(path_or_arg)
                 logger.print_info_line(
-                    path_or_arg.strip(), f"inside project ({containing_dir})"
+                    str(abs_path), f"inside project ({containing_dir})"
                 )
+                if block == FilterType.BLOCK_IF_INSIDE:
+                    blocklist.append(abs_path)
             else:
-                block_outside.append(path_or_arg)
-                logger.print_info_line(path_or_arg.strip(), "not inside any project")
+                logger.print_info_line(str(abs_path), "not inside any project")
+                if block == FilterType.BLOCK_IF_OUTSIDE:
+                    blocklist.append(abs_path)
 
-        return block_inside, block_outside
+        return blocklist
 
-    def _file_in_project(self, file: Path, project_paths: set[Path]) -> Optional[Path]:
-        """Check if a specific file is somewhere in one of the project paths."""
-        for project_path in project_paths:
+    def _is_file_contained_in_any_path(
+        self, file: Path, paths: set[Path]
+    ) -> Optional[Path]:
+        """Check if a specific file is somewhere in one of the paths."""
+        for path in paths:
             try:
-                file.relative_to(project_path)
-                return project_path
+                file.relative_to(path)
+                return path
             except ValueError:
                 continue
         return None

features/filter-projects.feature

@@ -0,0 +1,33 @@
+@wip
+Feature: Filtering file paths before executing a tool
+
+    Projects are dfetched and used in parent projects, users would like to run
+    static analysis tools but ignore externally vendored projects. The dfetch filter
+    command makes it possible to wrap a cal to another tool and filter out any external files.
+    Also it is possible to list all files that are under control of dfetch and this can be used
+    to automate various tasks. Paths outside the top-level directory should be excluded to prevent
+    any path traversal.
+
+    Background:
+        Given a git repository "SomeProject.git"
+        And a fetched and committed MyProject with the manifest
+            """
+            manifest:
+                version: 0.0
+                projects:
+                  - name: SomeProject
+                    url: some-remote-server/SomeProject.git
+            """
+
+    Scenario: Tool receives only managed files
+        When I run "dfetch filter"
+        Then the output shows
+            """
+            /some/dir/MyProject/SomeProject
+            /some/dir/MyProject/SomeProject/README.md
+            /some/dir/MyProject/SomeProject/.dfetch_data.yaml
+            """
+
+#   Scenario: Tool receives only unmanaged files
+
+#   Scenario: Fail on path traversal outside top-level manifest directory

features/steps/generic_steps.py

@@ -25,6 +25,32 @@
 urn_uuid = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
 bom_ref = re.compile(r"BomRef\.[0-9]+\.[0-9]+")
 svn_error = re.compile(r"svn: E\d{6}: .+")
+abs_path = re.compile(r"/some/dir")
+
+
+def make_relative_if_path(
+    some_string: str, base_dir: Union[str, os.PathLike] = "."
+) -> str:
+    """
+    Convert a string to a relative path if it's a path, relative to base_dir.
+    Works even if the path is outside the base_dir.
+
+    Args:
+        some_string (str): String to check.
+        base_dir (str or Path): Base directory for relative path.
+
+    Returns:
+        str: Relative path if s is a path, otherwise original string.
+    """
+    the_path = pathlib.Path(some_string)
+
+    if not the_path.is_absolute():
+        return some_string
+
+    try:
+        return os.path.relpath(str(the_path), base_dir)
+    except ValueError:
+        return some_string
 
 
 def remote_server_path(context):
@@ -84,6 +110,7 @@ def check_content(
                 (iso_timestamp, "[timestamp]"),
                 (urn_uuid, "[urn-uuid]"),
                 (bom_ref, "[bom-ref]"),
+                (abs_path, "."),
             ],
             text=expected,
         )
@@ -98,6 +125,8 @@ def check_content(
             text=actual,
         )
 
+        actual = make_relative_if_path(actual)
+
         assert actual.strip() == expected.strip(), (
             f"Line {line_nr}: Actual >>{actual.strip()}<< != Expected >>{expected.strip()}<<\n"
             f"ACTUAL:\n{''.join(actual_content)}"
@@ -170,6 +199,7 @@ def step_impl(context, path=None):
 @when('I run "dfetch {args}"')
 def step_impl(context, args, path=None):
     """Call a command."""
+    context.cmd_output = ""
     call_command(context, args.split(), path)