Remove deprecated IBM finding plugin (#157)

BREAKING CHANGES: this notifier will be removed.

Seems like ibm_cloud_security_advisor is [now deprecated](https://github.com/ibm-cloud-security/security-advisor-sdk-python/).
It is having issues compatibility issues with Python 3.12.

Author: cletomartin

.flake8

@@ -0,0 +1,8 @@
+[flake8]
+max-line-length = 120
+exclude =
+   .venv
+   .build
+   __pycache__
+   .idea
+   build

.pre-commit-config.yaml

@@ -1,22 +1,22 @@
 repos:
 -   repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v4.4.0
+    rev: v4.6.0
     hooks:
       -   id: trailing-whitespace
       -   id: check-yaml
       -   id: fix-encoding-pragma
           args: ["--remove"]  # Not needed on python3
 -   repo: https://github.com/ambv/black
-    rev: 22.12.0
+    rev: 24.4.2
     hooks:
       - id: black
 -   repo: https://github.com/PyCQA/flake8
-    rev: 6.0.0
+    rev: 7.1.0
     hooks:
     -   id: flake8
         files: "^(compliance|test|demo)"
 -   repo: https://github.com/PyCQA/bandit
-    rev: 1.7.4
+    rev: 1.7.9
     hooks:
     -   id: bandit
         args: [--recursive]

CHANGES.md

@@ -1,3 +1,7 @@
+# [3.0.0](https://github.com/ComplianceAsCode/auditree-framework/releases/tag/v3.0.0)
+
+- [CHANGED] Remove IBM findings notifier.
+
 # [2.0.1](https://github.com/ComplianceAsCode/auditree-framework/releases/tag/v2.0.1)
 
 - [FIXED] Enable universal newlines when executing local commands.

compliance/__init__.py

@@ -13,4 +13,4 @@
 # limitations under the License.
 """Compliance automation package."""
 
-__version__ = "2.0.1"
+__version__ = "3.0.0"

compliance/config.py

@@ -49,9 +49,6 @@ class ComplianceConfig(object):
             # Pagerduty service id to use for an accreditation
             # E.g. {"mycompany.soc2": "ABCDEFG"}
             "pagerduty": {},
-            # Security Advisor FindingsAPI endpoint to use for an accreditation
-            # E.g. {"mycompany.soc2": "https://my.findings.api/findings"}
-            "findings": {},
         },
         "org": {"name": "YOUR_ORG", "settings": {}},
     }

compliance/fetch.py

@@ -90,7 +90,7 @@ def fetchURL(self, url, params=None, creds=None):  # noqa: N802
         org = self.config.raw_config.get("org", {}).get("name", "")
         ua = f'{org.lower().replace(" ", "-")}-compliance-checks'
         response = requests.get(
-            url, params=params, auth=creds, headers={"User-Agent": ua}
+            url, params=params, auth=creds, headers={"User-Agent": ua}, timeout=3600
         )
         response.raise_for_status()
         return response.content
