From d6bb185a8d7485cc4a253aed2e12e9baa18446ac Mon Sep 17 00:00:00 2001 From: Sangharsha Date: Sat, 23 Aug 2025 22:24:29 +0545 Subject: [PATCH 1/6] 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/6] 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 From 108e03833b3f3177091e0ce2ed4052f854583eec Mon Sep 17 00:00:00 2001 From: liquidsec Date: Mon, 25 Aug 2025 12:58:13 -0400 Subject: [PATCH 3/6] reverted unnecessary changes --- bbot/modules/git_clone.py | 79 ++++++++++++++++----------------------- 1 file changed, 33 insertions(+), 46 deletions(-) diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index 0efd7b45b9..f84fa25f30 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -1,6 +1,5 @@ from pathlib import Path from subprocess import CalledProcessError -import os from bbot.modules.templates.github import github @@ -9,20 +8,18 @@ class git_clone(github): produced_events = ["FILESYSTEM"] flags = ["passive", "safe", "slow", "code-enum"] meta = { - "description": "Clone or update github repositories safely without exposing tokens", + "description": "Clone code github repositories", "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): @@ -40,58 +37,48 @@ 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/updated {repo_url} at {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) await self.emit_event( codebase_event, - context=f"{{module}} cloned/updated git repo at {repo_url} to {{event.type}}: {repo_path}", + context=f"{{module}} cloned git repo at {repo_url} to {{event.type}}: {repo_path}", ) async def clone_git_repository(self, repository_url): - # owner and repo name - owner = repository_url.rstrip("/").split("/")[-2] + owner = repository_url.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: - 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 new repo or fetch if exists + command = ["git", "-C", folder, "clone", repository_url] try: - if repo_path.exists(): - # Update existing repo - command = base_command + ["-C", str(repo_path), "fetch", "--all"] + if self.api_key: + env = { + "GIT_TERMINAL_PROMPT": "0", + "GIT_HELPER": ( + f'!f() {{ case "$1" in get) ' + f"echo username=x-access-token; " + f"echo password={self.api_key};; " + f'esac; }}; f "$@"' + ), + } + command = ( + command[:1] + + [ + "-c", + "credential.helper=", + "-c", + "credential.useHttpPath=true", + "--config-env=credential.helper=GIT_HELPER", + ] + + command[1:] + ) else: - # Clone fresh - command = base_command + ["-C", str(folder), "clone", repository_url] - - await self.run_process(command, env=env, check=True) + env = {"GIT_TERMINAL_PROMPT": "0"} + output = await self.run_process(command, env=env, check=True) except CalledProcessError as e: - self.debug(f"Error cloning/updating {repository_url}. STDERR: {repr(e.stderr)}") + self.debug(f"Error cloning {repository_url}. STDERR: {repr(e.stderr)}") return - return repo_path + folder_name = output.stderr.split("Cloning into '")[1].split("'")[0] + return folder / folder_name From c6ec535aae165c4c0d3dd0df5ec0fb91cdba78c2 Mon Sep 17 00:00:00 2001 From: liquidsec Date: Mon, 25 Aug 2025 14:01:21 -0400 Subject: [PATCH 4/6] github domain check --- bbot/modules/git_clone.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index f84fa25f30..ffcf6267a7 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -51,7 +51,9 @@ async def clone_git_repository(self, repository_url): command = ["git", "-C", folder, "clone", repository_url] try: - if self.api_key: + _, domain = self.helpers.split_domain(self.helpers.urlparse(repository_url).hostname) + # only use the api key if the domain is github.com + if domain == "github.com" and self.api_key: env = { "GIT_TERMINAL_PROMPT": "0", "GIT_HELPER": ( From bb80db22060ea95b51ac74a4ea0326de14a6046f Mon Sep 17 00:00:00 2001 From: liquidsec Date: Mon, 25 Aug 2025 14:07:37 -0400 Subject: [PATCH 5/6] consistent variable naming --- bbot/modules/git_clone.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index ffcf6267a7..b6e8dd7453 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -34,14 +34,14 @@ async def filter_event(self, event): return True async def handle_event(self, event): - repo_url = event.data.get("url") - repo_path = await self.clone_git_repository(repo_url) + repository_url = event.data.get("url") + repo_path = await self.clone_git_repository(repository_url) if repo_path: - self.verbose(f"Cloned {repo_url} to {repo_path}") + self.verbose(f"Cloned {repository_url} to {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}} cloned git repo at {repo_url} to {{event.type}}: {repo_path}", + context=f"{{module}} cloned git repo at {repository_url} to {{event.type}}: {repo_path}", ) async def clone_git_repository(self, repository_url): From da6370be9ee8a08dffa2d90b5e87fa553dbbfad8 Mon Sep 17 00:00:00 2001 From: liquidsec Date: Mon, 25 Aug 2025 15:28:29 -0400 Subject: [PATCH 6/6] fixing non http use case, variable name consistency --- bbot/modules/git_clone.py | 51 +++++++++++++++++++-------------------- 1 file changed, 25 insertions(+), 26 deletions(-) diff --git a/bbot/modules/git_clone.py b/bbot/modules/git_clone.py index b6e8dd7453..2ebf87b6a3 100644 --- a/bbot/modules/git_clone.py +++ b/bbot/modules/git_clone.py @@ -35,13 +35,13 @@ async def filter_event(self, event): async def handle_event(self, event): repository_url = event.data.get("url") - repo_path = await self.clone_git_repository(repository_url) - if repo_path: - self.verbose(f"Cloned {repository_url} to {repo_path}") - codebase_event = self.make_event({"path": str(repo_path)}, "FILESYSTEM", tags=["git"], parent=event) + repository_path = await self.clone_git_repository(repository_url) + if repository_path: + self.verbose(f"Cloned {repository_url} to {repository_path}") + codebase_event = self.make_event({"path": str(repository_path)}, "FILESYSTEM", tags=["git"], parent=event) await self.emit_event( codebase_event, - context=f"{{module}} cloned git repo at {repository_url} to {{event.type}}: {repo_path}", + context=f"{{module}} cloned git repository at {repository_url} to {{event.type}}: {repository_path}", ) async def clone_git_repository(self, repository_url): @@ -50,32 +50,31 @@ async def clone_git_repository(self, repository_url): self.helpers.mkdir(folder) command = ["git", "-C", folder, "clone", repository_url] + env = {"GIT_TERMINAL_PROMPT": "0"} + try: - _, domain = self.helpers.split_domain(self.helpers.urlparse(repository_url).hostname) - # only use the api key if the domain is github.com - if domain == "github.com" and self.api_key: - env = { - "GIT_TERMINAL_PROMPT": "0", - "GIT_HELPER": ( + hostname = self.helpers.urlparse(repository_url).hostname + if hostname and self.api_key: + _, domain = self.helpers.split_domain(hostname) + # only use the api key if the domain is github.com + if domain == "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 "$@"' - ), - } - command = ( - command[:1] - + [ - "-c", - "credential.helper=", - "-c", - "credential.useHttpPath=true", - "--config-env=credential.helper=GIT_HELPER", - ] - + command[1:] - ) - else: - env = {"GIT_TERMINAL_PROMPT": "0"} + ) + command = ( + command[:1] + + [ + "-c", + "credential.helper=", + "-c", + "credential.useHttpPath=true", + "--config-env=credential.helper=GIT_HELPER", + ] + + command[1:] + ) output = await self.run_process(command, env=env, check=True) except CalledProcessError as e: