From d6bb185a8d7485cc4a253aed2e12e9baa18446ac Mon Sep 17 00:00:00 2001 From: Sangharsha Date: Sat, 23 Aug 2025 22:24:29 +0545 Subject: [PATCH 1/2] Fix: Fix with GIT_ASKPASS for .git/config token --- bbot/modules/git_clone.py | 49 +++++++++++++++++++++++++++++++-------- 1 file changed, 39 insertions(+), 10 deletions(-) diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index 0e303b912b..066176d2f0 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -1,5 +1,7 @@ from pathlib import Path from subprocess import CalledProcessError +import os +import tempfile from bbot.modules.templates.github import github @@ -8,7 +10,7 @@ class git_clone(github): produced_events = ["FILESYSTEM"] flags = ["passive", "safe", "slow", "code-enum"] meta = { - "description": "Clone code github repositories", + "description": "Clone code github repositories safely without exposing tokens", "created_date": "2024-03-08", "author": "@domwhewell-sage", } @@ -42,7 +44,9 @@ async def handle_event(self, event): repo_path = await self.clone_git_repository(repo_url) if repo_path: self.verbose(f"Cloned {repo_url} to {repo_path}") - codebase_event = self.make_event({"path": str(repo_path)}, "FILESYSTEM", tags=["git"], parent=event) + 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}", @@ -52,16 +56,41 @@ async def clone_git_repository(self, repository_url): owner = repository_url.split("/")[-2] folder = self.output_dir / owner self.helpers.mkdir(folder) + + # env + clone_env = os.environ.copy() + clone_env["GIT_TERMINAL_PROMPT"] = "0" # disable interactive prompts + + askpass_script_path = None if self.api_key: - url = repository_url.replace("https://github.com", f"https://user:{self.api_key}@github.com") - else: - url = repository_url - command = ["git", "-C", folder, "clone", url] + # Create temp GIT_ASKPASS script to supply token safely + askpass_script = tempfile.NamedTemporaryFile(delete=False, mode="w") + askpass_script.write(f'#!/bin/sh\necho "{self.api_key}"\n') + askpass_script.close() + os.chmod(askpass_script.name, 0o700) + clone_env["GIT_ASKPASS"] = askpass_script.name + askpass_script_path = askpass_script.name + + # Clone repository without embedding token in URL + command = ["git", "-C", str(folder), "clone", repository_url] try: - output = await self.run_process(command, env={"GIT_TERMINAL_PROMPT": "0"}, check=True) + await self.run_process(command, env=clone_env, check=True) except CalledProcessError as e: - self.debug(f"Error cloning {url}. STDERR: {repr(e.stderr)}") + self.debug(f"Error cloning {repository_url}. STDERR: {repr(e.stderr)}") + if askpass_script_path: + os.unlink(askpass_script_path) return - folder_name = output.stderr.split("Cloning into '")[1].split("'")[0] - return folder / folder_name + # Clean .git/config to remove any accidental token + repo_name = repository_url.rstrip("/").split("/")[-1].replace(".git", "") + git_config = folder / repo_name / ".git" / "config" + if git_config.exists() and self.api_key: + text = git_config.read_text() + text = text.replace(self.api_key, "") + git_config.write_text(text) + + # Remove temp GIT_ASKPASS script + if askpass_script_path: + os.unlink(askpass_script_path) + + return folder / repo_name From 99e15b15700282c9621aaf910be69de5448b124c Mon Sep 17 00:00:00 2001 From: Sangharsha Date: Sun, 24 Aug 2025 07:27:38 +0545 Subject: [PATCH 2/2] Fix: With Inline & no temp file --- bbot/modules/git_clone.py | 95 ++++++++++++++++++++------------------- 1 file changed, 48 insertions(+), 47 deletions(-) diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index 066176d2f0..0efd7b45b9 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -1,7 +1,6 @@ from pathlib import Path from subprocess import CalledProcessError import os -import tempfile from bbot.modules.templates.github import github @@ -10,87 +9,89 @@ class git_clone(github): produced_events = ["FILESYSTEM"] flags = ["passive", "safe", "slow", "code-enum"] meta = { - "description": "Clone code github repositories safely without exposing tokens", + "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}") - codebase_event = self.make_event( - {"path": str(repo_path)}, "FILESYSTEM", tags=["git"], parent=event - ) + 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) - # env - clone_env = os.environ.copy() - clone_env["GIT_TERMINAL_PROMPT"] = "0" # disable interactive prompts + 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" - askpass_script_path = None if self.api_key: - # Create temp GIT_ASKPASS script to supply token safely - askpass_script = tempfile.NamedTemporaryFile(delete=False, mode="w") - askpass_script.write(f'#!/bin/sh\necho "{self.api_key}"\n') - askpass_script.close() - os.chmod(askpass_script.name, 0o700) - clone_env["GIT_ASKPASS"] = askpass_script.name - askpass_script_path = askpass_script.name + 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: + base_command = [] - # Clone repository without embedding token in URL - command = ["git", "-C", str(folder), "clone", repository_url] + # Clone new repo or fetch if exists try: - await self.run_process(command, env=clone_env, check=True) - except CalledProcessError as e: - self.debug(f"Error cloning {repository_url}. STDERR: {repr(e.stderr)}") - if askpass_script_path: - os.unlink(askpass_script_path) - return + 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] - # Clean .git/config to remove any accidental token - repo_name = repository_url.rstrip("/").split("/")[-1].replace(".git", "") - git_config = folder / repo_name / ".git" / "config" - if git_config.exists() and self.api_key: - text = git_config.read_text() - text = text.replace(self.api_key, "") - git_config.write_text(text) + await self.run_process(command, env=env, check=True) - # Remove temp GIT_ASKPASS script - if askpass_script_path: - os.unlink(askpass_script_path) + except CalledProcessError as e: + self.debug(f"Error cloning/updating {repository_url}. STDERR: {repr(e.stderr)}") + return - return folder / repo_name + return repo_path