Added support to scan for leaked secrets in GitHub

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)