@@ -151,7 +151,7 @@ def fetchLocalCommands(  # noqa: N802
         if not cwd:
             cwd = os.path.expanduser("~")
         stdin = "\n".join(commands) + "\n"
-        return check_output(
+        return check_output(  # nosec: B603 The input command can be anything.
             cmd, cwd=cwd, env=env, input=stdin, timeout=timeout, universal_newlines=True
         ).rstrip()
 
@@ -165,7 +165,7 @@ def fetch(url, name):
 
     :returns: the path to the file.
     """
-    r = requests.get(url)
+    r = requests.get(url, timeout=3600)
     r.raise_for_status()
     path = Path(tempfile.gettempdir(), name)
     with path.open("wb") as f:

compliance/locker.py

@@ -654,7 +654,7 @@ def get_locker_repo(self, locker="evidence locker"):
     def init_config(self):
         """Apply the git configuration."""
         with self.repo.config_writer() as cw:
-            for (section, cfg) in self.gitconfig.items():
+            for section, cfg in self.gitconfig.items():
                 for key, value in cfg.items():
                     cw.set_value(section, key, value)
 

compliance/notify.py

@@ -27,11 +27,6 @@
 from compliance.utils.services.github import Github
 from compliance.utils.test import parse_test_id
 
-from ibm_cloud_sdk_core.api_exception import ApiException
-from ibm_cloud_sdk_core.authenticators import IAMAuthenticator
-
-from ibm_cloud_security_advisor import FindingsApiV1
-
 import requests
 
 
@@ -861,7 +856,9 @@ def _send_message(self, message, channels):
             retries = self._config.get("retries", 3)
             retry = 0
             while retry < retries:
-                response = requests.post(url, headers=headers, data=json.dumps(msg))
+                response = requests.post(
+                    url, headers=headers, data=json.dumps(msg), timeout=180
+                )
                 if response.status_code == 429:
                     time.sleep(int(response.headers.get("Retry-After", retry)) + 1)
                     retry += 1
@@ -1050,77 +1047,6 @@ def _resolve_alert(self, test_id, test_desc, msg, accreditation):
         )
 
 
-class FindingsNotifier(_BaseNotifier):
-    """
-    Findings notifier class.
-
-    Notifications are sent using the Findings API.  This notifier is
-    configurable via :class:`compliance.config.ComplianceConfig`.
-    """
-
-    def __init__(self, results, controls, push_error=False):
-        """
-        Construct and initialize the Findings notifier object.
-
-        :param results: dictionary generated by
-          :py:class:`compliance.runners.CheckMode` at the end of the execution.
-        :param controls: the control descriptor that manages accreditations.
-        """
-        super(FindingsNotifier, self).__init__(results, controls, push_error)
-        self._config = get_config().get("notify.findings")
-        self._creds = get_config().creds
-        api_key = self._creds["findings"].api_key
-        authenticator = IAMAuthenticator(apikey=api_key)
-        self.findings_api = FindingsApiV1(authenticator=authenticator)
-
-    def notify(self):
-        """Send notifications to the Findings API."""
-        if self._push_error:
-            self.logger.error(
-                "Remote locker push failed.  Findings notifier not triggered."
-            )
-            return
-        self.logger.info("Running the Findings notifier...")
-        if not self._config:
-            self.logger.warning("Using findings notification without config")
-
-        messages = list(self._messages_by_accreditations().items())
-        messages.sort(key=lambda x: x[0])
-        for accreditation, desc in messages:
-            if accreditation not in self._config:
-                continue
-            findings_api_endpoint = self._config[accreditation]
-            self.findings_api.set_service_url(findings_api_endpoint)
-
-            passed, failed, warned, errored = self._split_by_status(desc)
-            for _, _, msg in failed + errored + passed + warned:
-                self._create_findings(msg["body"])
-
-    def _create_findings(self, data):
-        occurrence_list = data["occurrence_list"]
-        account_id = data["account_id"]
-        provider_id = data["provider_id"]
-        status = 0
-
-        for occurrence in occurrence_list:
-            try:
-                response = self.findings_api.create_occurrence(
-                    account_id=account_id, provider_id=provider_id, **occurrence
-                )
-                self.logger.info(response.status_code)
-            except ApiException as e:
-                status = e.code
-                self.logger.error(
-                    "Finding creation failed "
-                    f'for occurrence id {occurrence["id"]} '
-                    f"with {str(e.code)}: {str(e)}"
-                )
-            except Exception as e:
-                status = -1
-                self.logger.error(f"Unexpected error occurred: {str(e)}")
-        return status
-
-
 def get_notifiers():
     """
     Provide a dictionary of all notifier class objects.
@@ -1137,5 +1063,4 @@ def get_notifiers():
         "pagerduty": PagerDutyNotifier,
         "gh_issues": GHIssuesNotifier,
         "locker": LockerNotifier,
-        "findings": FindingsNotifier,
     }

compliance/utils/services/pagerduty.py

@@ -56,7 +56,7 @@ def get(path, params=None, headers=None, creds=None):
     params.update({"limit": PAGES_LIMIT, "offset": offset})
     more = True
     while more:
-        r = requests.get(url, headers=hdrs, params=params)
+        r = requests.get(url, headers=hdrs, params=params, timeout=180)
         yield r
         more = r.json().get("more", False)
         if more:
@@ -77,7 +77,7 @@ def delete(path, params=None, headers=None, creds=None):
     :param creds: a Config object with PagerDuty credentials
     """
     url, params, hdrs = _init_request(path, params, headers, creds)
-    return requests.delete(url, headers=hdrs, params=params)
+    return requests.delete(url, headers=hdrs, params=params, timeout=180)
 
 
 def put(path, params=None, headers=None, creds=None):
@@ -93,7 +93,7 @@ def put(path, params=None, headers=None, creds=None):
     :param creds: a Config object with PagerDuty credentials
     """
     url, params, hdrs = _init_request(path, params, headers, creds)
-    return requests.put(url, headers=hdrs, params=params)
+    return requests.put(url, headers=hdrs, params=params, timeout=180)
 
 
 def post(path, params=None, headers=None, creds=None):
@@ -109,7 +109,7 @@ def post(path, params=None, headers=None, creds=None):
     :param creds: a Config object with PagerDuty credentials
     """
     url, params, hdrs = _init_request(path, params, headers, creds)
-    return requests.post(url, headers=hdrs, params=params)
+    return requests.post(url, headers=hdrs, params=params, timeout=180)
 
 
 def send_event(
@@ -132,7 +132,9 @@ def send_event(
     }
     headers = {"Content-Type": "application/json"}
 
-    response = requests.post(PD_EVENTS_V2_URL, headers=headers, data=json.dumps(msg))
+    response = requests.post(
+        PD_EVENTS_V2_URL, headers=headers, data=json.dumps(msg), timeout=180
+    )
     response.raise_for_status()
     if response.json().get("status") != "success":
         raise RuntimeError("PagerDuty Error: " + response.json())

doc-source/notifiers.rst

@@ -8,7 +8,7 @@ Notifiers
 The last phase in a typical framework check run is the notification
 system.  Multiple notifiers can be targeted as part of this phase by using
 the ``--notify`` option on the ``compliance --check`` command.  Valid
-notifier options are ``stdout``, ``slack``, ``pagerduty``, ``findings``,
+notifier options are ``stdout``, ``slack``, ``pagerduty``,
 ``gh_issues`` and, ``locker``.  The general idea behind the notification
 system is that each ``test_`` can generate a short notification that has the
 following components:
@@ -347,33 +347,3 @@ locker.  The summary markdown file will **only** be pushed to the remote
 evidence locker if the ``full-remote`` argument is applied to the ``evidence``
 option when executing your checks otherwise the file will remain in the local
 evidence locker.  No additional configuration is required for this notifier.
-
-Security Advisor Findings
--------------------------
-
-This configurable notifier will post findings to Security Advisor Findings API
-per accreditation. The following is an example configuration for this notifier
-to be added to a configuration file and used with the ``-C`` option when
-executing your compliance checks::
-
-  {
-    "notify": {
-      "findings": {
-        "accr1": "https://us-south.secadvisor.cloud.ibm.com/findings",
-        "accr2": "https://eu-gb.secadvisor.cloud.ibm.com/findings"
-      }
-    }
-  }
-
-Supported regions for Security Advisor Findings API
-  - us-south: https://us-south.secadvisor.cloud.ibm.com/findings
-  - eu-gb: https://eu-gb.secadvisor.cloud.ibm.com/findings
-
-This notifier also needs to know the credentials for sending findings
-to Security Advisor Findings API. Include the following in your credentials
-file::
-
-  [findings]
-  api_key=platform-api-key
-
-``api_key`` is your IBM Cloud Platform API Key.