diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index 0e303b912b..0efd7b45b9 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -1,5 +1,6 @@ from pathlib import Path from subprocess import CalledProcessError +import os from bbot.modules.templates.github import github @@ -8,60 +9,89 @@ class git_clone(github): produced_events = ["FILESYSTEM"] flags = ["passive", "safe", "slow", "code-enum"] meta = { - "description": "Clone code github repositories", + "description": "Clone or update github repositories safely without exposing tokens", "created_date": "2024-03-08", "author": "@domwhewell-sage", } options = {"api_key": "", "output_folder": ""} options_desc = { "api_key": "Github token", - "output_folder": "Folder to clone repositories to. If not specified, cloned repositories will be deleted when the scan completes, to minimize disk usage.", + "output_folder": ( + "Folder to clone repositories to. " + "If not specified, cloned repositories will be deleted when the scan completes, to minimize disk usage." + ), } deps_apt = ["git"] - scope_distance_modifier = 2 async def setup(self): output_folder = self.config.get("output_folder") - if output_folder: - self.output_dir = Path(output_folder) / "git_repos" - else: - self.output_dir = self.scan.temp_dir / "git_repos" + self.output_dir = Path(output_folder) / "git_repos" if output_folder else self.scan.temp_dir / "git_repos" self.helpers.mkdir(self.output_dir) return await super().setup() async def filter_event(self, event): - if event.type == "CODE_REPOSITORY": - if "git" not in event.tags: - return False, "event is not a git repository" + if event.type == "CODE_REPOSITORY" and "git" not in event.tags: + return False, "event is not a git repository" return True async def handle_event(self, event): repo_url = event.data.get("url") repo_path = await self.clone_git_repository(repo_url) if repo_path: - self.verbose(f"Cloned {repo_url} to {repo_path}") + self.verbose(f"Cloned/updated {repo_url} at {repo_path}") codebase_event = self.make_event({"path": str(repo_path)}, "FILESYSTEM", tags=["git"], parent=event) await self.emit_event( codebase_event, - context=f"{{module}} downloaded git repo at {repo_url} to {{event.type}}: {repo_path}", + context=f"{{module}} cloned/updated git repo at {repo_url} to {{event.type}}: {repo_path}", ) async def clone_git_repository(self, repository_url): - owner = repository_url.split("/")[-2] + # owner and repo name + owner = repository_url.rstrip("/").split("/")[-2] folder = self.output_dir / owner self.helpers.mkdir(folder) + + repo_name = repository_url.rstrip("/").split("/")[-1] + if repo_name.endswith(".git"): + repo_name = repo_name[:-4] + repo_path = folder / repo_name + + env = os.environ.copy() + env["GIT_TERMINAL_PROMPT"] = "0" + if self.api_key: - url = repository_url.replace("https://github.com", f"https://user:{self.api_key}@github.com") + env["GIT_HELPER"] = ( + f'!f() {{ case "$1" in get) ' + f"echo username=x-access-token; " + f"echo password={self.api_key};; " + f'esac; }}; f "$@"' + ) + base_command = [ + "git", + "-c", + "credential.helper=", + "-c", + "credential.useHttpPath=true", + "--config-env=credential.helper=GIT_HELPER", + ] else: - url = repository_url - command = ["git", "-C", folder, "clone", url] + base_command = [] + + # Clone new repo or fetch if exists try: - output = await self.run_process(command, env={"GIT_TERMINAL_PROMPT": "0"}, check=True) + if repo_path.exists(): + # Update existing repo + command = base_command + ["-C", str(repo_path), "fetch", "--all"] + else: + # Clone fresh + command = base_command + ["-C", str(folder), "clone", repository_url] + + await self.run_process(command, env=env, check=True) + except CalledProcessError as e: - self.debug(f"Error cloning {url}. STDERR: {repr(e.stderr)}") + self.debug(f"Error cloning/updating {repository_url}. STDERR: {repr(e.stderr)}") return - folder_name = output.stderr.split("Cloning into '")[1].split("'")[0] - return folder / folder_name + return repo_path