fix: correct repo path handling and directory traversal

Signed-off-by: Vladimir Belousov <[email protected]>

Author: vbelouso

src/exploit_iq_commons/embedding/document_embedding.py

@@ -40,6 +40,7 @@
 from exploit_iq_commons.embedding.go_segmenters_with_methods import GoSegmenterWithMethods
 from exploit_iq_commons.embedding.js_extended_parser import ExtendedJavaScriptSegmenter
 from exploit_iq_commons.embedding.source_code_git_loader import SourceCodeGitLoader
+from vuln_analysis.utils.git_utils import sanitize_git_url_for_path
 from exploit_iq_commons.embedding.transitive_code_searcher_tool import TransitiveCodeSearcher
 from exploit_iq_commons.logging.loggers_factory import LoggingFactory
 
@@ -348,7 +349,11 @@ def get_repo_path(self, source_info: SourceDocumentsInfo):
         Path
             Returns the path to the git repository.
         """
-        return self._git_directory / PurePath(source_info.git_repo)
+        # Sanitize the git repo URL to create a valid filesystem path
+        # Remove protocol separators and path separators that could cause issues
+        # Example: 'https://github.com/RHEcosystemAppEng/vulnerability-analysis' -> 'https.github.com.RHEcosystemAppEng.vulnerability-analysis'
+        sanitized_repo_path = sanitize_git_url_for_path(source_info.git_repo)
+        return self._git_directory / PurePath(sanitized_repo_path)
 
     def collect_documents(self, source_info: SourceDocumentsInfo) -> list[Document]:
         """

src/exploit_iq_commons/embedding/source_code_git_loader.py

@@ -179,18 +179,18 @@ def yield_blobs(self) -> typing.Iterator[Blob]:
         logger.info("Processing %d files in the Git repository at path: '%s'", len(final_files), self.repo_path)
 
         for f in tqdm(final_files):
-
-            file_path = Path(f)
-
-            abs_file_path = base_path / file_path
-
-            rel_file_path = str(file_path)
-
-            metadata = {
-                "source": rel_file_path,
-                "file_path": rel_file_path,
-                "file_name": file_path.name,
-                "file_type": file_path.suffix,
-            }
-
-            yield Blob.from_path(abs_file_path, metadata=metadata)
+            abs_file_path = base_path / f
+            if abs_file_path.is_file():
+                try:
+                    rel_file_path = str(f)
+                    metadata = {
+                        "source": rel_file_path,
+                        "file_path": rel_file_path,
+                        "file_name": abs_file_path.name,
+                        "file_type": abs_file_path.suffix,
+                    }
+                    yield Blob.from_path(abs_file_path, metadata=metadata)
+                except Exception as e:
+                    logger.warning("Failed to read blob for '%s'. Ignoring this file. Error: %s", abs_file_path, e)
+            else:
+                logger.debug("Skipping path as it is a directory, not a file: '%s'", abs_file_path)

src/vuln_analysis/functions/cve_generate_vdbs.py

@@ -202,10 +202,12 @@ async def _arun(message: AgentMorpheusInput) -> AgentMorpheusEngineInput:
             # Replace ref with specific commit hash for each source info
             for si in source_infos:
                 try:
-                    repo = get_repo_from_path(config.base_git_dir, si.git_repo)
+                    # Get the sanitized path from the embedder instance
+                    repo_path = embedder.get_repo_path(si)
+                    repo = get_repo_from_path(str(repo_path.parent), repo_path.name)
                     si.ref = repo.commit().hexsha
                 except ValueError as e:
-                    logger.warning("Failed to get commit hash for %s/%s: %s", config.base_git_dir, si.git_repo, e)
+                    logger.warning("Failed to get commit hash for repo defined in %s: %s", si, e)
                     continue
 
         except Exception as e:

src/vuln_analysis/utils/git_utils.py

@@ -21,6 +21,12 @@
 from git import Repo
 
 
+def sanitize_git_url_for_path(git_url: str) -> str:
+    """Sanitizes a git repo URL to create a valid filesystem path component."""
+    # Example: 'https://github.com/some/repo' -> 'https.github.com.some.repo'
+    return git_url.replace('//', '.').replace('/', '.').replace(':', '')
+
+
 def get_repo_from_path(base_dir: str, git_repo: str = ".git") -> Repo:
     """
     Utility function for getting GitPython `Repo` object representing a Git repository.