Housekeeper-delete-tokens-projects-and-branches-separately (#2018)

* Fixes #2014

* Filter for housekeeper mode

* Quality pass

Author: okorach

cli/housekeeper.py

@@ -31,7 +31,8 @@
 from requests import RequestException
 from cli import options
 import sonar.logging as log
-from sonar import platform, tokens, users, projects, branches, version, errcodes
+from sonar import platform, tokens, users, projects, branches, pull_requests, version, errcodes
+from sonar.util import types
 import sonar.util.constants as c
 import sonar.utilities as util
 import sonar.exceptions as ex
@@ -44,7 +45,7 @@
 def get_project_problems(settings: dict[str, str], endpoint: object) -> list[problem.Problem]:
     """Returns the list of problems that would require housekeeping for a given project"""
     problems = []
-    if settings[PROJ_MAX_AGE] < 90:
+    if settings[PROJ_MAX_AGE] != 0 and settings[PROJ_MAX_AGE] < 90:
         log.error("As a safety measure, can't delete projects more recent than 90 days")
         return problems
 
@@ -90,10 +91,6 @@ def get_user_problems(settings: dict[str, str], endpoint: platform.Platform) ->
 
 def _parse_arguments() -> object:
     """Parses CLI arguments"""
-    _DEFAULT_PROJECT_OBSOLESCENCE = 365
-    _DEFAULT_BRANCH_OBSOLESCENCE = 90
-    _DEFAULT_PR_OBSOLESCENCE = 30
-    _DEFAULT_TOKEN_OBSOLESCENCE = 365
     parser = options.set_common_args("Deletes projects, branches, PR, user tokens not used since a given number of days")
     parser = options.set_output_file_args(parser, allowed_formats=("csv",))
     parser = options.add_thread_arg(parser, "auditing before housekeeping")
@@ -112,16 +109,16 @@ def _parse_arguments() -> object:
         "--projectsMaxAge",
         required=False,
         type=int,
-        default=_DEFAULT_PROJECT_OBSOLESCENCE,
-        help=f"Deletes projects not analyzed since a given number of days, by default {_DEFAULT_PROJECT_OBSOLESCENCE} days",
+        default=0,
+        help="Deletes projects not analyzed since a given number of days",
     )
     parser.add_argument(
         "-B",
         "--branchesMaxAge",
         required=False,
         type=int,
-        default=_DEFAULT_BRANCH_OBSOLESCENCE,
-        help=f"Deletes branches not to be kept and not analyzed since a given number of days, by default {_DEFAULT_BRANCH_OBSOLESCENCE} days",
+        default=0,
+        help="Deletes branches not to be kept and not analyzed since a given number of days",
     )
     parser.add_argument(
         "--keepWhenInactive",
@@ -134,21 +131,21 @@ def _parse_arguments() -> object:
         "--pullrequestsMaxAge",
         required=False,
         type=int,
-        default=_DEFAULT_BRANCH_OBSOLESCENCE,
-        help=f"Deletes pull requests not analyzed since a given number of days, by default {_DEFAULT_PR_OBSOLESCENCE} days",
+        default=0,
+        help="Deletes pull requests not analyzed since a given number of days",
     )
     parser.add_argument(
         "-T",
         "--tokensMaxAge",
         required=False,
         type=int,
-        default=_DEFAULT_TOKEN_OBSOLESCENCE,
-        help=f"Deletes user tokens older than a certain number of days, by default {_DEFAULT_TOKEN_OBSOLESCENCE} days",
+        default=0,
+        help="Deletes user tokens older than a certain number of days",
     )
     return options.parse_and_check(parser=parser, logger_name=TOOL_NAME)
 
 
-def _delete_objects(problems: problem.Problem, mode: str) -> tuple[int, int, int, int, int]:
+def _delete_objects(problems: problem.Problem, mode: str, settings: types.ConfigSettings) -> tuple[int, int, int, int, int]:
     """Deletes objects (that should be housekept)"""
     revoked_token_count = 0
     deleted_projects = {}
