Merge pull request #32 from spark1security/github-scan-rebased

Added support to scan for leaked secrets in GitHub

Author: blupants

README.md

@@ -11,7 +11,7 @@
 
 
 # n0s1 - Secret Scanner
-n0s1 ([pronunciation](https://en.wiktionary.org/wiki/nosy#Pronunciation)) is a secret scanner for Slack, Jira, Confluence, Asana, Wrike, Linear and Zendesk. It scans all channels/tickets/items/issues within the chosen platform in search of any leaked secrets in the titles, bodies, messages and comments. It is open-source and it can be easily extended to support scanning many others ticketing and messaging platforms.
+n0s1 ([pronunciation](https://en.wiktionary.org/wiki/nosy#Pronunciation)) is a secret scanner for Slack, Jira, Confluence, Asana, Wrike, Linear, Zendesk and GitHub. It scans all channels/tickets/items/issues within the chosen platform in search of any leaked secrets in the titles, bodies, messages and comments. It is open-source and it can be easily extended to support scanning many others ticketing and messaging platforms.
 
 These secrets are identified by comparing them against an adaptable configuration file named [regex.yaml](https://github.com/spark1security/n0s1/blob/main/src/n0s1/config/regex.yaml). Alternative TOML format is also supported: [regex.toml](https://github.com/spark1security/n0s1/blob/main/src/n0s1/config/regex.toml). The scanner specifically looks for sensitive information, which includes:
 * Github Personal Access Tokens
@@ -30,6 +30,7 @@ These secrets are identified by comparing them against an adaptable configuratio
 * [Wrike](https://www.wrike.com)
 * [Linear](https://linear.app/)
 * [Zendesk](https://www.zendesk.com/)
+* [GitHub](https://github.com/)
 
 ### Install
 ```bash

requirements.txt

@@ -5,6 +5,7 @@ pyyaml
 atlassian-python-api
 asana==3.2.2
 zenpy
+PyGithub
 WrikePy
 BeautifulSoup4
 slack_sdk

setup.py

@@ -27,7 +27,7 @@ def get_version():
 setup(
  name="n0s1",
  version=get_version(),
- description="Secret Scanner for Slack, Jira, Confluence, Asana, Wrike, Linear and Zendesk. Prevent credential leaks with n0s1.",
+ description="Secret Scanner for Slack, Jira, Confluence, Asana, Wrike, Linear, Zendesk and GitHub. Prevent credential leaks with n0s1.",
  long_description=long_description,
  long_description_content_type="text/markdown",
  url="https://spark1.us/n0s1",
@@ -48,7 +48,7 @@ def get_version():
  "Programming Language :: Python :: 3.12",
  "Programming Language :: Python :: 3.13",
  ], # Classifiers help users find your project by categorizing it https://pypi.org/classifiers/
- keywords="security, cybersecurity, scanner, secret scanner, secret leak, data leak, Slack, Jira, Confluence, Asana, Wrike, Linear, Zendesk, security scanner, data loss prevention",
+ keywords="security, cybersecurity, scanner, secret scanner, secret leak, data leak, Slack, Jira, Confluence, Asana, Wrike, Linear, Zendesk, GitHub, security scanner, data loss prevention",
  package_dir={"": "src"},
  packages=find_packages(where="src"),
  python_requires=">=3.9, <4",

src/n0s1/__init__.py

@@ -1 +1 @@
-__version__ = "1.0.25"
+__version__ = "1.0.26"

src/n0s1/controllers/asana_controller.py

@@ -131,7 +131,7 @@ def get_mapping(self, levels=-1, limit=None):
  "name": t.get("name", ""),
  "stories": {}
  }
- if len(project_gid) > 0:
+ if len(task_gid) > 0:
  map_data["workspaces"][workspace_gid]["projects"][project_gid]["tasks"][task_gid] = t_item
  if levels > 0 and levels <= 3:
  continue

src/n0s1/controllers/github_controller.py

@@ -0,0 +1,247 @@
+import logging
+
+
+try:
+ from . import hollow_controller as hollow_controller
+except Exception:
+ import n0s1.controllers.hollow_controller as hollow_controller
+
+
+class GitHubController(hollow_controller.HollowController):
+ def __init__(self):
+ super().__init__()
+ self._client = None
+
+ def set_config(self, config=None):
+ super().set_config(config)
+ from github import Github
+ TOKEN = config.get("token", "")
+ self._client = Github(TOKEN)
+ self._owner = config.get("owner", "")
+ self._repo = config.get("repo", "")
+ return self.is_connected()
+
+ def get_name(self):
+ return "GitHub"
+
+ def is_connected(self):
+ if self._client:
+ if user := self._client.get_user():
+ self.log_message(f"Logged to {self.get_name()} as {user}")
+ return True
+ else:
+ self.log_message(f"Unable to connect to {self.get_name()}. Check your credentials.", logging.ERROR)
+ return False
+ return False
+
+ def _get_repo_fullname(self, repo_name):
+ owner = self._owner
+ repo_fullname = ""
+
+ git_suffix = ".git"
+ suffix_len = len(git_suffix)
+ repo_name_len = len(repo_name)
+ git_url_suffix = repo_name.lower()[repo_name_len-suffix_len:]
+ if git_url_suffix == git_suffix:
+ repo_fullname = repo_name.lower().replace("git://github.com/", "")
+ repo_fullname = repo_fullname[:-suffix_len]
+ return repo_fullname
+
+ if not owner or len(owner) <= 0:
+ owner = self._client.get_user().login
+ repo_fullname = owner + "/" + repo_name
+ return repo_fullname
+
+ def _get_repo_obj(self, repo_gid=None):
+ if self._repo and len(self._repo) > 0:
+ repo_name = self._get_repo_fullname(self._repo)
+ return self._client.get_repo(repo_name)
+ full_name = self._get_repo_fullname(repo_gid)
+ return self._client.get_repo(full_name)
+
+ def _get_repos(self):
+ owner = {}
+ repos = []
+
+ if self._scan_scope:
+ owner = self._scan_scope.get("owner", {})
+ for key in self._scan_scope.get("repos", {}):
+ repos.append(key)
+ if len(repos) > 0:
+ return repos, owner
+ self.connect()
+ if self._owner and len(self._owner) > 0:
+ from github import UnknownObjectException
+ try:
+ org = self._client.get_organization(self._owner)
+ repos = org.get_repos()
+ owner = {"type": "org", "name": self._owner}
+ except UnknownObjectException as e:
+ try:
+ message = f"Unable to get ORG {self._owner} as owner: {e}"
+ self.log_message(message, logging.WARNING)
+ message = f"Trying to get user {self._owner} as owner..."
+ self.log_message(message, logging.WARNING)
+ repos = self._client.get_user(self._owner).get_repos()
+ owner = {"type": "user", "name": self._owner}
+ except Exception as e:
+ message = f"Unable to get repos from {self._owner}: {e}"
+ self.log_message(message, logging.ERROR)
+ else:
+ user = self._client.get_user()
+ repos = user.get_repos()
+ owner = {"type": "authenticated_user", "name": user.login}
+
+ if self._repo and len(self._repo) > 0:
+ for r in repos:
+ if r.name.lower() == self._repo.lower():
+ return [r], owner
+ return [], owner
+
+ return repos, owner
+
+ def _get_branches(self, repo_gid, limit=None):
+ branches = []
+ if self._scan_scope:
+ branches = self._scan_scope.get("repos", {}).get(repo_gid, {}).get("branches", {})
+ if len(branches) > 0:
+ return branches
+ self.connect()
+ repo_obj = self._get_repo_obj(repo_gid)
+ if repo_obj:
+ branches = repo_obj.get_branches()
+ return branches
+
+ def _get_files(self, repo_gid, branch_gid, limit=None):
+ files = []
+ files_content = []
+ if self._scan_scope:
+ files = self._scan_scope.get("repos", {}).get(repo_gid, {}).get("branches", {}).get(branch_gid, {}).get("files", [])
+ if len(files) > 0:
+ return files, files_content
+ self.connect()
+ repo_obj = self._get_repo_obj(repo_gid)
+ if repo_obj:
+ files = []
+ try:
+ contents = repo_obj.get_contents("", ref=branch_gid)
+ while contents:
+ file_content = contents.pop(0)
+ if file_content.type == "dir":
+ contents.extend(repo_obj.get_contents(file_content.path, ref=branch_gid))
+ else:
+ files.append(file_content.path)
+ files_content.append(file_content)
+ except Exception as e:
+ message = f"Error listing files from branch {branch_gid}: {e}"
+ self.log_message(message, logging.ERROR)
+ return files, files_content
+
+ def get_mapping(self, levels=-1, limit=None):
+ if not self._client:
+ return {}
+ repos, owner = self._get_repos()
+ map_data = {"owner": owner, "repos": {}}
+ if repos:
+ for repo in repos:
+ repo_gid = repo.git_url
+ message = f"Searching in repository: {repo.html_url}"
+ self.log_message(message, logging.INFO)
+ if len(repo_gid) > 0:
+ r_item = {
+ "gid": repo_gid,
+ "name": repo.name,
+ "branches": {}
+ }
+ map_data["repos"][repo_gid] = r_item
+ if levels > 0 and levels <= 1:
+ continue
+ if branches := self._get_branches(repo_gid, limit):
+ for branch in branches:
+ message = f"Searching in branch: {branch.name}"
+ self.log_message(message, logging.INFO)
+ branch_gid = branch.name
+ b_item = {
+ "gid": branch.commit.sha,
+ "name": branch.name,
+ "files": {}
+ }
+ if len(branch_gid) > 0:
+ map_data["repos"][repo_gid]["branches"][branch_gid] = b_item
+ if levels > 0 and levels <= 2:
+ continue
+ files, files_content = self._get_files(repo_gid, branch_gid)
+ map_data["repos"][repo_gid]["branches"][branch_gid]["files"] = files
+ if levels > 0 and levels <= 3:
+ continue
+ return map_data
+
+ def get_data(self, include_comments=False, limit=None):
+ if not self._client:
+ return {}
+
+ repos, owner = self._get_repos()
+ if repos:
+ for repo in repos:
+ if isinstance(repo, str):
+ repo = self._get_repo_obj(repo)
+ repo_gid = repo.git_url
+ repo_html_url = repo.html_url
+ message = f"Searching in repository: {repo_html_url}"
+ self.log_message(message, logging.INFO)
+
+ # Iterate through each branch
+ for branch in self._get_branches(repo_gid):
+ branch_gid = ""
+ if isinstance(branch, str):
+ branch_gid = branch
+ else:
+ branch_gid = branch.name
+ message = f"Searching in branch: {branch_gid}"
+ self.log_message(message, logging.INFO)
+
+ # Iterate through each file in the branch
+ try:
+ files, files_content = self._get_files(repo_gid, branch_gid)
+ use_preloaded_content = False
+ if len(files_content) > 0:
+ use_preloaded_content = True
+ files = files_content
+ for f in files:
+ try:
+ if use_preloaded_content:
+ file_data = f.decoded_content.decode(errors='ignore')
+ f = f.path
+ else:
+ # Fetch file content
+ file_data = repo.get_contents(f, ref=branch_gid).decoded_content.decode(errors='ignore')
+ url = repo.html_url + f"/blob/{branch_gid}/{f}"
+ file = self.pack_data(file_data, url)
+ yield file
+ except Exception as e:
+ message = f"Error accessing file {f} from branch {branch_gid}: {e}"
+ self.log_message(message, logging.ERROR)
+ except Exception as e:
+ message = f"Error accessing branch {branch_gid}: {e}"
+ self.log_message(message, logging.ERROR)
+
+ def post_comment(self, issue, comment):
+ if not self._client:
+ return False
+ message = f"Unable to post comment to {issue}!"
+ self.log_message(message, logging.ERROR)
+ return False
+
+ def pack_data(self, file_data, url):
+ ticket_data = {
+ "ticket": {
+ "file": {
+ "name": "file",
+ "data": file_data,
+ "data_type": "str"
+ },
+ },
+ "url": url,
+ "issue_id": url
+ }
+ return ticket_data

src/n0s1/controllers/platform_controller.py

@@ -26,6 +26,7 @@ def get_platform(self, platform):
  from . import linear_controller as linear_controller
  from . import asana_controller as asana_controller
  from . import zendesk_controller as zendesk_controller
+ from . import github_controller as github_controller
  from . import wrike_controller as wrike_controller
  from . import slack_controller as slack_controller
 except Exception:
@@ -34,6 +35,7 @@ def get_platform(self, platform):
  import n0s1.controllers.linear_controller as linear_controller
  import n0s1.controllers.asana_controller as asana_controller
  import n0s1.controllers.zendesk_controller as zendesk_controller
+ import n0s1.controllers.github_controller as github_controller
  import n0s1.controllers.wrike_controller as wrike_controller
  import n0s1.controllers.slack_controller as slack_controller
 
@@ -48,6 +50,8 @@ def get_platform(self, platform):
 factory.register_platform("asana_scan", asana_controller.AsanaController)
 factory.register_platform("zendesk", zendesk_controller.ZendeskController)
 factory.register_platform("zendesk_scan", zendesk_controller.ZendeskController)
+factory.register_platform("github", github_controller.GitHubController)
+factory.register_platform("github_scan", github_controller.GitHubController)
 factory.register_platform("wrike", wrike_controller.WrikeController)
 factory.register_platform("wrike_scan", wrike_controller.WrikeController)
 factory.register_platform("slack", slack_controller.SlackController)