feat: add ability to delete or deactivate projects for archived status (

#123)

* feat: add ability to delete or deactivate projects when status in GH is archived

* feat: update deactivate to delete call, ensure repo has archived populated

* feat: add ability to reactivate projects should a GH repo become unarchived

* fix: fix tests

* fix: fix tests

* fix: fix tests

* fix: pylint

* feat: add tests for delete/activation/reactivation

* Change unarchive options to ignore

* fix: fixes from PR comments

README.md

@@ -46,6 +46,11 @@ optional arguments:
   --container {on,off}  scan for container projects, e.g. Dockerfile (on by default)
   --iac {on,off}        scan for IAC manifests (experimental, off by default)
   --code {off}          code analysis is deprecated with off only option
+  --on-archived {ignore,deactivate,delete}
+                        Deletes or deactivates projects associated with archived repos (ignore by default)
+  --on-unarchived {ignore,reactivate}
+                        If there is a deactivated project in Snyk, should the tool reactivate it if the repo is not
+                        archived? (Warning: Use with caution, this will reactivate ALL projects associated with a repo)
   --dry-run             Simulate processing of the script without making changes to Snyk
   --skip-scm-validation
                         Skip validation of the TLS certificate used by the SCM

app/app.py

@@ -21,6 +21,7 @@
     log_audit_large_repo_result
 )
 
+
 def run():
     """Begin application logic"""
     # pylint: disable=too-many-locals, too-many-branches, too-many-statements
@@ -59,7 +60,7 @@ def run():
         is_default_renamed = False
         app_print(snyk_repo.org_name,
                   snyk_repo.full_name,
-                  f"Processing {str(i+1)}/{str(len(snyk_repos))}")
+                  f"Processing {str(i + 1)}/{str(len(snyk_repos))}")
 
         try:
             gh_repo_status = get_gh_repo_status(snyk_repo)
@@ -70,13 +71,13 @@ def run():
         app_print(snyk_repo.org_name,
                   snyk_repo.full_name,
                   f"Github Status {gh_repo_status.response_code}" \
-                      f"({gh_repo_status.response_message}) [{snyk_repo.origin}]")
+                  f"({gh_repo_status.response_message}) [{snyk_repo.origin}]")
 
-        #if snyk_repo does not still exist (removed/404), then log and skip to next repo
-        if gh_repo_status.response_code == 404: # project no longer exists
+        # if snyk_repo does not still exist (removed/404), then log and skip to next repo
+        if gh_repo_status.response_code == 404:  # project no longer exists
             log_potential_delete(snyk_repo.org_name, snyk_repo.full_name)
 
-        elif gh_repo_status.response_code == 200: # project exists and has not been renamed
+        elif gh_repo_status.response_code == 200:  # project exists and has not been renamed
             # if --audit-large-repos is on
             if common.ARGS.audit_large_repos:
                 is_truncated_str = \
@@ -90,12 +91,54 @@ def run():
                 )
                 # move to next repo without processing the rest of the code
                 continue
+
+            # If we've previously deactivated projects, we should activate them again
+            # if the repo becomes "unarchived"
+            if not gh_repo_status.archived and common.ARGS.on_unarchived == "reactivate":
+                for project in snyk_repo.snyk_projects:
+                    if not project["is_monitored"]:
+                        activated_projects = snyk_repo.activate_manifests(common.ARGS.dry_run)
+                        for activated_project in activated_projects:
+                            if not common.ARGS.dry_run:
+                                app_print(snyk_repo.org_name,
+                                          snyk_repo.full_name,
+                                          f"Activated manifest: {activated_project['manifest']}")
+                            else:
+                                app_print(snyk_repo.org_name,
+                                          snyk_repo.full_name,
+                                          f"Would activate manifest: "
+                                          f"{activated_project['manifest']}")
+                        break  # We just needed to check if any one of the projects wasn't active
+
+            if gh_repo_status.archived and common.ARGS.on_archived != "ignore":
+                app_print(snyk_repo.org_name,
+                          snyk_repo.full_name,
+                          f"Repo is archived")
+
+                # Check what archival mode we're running in
+                on_archival_action = common.ARGS.on_archived
+                if on_archival_action == "deactivate":
+                    deleted_projects = snyk_repo.deactivate_manifests(common.ARGS.dry_run)
+                elif on_archival_action == "delete":
+                    deleted_projects = snyk_repo.delete_manifests(common.ARGS.dry_run)
+
+                # And tell the user what has or would have happened
+                for project in deleted_projects:
+                    if not common.ARGS.dry_run:
+                        app_print(snyk_repo.org_name,
+                                  snyk_repo.full_name,
+                                  f"{on_archival_action.capitalize()}d manifest: "
+                                  f"{project['manifest']}")
+                    else:
+                        app_print(snyk_repo.org_name,
+                                  snyk_repo.full_name,
+                                  f"Would {on_archival_action} manifest: {project['manifest']}")
             # snyk has the wrong branch, re-import
-            if gh_repo_status.repo_default_branch != snyk_repo.branch:
+            elif gh_repo_status.repo_default_branch != snyk_repo.branch:
                 app_print(snyk_repo.org_name,
                           snyk_repo.full_name,
-                          f"Default branch name changed from {snyk_repo.branch}" \
-                          f" -> {gh_repo_status.repo_default_branch}")
+                          f"Default branch name changed from {snyk_repo.branch}" f" -> "
+                          f"{gh_repo_status.repo_default_branch}")
                 updated_projects = snyk_repo.update_branch(
                     gh_repo_status.repo_default_branch,
                     common.ARGS.dry_run)