@@ -171,12 +168,13 @@ def _delete_objects(problems: problem.Problem, mode: str) -> tuple[int, int, int
                     deleted_loc += loc
             if isinstance(obj, (tokens.UserToken, users.User)) and (mode != "delete" or obj.revoke()):
                 revoked_token_count += 1
-            elif obj.project().key in deleted_projects:
+            elif settings[PROJ_MAX_AGE] > 0 and obj.project().key in deleted_projects:
                 log.info("%s deleted, so no need to delete %s", str(obj.project()), str(obj))
             elif mode != "delete" or obj.delete():
+                log.info("%s to delete", str(obj))
                 if isinstance(obj, branches.Branch):
                     deleted_branch_count += 1
-                else:
+                elif isinstance(obj, pull_requests.PullRequest):
                     deleted_pr_count += 1
 
         except ex.ObjectNotFound:
@@ -211,7 +209,6 @@ def main() -> None:
         settings = {
             "audit.tokens.maxAge": token_age,
             "audit.tokens.maxUnusedAge": 90,
-            # "audit.groups.empty": True,
             PROJ_MAX_AGE: proj_age,
             "audit.projects.branches.maxLastAnalysisAge": branch_age,
             "audit.projects.pullRequests.maxLastAnalysisAge": pr_age,
@@ -232,12 +229,16 @@ def main() -> None:
         op = "to delete"
         if mode == "delete":
             op = "deleted"
-        (deleted_proj, deleted_loc, deleted_branches, deleted_prs, revoked_tokens) = _delete_objects(problems, mode)
-
-        log.info("%d projects older than %d days (%d LoCs) %s", deleted_proj, proj_age, deleted_loc, op)
-        log.info("%d branches older than %d days %s", deleted_branches, branch_age, op)
-        log.info("%d pull requests older than %d days %s", deleted_prs, pr_age, op)
-        log.info("%d tokens older than %d days %s", revoked_tokens, token_age, "revoked" if mode == "deleted" else "to revoke")
+        (deleted_proj, deleted_loc, deleted_branches, deleted_prs, revoked_tokens) = _delete_objects(problems, mode, settings=settings)
+
+        if proj_age > 0:
+            log.info("%d projects older than %d days (%d LoCs) %s", deleted_proj, proj_age, deleted_loc, op)
+        if branch_age > 0:
+            log.info("%d branches older than %d days %s", deleted_branches, branch_age, op)
+        if pr_age > 0:
+            log.info("%d pull requests older than %d days %s", deleted_prs, pr_age, op)
+        if token_age > 0:
+            log.info("%d tokens older than %d days %s", revoked_tokens, token_age, "revoked" if mode == "deleted" else "to revoke")
 
     except (PermissionError, FileNotFoundError) as e:
         util.final_exit(errcodes.OS_ERROR, f"OS error while housekeeping: {str(e)}")

sonar/audit/sonar-audit.properties

@@ -323,9 +323,11 @@ audit.qualityProfiles.maxPerLanguage = 5
 audit.users.maxLoginAge = 180
 
 # Audit for days after which a token should be revoked (and potentially renewed)
+# Set to 0 to turn off this audit check
 audit.tokens.maxAge = 90
 
 # Audit  for days after which an unused token should be revoked (and potentially renewed)
+# Set to 0 to turn off this audit check
 audit.tokens.maxUnusedAge = 30
 
 # Comma separated list of SonarQube users whose tokens are not considered for expiration

sonar/branches.py

@@ -363,13 +363,14 @@ def __audit_last_analysis(self, audit_settings: types.ConfigSettings) -> list[Pr
         if self.is_main():
             log.info("%s is main (not purgeable)", str(self))
             return []
-        if (age := util.age(self.last_analysis())) is None:
+        if (max_age := audit_settings.get("audit.projects.branches.maxLastAnalysisAge", 30)) == 0:
             log.debug("%s last analysis audit is disabled, skipped...", str(self))
             return []
-        max_age = audit_settings.get("audit.projects.branches.maxLastAnalysisAge", 30)
+        if (age := util.age(self.last_analysis())) is None:
+            log.warning("%s: Can't get last analysis date for audit, skipped", str(self))
+            return []
         preserved = audit_settings.get("audit.projects.branches.keepWhenInactive", None)
         problems = []
-        log.info("EVALUATING %s age %d greater than %d days and not matches '%s'", str(self), age, max_age, preserved)
         if preserved is not None and age > max_age and not re.match(rf"^{preserved}$", self.name):
             log.info("%s age %d greater than %d days and not matches '%s'", str(self), age, max_age, preserved)
             problems.append(Problem(get_rule(RuleId.BRANCH_LAST_ANALYSIS), self, str(self), age))
@@ -392,7 +393,10 @@ def audit(self, audit_settings: types.ConfigSettings) -> list[Problem]:
             return []
         log.debug("Auditing %s", str(self))
         try:
-            return self.__audit_last_analysis(audit_settings) + self.__audit_never_analyzed() + self._audit_component(audit_settings)
+            if audit_settings.get(c.AUDIT_MODE_PARAM, "") == "housekeeper":
+                return self.__audit_last_analysis(audit_settings)
+            else:
+                return self.__audit_last_analysis(audit_settings) + self.__audit_never_analyzed() + self._audit_component(audit_settings)
         except Exception as e:
             log.error("%s while auditing %s, audit skipped", util.error_msg(e), str(self))
         return []

sonar/components.py

@@ -288,7 +288,7 @@ def get_ai_code_assurance(self) -> Optional[str]:
 
     def _audit_bg_task(self, audit_settings: types.ConfigSettings) -> list[Problem]:
         """Audits project background tasks"""
-        if audit_settings.get("audit.mode", "") == "housekeeper":
+        if audit_settings.get(c.AUDIT_MODE_PARAM, "") == "housekeeper":
             return []
         # Cutting short if background task audit is disabled because getting last task is costly
         if (

sonar/projects.py

@@ -417,6 +417,8 @@ def __audit_last_analysis(self, audit_settings: types.ConfigSettings) -> list[Pr
         problems = []
         age = util.age(self.last_analysis(include_branches=True), True)
         if age is None:
+            if audit_settings.get(c.AUDIT_MODE_PARAM, "") == "housekeeper":
+                return problems
             if not audit_settings.get("audit.projects.neverAnalyzed", True):
                 log.debug("Auditing of never analyzed projects is disabled, skipping")
             else:
@@ -457,19 +459,11 @@ def __audit_branches(self, audit_settings: types.ConfigSettings) -> list[Problem
         return problems
 
     def __audit_pull_requests(self, audit_settings: types.ConfigSettings) -> list[Problem]:
-        """Audits project pul requests
+        """Audits project pull requests
 
-        :param audit_settings: Settings (thresholds) to raise problems
-        :type audit_settings: dict
-        :return: List of problems found, or empty list
-        :rtype: list[Problem]
+        :param ConfigSettings audit_settings: Settings (thresholds) to raise problems
+        :return: List of problems found
         """
-        if audit_settings.get(c.AUDIT_MODE_PARAM, "") == "housekeeper":
-            return []
-        max_age = audit_settings.get("audit.projects.pullRequests.maxLastAnalysisAge", 30)
-        if max_age == 0:
-            log.debug("Auditing of pull request last analysis age is disabled, skipping...")
-            return []
         problems = []
         for pr in self.pull_requests().values():
             problems += pr.audit(audit_settings)
@@ -650,16 +644,16 @@ def audit(self, audit_settings: types.ConfigSettings) -> list[Problem]:
         problems = []
         try:
             problems = self.__audit_last_analysis(audit_settings)
-            problems += self.audit_visibility(audit_settings)
-            problems += self.__audit_binding_valid(audit_settings)
-            # Skip language audit, as this can be problematic
-            # problems += self.__audit_languages(audit_settings)
             if audit_settings.get(c.AUDIT_MODE_PARAM, "") != "housekeeper":
+                problems += self.audit_visibility(audit_settings)
+                problems += self.__audit_binding_valid(audit_settings)
+                # Skip language audit, as this can be problematic
+                # problems += self.__audit_languages(audit_settings)
                 problems += self.permissions().audit(audit_settings)
 
-            problems += self.__audit_scanner(audit_settings)
-            problems += self._audit_component(audit_settings)
-            problems += self.__audit_key_pattern(audit_settings)
+                problems += self.__audit_scanner(audit_settings)
+                problems += self._audit_component(audit_settings)
+                problems += self.__audit_key_pattern(audit_settings)
             if self.endpoint.edition() != c.CE and audit_settings.get("audit.project.branches", True):
                 problems += self.__audit_branches(audit_settings)
                 problems += self.__audit_pull_requests(audit_settings)

sonar/pull_requests.py

@@ -87,10 +87,13 @@ def last_analysis(self) -> datetime:
 
     def audit(self, audit_settings: types.ConfigSettings) -> list[Problem]:
         """Audits the pull request according to the audit settings"""
-        problems = self._audit_component(audit_settings)
-        if (age := util.age(self.last_analysis())) is None:  # Main branch not analyzed yet
+        problems = [] if audit_settings.get(c.AUDIT_MODE_PARAM, "") == "housekeeper" else self._audit_component(audit_settings)
+        if (age := util.age(self.last_analysis())) is None:
+            log.warning("%s: Can't get last analysis date for audit, skipped")
+            return problems
+        if (max_age := audit_settings.get("audit.projects.pullRequests.maxLastAnalysisAge", 30)) == 0:
+            log.info("%s: Audit of last analysis date is disabled", self)
             return problems
-        max_age = audit_settings.get("audit.projects.pullRequests.maxLastAnalysisAge", 30)
         if age > max_age:
             problems.append(Problem(get_rule(RuleId.PULL_REQUEST_LAST_ANALYSIS), self, str(self), age))
         else:

sonar/tokens.py

@@ -94,15 +94,18 @@ def audit(self, settings: types.ConfigSettings, today: Optional[datetime.datetim
         if self.sq_json.get("isExpired", False):
             return [Problem(get_rule(RuleId.TOKEN_EXPIRED), self, str(self))]
         problems = []
-        mode = settings.get("audit.mode", "")
+        mode = settings.get(c.AUDIT_MODE_PARAM, "")
+        max_age = settings.get("audit.tokens.maxAge", 90)
         if not today:
             today = datetime.datetime.now(datetime.timezone.utc).astimezone()
         age = util.age(self.created_at, now=today)
         if mode != "housekeeper" and not self.expiration_date:
             problems.append(Problem(get_rule(RuleId.TOKEN_WITHOUT_EXPIRATION), self, str(self), age))
-        if age > settings.get("audit.tokens.maxAge", 90):
+        if max_age == 0:
+            log.info("%s: Audit of token max age is disabled, skipped")
+        elif age > max_age:
             problems.append(Problem(get_rule(RuleId.TOKEN_TOO_OLD), self, str(self), age))
-        if self.last_connection_date:
+        if self.last_connection_date and mode != "housekeeper":
             last_cnx_age = util.age(self.last_connection_date, now=today)
             if last_cnx_age > settings.get("audit.tokens.maxUnusedAge", 30):
                 problems.append(Problem(get_rule(RuleId.TOKEN_UNUSED), self, str(self), last_cnx_age))

sonar/users.py

@@ -423,7 +423,7 @@ def audit(self, settings: types.ConfigSettings = None) -> list[Problem]:
 
         today = dt.datetime.now(dt.timezone.utc).astimezone()
         problems = [p for t in self.tokens() for p in t.audit(settings=settings, today=today)]
-        if self.last_login:
+        if self.last_login and settings.get(c.AUDIT_MODE_PARAM, "") != "housekeeper":
             age = util.age(self.last_login, now=today)
             if age > settings.get("audit.users.maxLoginAge", 180):
                 problems.append(Problem(get_rule(RuleId.USER_UNUSED), self, str(self), age))