Merge pull request #72 from Malwarebytes/60-export-secrets-by-type

Allow filtering secret type

src/cli.py

@@ -419,7 +419,6 @@ def repositories_archivable(
 
     print(len(res))
     for repo in res:
-
         repo = repo.rstrip("\n")
 
         # 2. get default branch
@@ -756,16 +755,45 @@ def secret_alerts_cli() -> None:
     show_envvar=True,
 )
 @click.option("-o", "--organization", prompt="Organization name", type=str)
+@click.option(
+    "-f",
+    "--secrets-filter",
+    prompt=True,
+    type=click.Choice(
+        [
+            "all",
+            "atlassian_api_token",
+            "slack_incoming_webhook_url",
+            "github_ssh_private_key",
+            "slack_api_token",
+            "mailgun_api_key",
+            "firebase_cloud_messaging_server_key",
+            "jfrog_platform_api_key",
+            "google_api_key",
+            "azure_storage_account_key",
+            "new_relic_license_key",
+            "github_personal_access_token",
+            "sendgrid_api_key",
+            "azure_devops_personal_access_token",
+            "jfrog_platform_access_token",
+            "google_cloud_private_key_id",
+            "google_oauth_access_token",
+        ]
+    ),
+    default="all",
+    hide_input=False,
+)
 def secret_alerts_export(
-    state: str, output_csv: Any, token: str, organization: str
+    state: str, output_csv: Any, token: str, organization: str, secrets_filter: str
 ) -> None:
     """Export secrets to a csv"""
 
-    secrets_list = secrets.export_secrets(state, token, organization)
+    secrets_list = secrets.export_secrets(state, token, organization, secrets_filter)
     for secret in secrets_list:
         output_csv.write(
             f"{secret['state']}, {secret['resolution']}, {secret['resolved_at']}, {secret['repository_full_name']}, {secret['url']}, {secret['secret_type']}, {secret['secret']}\n"
         )
+    print(f"Retrieved {len(secrets_list)} secrets.")
 
 
 ##############
@@ -897,7 +925,6 @@ def roles_add(
     organization: str,
     token: str,
 ) -> None:
-
     if roles.create_role(
         name, description, base_role, permissions, organization, token
     ):
@@ -1051,7 +1078,6 @@ def mass_deploy(
     )
 
     for repo in repos_list:
-
         repo = repo.rstrip("\n")
         issue_secretscanner_res = None
         issue_pushprotection_res = None
@@ -1171,7 +1197,6 @@ def mass_archive(
     repos_list = input_repos_list.readlines()
 
     for repo in repos_list:
-
         repo = repo.rstrip("\n")
 
         click.echo(f"{repo}...", nl=False)
@@ -1214,7 +1239,6 @@ def mass_issue_archive(
     repos_list = input_repos_list.readlines()
 
     for repo in repos_list:
-
         repo = repo.rstrip("\n")
 
         issue_res = issues.create(

src/ghas_cli/utils/network.py

@@ -26,7 +26,6 @@ def get_github_headers(token: str) -> Dict:
 
 
 def check_rate_limit(response: Any) -> bool:
-
     if "0" == response.headers["x-ratelimit-remaining"]:
         reset_time = datetime.fromtimestamp(int(response.headers["x-ratelimit-reset"]))
         print(
@@ -37,7 +36,8 @@ def check_rate_limit(response: Any) -> bool:
         return True
 
     if response.status_code == 403:
-        print("Secondary rate limit reached. Need to wait...")
+        # This can be secondary rate limit or SSO error
+        print(response.json()["message"])
         return True
 
     time.sleep(SLEEP_BETWEEN_REQUESTS)

src/ghas_cli/utils/repositories.py

@@ -179,7 +179,6 @@ def get_org_repositories(
             break
 
         for r in repos.json():
-
             repo = Repository()
             repo.load_json(r, token=token)
             # repo.load_json(r, token=None)
@@ -260,7 +259,6 @@ def archive(organization: str, token: str, repository: str) -> bool:
 def check_dependabot_alerts_enabled(
     token: str, organization: str, repository_name: str
 ) -> bool:
-
     headers = network.get_github_headers(token)
 
     status = requests.get(
@@ -496,7 +494,6 @@ def create_codeql_pr(
     )
 
     for language in languages:
-
         # Workflow config
         lang, template = load_codeql_base64_template(language, default_branch)
         payload = {

src/ghas_cli/utils/secrets.py

@@ -7,7 +7,9 @@
 from . import network
 
 
-def export_secrets(state: str, token: str, organization: str) -> List:
+def export_secrets(
+    state: str, token: str, organization: str, secrets_filter: str
+) -> List:
     """Get all secrets from the organization"""
 
     headers = network.get_github_headers(token)
@@ -22,6 +24,7 @@ def export_secrets(state: str, token: str, organization: str) -> List:
             params=params,
             headers=headers,
         )
+
         if network.check_rate_limit(secrets):
             break
         if secrets.status_code != 200:
@@ -40,8 +43,8 @@ def export_secrets(state: str, token: str, organization: str) -> List:
             s["secret_type"] = secret["secret_type"]
             s["secret"] = secret["secret"]
 
-            secret_list.append(s)
+            if secrets_filter is "all" or s["secret_type"] == secrets_filter:
+                secret_list.append(s)
 
         page += 1
-
     return secret_list

src/ghas_cli/utils/teams.py

@@ -15,7 +15,6 @@ def get_repositories(team_slug: str, organization: str, token: str) -> List:
     page = 1
 
     while True:
-
         params = {"per_page": 100, "page": page}
         repos = requests.get(
             url=f"https://api.github.com/orgs/{organization}/teams/{team_slug}/repos",
@@ -52,7 +51,6 @@ def list(organization: str, token: str) -> str:
     page = 1
 
     while True:
-
         params = {"per_page": 100, "page": page}
 
         teams_res = requests.get(

src/ghas_cli/utils/vulns.py

@@ -17,12 +17,10 @@ def get_codeql_alerts_repo(
     repositories_alerts = {}
 
     for repo in repos:
-
         alerts_repo = []
         page = 1
 
         while True:
-
             params = {"state": "open", "per_page": 100, "page": page}
             alerts = requests.get(
                 url=f"https://api.github.com/repos/{organization}/{repo.name}/code-scanning/alerts",