@@ -106,7 +149,7 @@ def run():
                                   f"Monitored branch set to " \
                                   f"{gh_repo_status.repo_default_branch} " \
                                   f"for: {project['manifest']}")
-            else: #find deltas
+            else:  # find deltas
                 app_print(snyk_repo.org_name,
                           snyk_repo.full_name,
                           f"Checking {str(len(snyk_repo.snyk_projects))} " \
@@ -127,7 +170,7 @@ def run():
                           snyk_repo.full_name,
                           "Checking for new manifests in source tree")
 
-                #if not common.ARGS.dry_run:
+                # if not common.ARGS.dry_run:
                 projects_import = snyk_repo.add_new_manifests(common.ARGS.dry_run)
 
                 if isinstance(projects_import, ImportStatus):

app/gh_repo.py

@@ -183,6 +183,7 @@ def get_gh_repo_status(snyk_gh_repo):
     response_message = ""
     response_status_code = ""
     repo_default_branch = ""
+    archived = False
 
     # logging.debug(f"snyk_gh_repo origin: {snyk_gh_repo.origin}")
 
@@ -214,6 +215,7 @@ def get_gh_repo_status(snyk_gh_repo):
         if response.status_code == 200:
             response_message = "Match"
             repo_default_branch = response.json()['default_branch']
+            archived = response.json()['archived']
 
         elif response.status_code == 404:
             response_message = "Not Found"
@@ -231,6 +233,7 @@ def get_gh_repo_status(snyk_gh_repo):
                 repo_new_full_name = follow_response.json()["full_name"]
                 repo_owner = repo_new_full_name.split("/")[0]
                 repo_name = repo_new_full_name.split("/")[1]
+                archived = follow_response.json()['archived']
             else:
                 repo_owner = ""
                 repo_name = ""
@@ -252,6 +255,7 @@ def get_gh_repo_status(snyk_gh_repo):
             snyk_gh_repo["org_id"],
             repo_owner,
             f"{repo_owner}/{repo_name}",
-            repo_default_branch
+            repo_default_branch,
+            archived
         )
     return repo_status

app/models.py

@@ -2,11 +2,13 @@
 from dataclasses import dataclass
 from typing import List
 
+
 @dataclass
 class ImportFile:
     """File being imported"""
     path: str
 
+
 @dataclass
 class PendingDelete:
     """Projects needing deletion"""
@@ -16,6 +18,7 @@ class PendingDelete:
     org_name: str
     pending_repo: str
 
+
 @dataclass
 class ImportStatus:
     """Import job response"""
@@ -29,6 +32,7 @@ class ImportStatus:
     files: List[ImportFile]
     pending_project_deletes: List[PendingDelete]
 
+# pylint: disable=too-many-instance-attributes
 @dataclass
 class GithubRepoStatus:
     """Status of a Github repository"""
@@ -39,4 +43,5 @@ class GithubRepoStatus:
     repo_owner: str
     repo_full_name: str
     repo_default_branch: str
+    archived: bool
 

app/snyk_repo.py

@@ -113,6 +113,58 @@ def delete_stale_manifests(self, dry_run):
                             f" in org {snyk_project['org_id']}")
         return result
 
+    def delete_manifests(self, dry_run):
+        """ delete all snyk projects corresponding to a repo """
+        result = []
+        for snyk_project in self.snyk_projects:
+            # delete project, append on success
+            if not dry_run:
+                try:
+                    app.utils.snyk_helper.delete_snyk_project(snyk_project["id"],
+                                                              snyk_project["org_id"])
+                    result.append(snyk_project)
+                except snyk.errors.SnykNotFoundError:
+                    print(f"    - Project {snyk_project['id']} not found" \
+                        f" in org {snyk_project['org_id']}")
+            else:
+                result.append(snyk_project)
+        return result
+
+
+    def deactivate_manifests(self, dry_run):
+        """ deactivate all snyk projects corresponding to a repo """
+        result = []
+        for snyk_project in [x for x in self.snyk_projects if x["is_monitored"]]:
+            # delete project, append on success
+            if not dry_run:
+                try:
+                    app.utils.snyk_helper.deactivate_snyk_project(snyk_project["id"],
+                                                                  snyk_project["org_id"])
+                    result.append(snyk_project)
+                except snyk.errors.SnykNotFoundError:
+                    print(f"    - Project {snyk_project['id']} not found" \
+                        f" in org {snyk_project['org_id']}")
+            else:
+                result.append(snyk_project)
+        return result
+
+    def activate_manifests(self, dry_run):
+        """ deactivate all snyk projects corresponding to a repo """
+        result = []
+        for snyk_project in [x for x in self.snyk_projects if not x["is_monitored"]]:
+            # delete project, append on success
+            if not dry_run:
+                try:
+                    app.utils.snyk_helper.activate_snyk_project(snyk_project["id"],
+                                                                snyk_project["org_id"])
+                    result.append(snyk_project)
+                except snyk.errors.SnykNotFoundError:
+                    print(f"    - Project {snyk_project['id']} not found" \
+                        f" in org {snyk_project['org_id']}")
+            else:
+                result.append(snyk_project)
+        return result
+
     def update_branch(self, new_branch_name, dry_run):
         """ update the branch for all snyk projects for this repo """
         result = []