Add Azure Cloud assessments for azure defender fetcher

arboretum/__init__.py

@@ -13,4 +13,4 @@
 # limitations under the License.
 """Arboretum - Checking your compliance & security posture, continuously."""
 
-__version__ = "0.16.0"
+__version__ = "0.16.1"

arboretum/azure_cloud/README.md

@@ -0,0 +1,141 @@
+# Azure Cloud library
+
+The fetchers and checks contained within this `azure_cloud` category folder are
+common tests that can be configured and executed for the purpose of generating
+compliance reports and notifications using the [auditree-framework][auditree-framework].
+See [auditree-framework documentation][auditree-framework documentation] for more details.
+
+These tests are normally executed by a CI/CD system like
+[Travis CI](https://travis-ci.com/) as part of another project that uses this
+library package as a dependency.
+
+## Usage as a library
+
+See [usage][usage] for specifics on including this library as a dependency and how
+to include the fetchers and checks from this library in your downstream project.
+
+## Fetchers
+
+### Assessments List
+
+* Class: [AssessmentsListFetcher][fetch-assessments]
+* Purpose: Write the list of Azure Cloud assessments to the evidence locker.
+* Behavior: Access [Azure Cloud API][azure-cloud-api-assessments] and save the list of assessments bound with specified account.
+* Configuration elements:
+  * `org.azure_cloud.accounts`
+    * Required
+    * List of accounts (string)
+    * Each account is an arbitrary name describing the Azure Cloud account. It is used to match to the token provided in the
+      credentials file in order for the fetcher to retrieve content from Azure Cloud for that account.
+* Example (required) configuration:
+
+  ```json
+  {
+    "org": {
+      "azure_cloud": {
+        "accounts": ["myaccount1"]
+      }
+    }
+  }
+  ```
+
+* Required credentials:
+  * `azure_cloud` credentials with read permissions are needed for this fetcher to successfully retrieve the evidence.
+    * `myaccount1_clientid`: The Application(client) ID of Azure [Service Principal][SPN] in Azure Active Directory.
+    * `myaccount1_clientsecret`: The Application(client) secret of Azure [Service Principal][SPN] in Azure Active Directory.
+    * `myaccount1_subscriptionid`: Your azure subscriptioin id.
+    * `myaccount1_tenantid`: Your azure tenant id.
+    * Example credential file entry:
+
+      ```ini
+      [azure_cloud]
+      myaccount1_clientid=your-azure-cloud-app-client-id
+      myaccount1_clientsecret=your-azure-cloud-app-client-secret
+      myaccount1_subscriptionid=your-azure-subscription-id
+      myaccount1_tenantid=your-azure-tenant-id
+      ```
+
+    * NOTE: An Azure [Service Principal][SPN] is a security identity used by user-created applications, services, and automation tools to access specific Azure resources. Assign your application client of SPN to the role of "Reader" under the given your subscription and grant MS Graph API permissions.
+
+* Import statement:
+
+   ```python
+   from arboretum.azure_cloud.fetchers.fetch_assessments import AssessmentsListFetcher
+   ```
+
+### Assessments MetaData List
+
+* Class: [AssessmentsMetaDataListFetcher][fetch-assessments-metadata]
+* Purpose: Write the list of Azure Cloud assessments metadata to the evidence locker.
+* Behavior: Access [Azure Cloud API][azure-cloud-api-assessments-metadata] and save the list of assessments metadata bound with specified account.
+* Configuration elements:
+  * `org.azure_cloud.accounts`
+    * Required
+    * List of accounts (string)
+    * Each account is an arbitrary name describing the Azure Cloud account. It is used to match to the token provided in the
+      credentials file in order for the fetcher to retrieve content from Azure Cloud for that account.
+* Example (required) configuration:
+
+  ```json
+  {
+    "org": {
+      "azure_cloud": {
+        "accounts": ["myaccount1"]
+      }
+    }
+  }
+  ```
+
+* Required credentials: Same with assessments list fetcher
+
+* Import statement:
+
+   ```python
+   from arboretum.azure_cloud.fetchers.fetch_assessments_metadata import AssessmentsMetaDataListFetcher
+   ```
+
+### Sub Assessments List
+
+* Class: [SubAssessmentsListFetcher][fetch-sub-assessments]
+* Purpose: Write the list of Azure Cloud sub assessments to the evidence locker.
+* Behavior: Access [Azure Cloud API][azure-cloud-api-sub-assessments] and save the list of sub assessments bound with specified account.
+* Configuration elements:
+  * `org.azure_cloud.accounts`
+    * Required
+    * List of accounts (string)
+    * Each account is an arbitrary name describing the Azure Cloud account. It is used to match to the token provided in the
+      credentials file in order for the fetcher to retrieve content from Azure Cloud for that account.
+* Example (required) configuration:
+
+  ```json
+  {
+    "org": {
+      "azure_cloud": {
+        "accounts": ["myaccount1"]
+      }
+    }
+  }
+  ```
+
+* Required credentials: Same with assessments list fetcher
+
+* Import statement:
+
+   ```python
+   from arboretum.azure_cloud.fetchers.fetch_sub_assessments import SubAssessmentsListFetcher
+   ```
+
+## Checks
+
+Checks coming soon...
+
+[auditree-framework]: https://github.com/ComplianceAsCode/auditree-framework
+[auditree-framework documentation]: https://complianceascode.github.io/auditree-framework/
+[usage]: https://github.com/ComplianceAsCode/auditree-arboretum#usage
+[azure-cloud-api-assessments]: https://docs.microsoft.com/en-us/rest/api/securitycenter/assessments
+[azure-cloud-api-sub-assessments]: https://docs.microsoft.com/en-us/rest/api/securitycenter/sub-assessments
+[azure-cloud-api-assessments-metadata]: https://docs.microsoft.com/en-us/rest/api/securitycenter/assessments-metadata
+[fetch-assessments]: https://github.com/ComplianceAsCode/auditree-arboretum/blob/main/arboretum/azure_cloud/fetchers/fetch_assessments.py
+[fetch-sub-assessments]: https://github.com/ComplianceAsCode/auditree-arboretum/blob/main/arboretum/azure_cloud/fetchers/fetch_sub_assessments.py
+[fetch-assessments-metadata]: https://github.com/ComplianceAsCode/auditree-arboretum/blob/main/arboretum/azure_cloud/fetchers/fetch_assessments_metadata.py
+[SPN]: https://docs.microsoft.com/en-us/azure/active-directory/develop/howto-create-service-principal-portal

arboretum/azure_cloud/__init__.py

@@ -0,0 +1 @@
+"""Azure Cloud fetchers, checks, and harvest reports."""

arboretum/azure_cloud/checks/__init__.py

@@ -0,0 +1 @@
+"""Azure Cloud validation checks."""

arboretum/azure_cloud/evidences/__init__.py

@@ -0,0 +1 @@
+"""Azure Cloud evidence helper modules/classes."""

arboretum/azure_cloud/fetchers/__init__.py

@@ -0,0 +1 @@
+"""Azure Cloud evidence gathering fetchers."""

arboretum/azure_cloud/fetchers/azure_constants.py

@@ -0,0 +1,4 @@
+"""Common Azure constants."""
+
+# AZURE management resource base url
+AZURE_MANAGEMENT_BASE_URL = "https://management.azure.com/"

arboretum/azure_cloud/fetchers/common.py

@@ -0,0 +1,75 @@
+"""Common utils."""
+
+import requests
+
+from collections import defaultdict
+from compliance.config import get_config
+
+
+def get_token(
+    client_id,
+    client_secret,
+    tenant_id,
+    grant_type="client_credentials",
+    scope="https://management.azure.com/.default",
+):
+    """Get Azure Cloud access token. returns: the access token."""
+    url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
+    headers = {
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Accept": "application/json",
+    }
+    data = (
+        f"grant_type={grant_type}&client_id={client_id}"
+        + f"&client_secret={client_secret}&scope={scope}"
+    )
+    resp = requests.post(url, headers=headers, data=data)
+    resp.raise_for_status()
+    token = resp.json()
+    return token["access_token"]
+
+
+def get_credentials(config, account):
+    """Get credential for the account."""
+    client_id = config.creds.get("azure_cloud", key="clientid", account=account)
+    client_secret = config.creds.get("azure_cloud", key="clientsecret", account=account)
+    tenant_id = config.creds.get("azure_cloud", key="tenantid", account=account)
+    subscription_id = config.creds.get(
+        "azure_cloud", key="subscriptionid", account=account
+    )
+
+    return client_id, client_secret, tenant_id, subscription_id
+
+
+def get_assessments_by_account(account, session, config, api):
+    """
+    Call the get_assessments generator for each configured account.
+
+    Build up a dict of results.
+    """
+    assessments = defaultdict(list)
+    for assessments_resp in paginate_api(account, session, config, api):
+        assessments[account].extend(assessments_resp.get("value", []))
+    return assessments
+
+
+def paginate_api(account, session, config, api_url):
+    """
+    Call the API until there's no longer a nextLink.
+
+    Yeilding data as we go.
+    """
+    client_id, client_secret, tenant_id, _ = get_credentials(config, account)
+    token = get_token(client_id, client_secret, tenant_id)
+    session.headers.update({"Authorization": f"Bearer {token}"})
+    while api_url:
+        # get credential for the account
+        response = session.get(api_url)
+        response.raise_for_status()
+        data = response.json()
+        api_url = data.get("nextLink")
+        yield data
+
+
+def get_azure_accounts():
+    return get_config().get("org.azure_cloud.accounts")

arboretum/azure_cloud/fetchers/fetch_assessments.py

@@ -0,0 +1,59 @@
+"""Azure Cloud assessment fetcher API.
+
+docs https://docs.microsoft.com/en-us/rest/api/
+securitycenter/assessments/list.
+"""
+
+import json
+
+from arboretum.azure_cloud.fetchers.common import (
+    get_assessments_by_account,
+    get_azure_accounts,
+    get_credentials,
+)
+from arboretum.azure_cloud.fetchers.azure_constants import AZURE_MANAGEMENT_BASE_URL
+
+from compliance.evidence import HOUR, RawEvidence, raw_evidence
+from compliance.fetch import ComplianceFetcher
+
+from parameterized import parameterized
+
+
+class AssessmentsListFetcher(ComplianceFetcher):
+    """Fetch the list of assessments."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Initialize the fetcher object with configuration settings."""
+        headers = {"Accept": "application/json"}
+        cls.session(AZURE_MANAGEMENT_BASE_URL, **headers)
+        return cls
+
+    @parameterized.expand(get_azure_accounts)
+    def fetch_full_assessment_list(self, account):
+        """Fetch Azure Cloud full assessments."""
+        _, _, _, subscription_id = get_credentials(self.config, account)
+
+        api = (
+            f"subscriptions/{subscription_id}"
+            + "/providers/Microsoft.Security"
+            + "/assessments?api-version=2020-01-01"
+        )
+
+        assessments_by_account = get_assessments_by_account(
+            account, self.session(), self.config, api
+        )
+        evidence_path = f"azure_cloud/assessment_{account}_list.json"
+        self.config.add_evidences(
+            [
+                RawEvidence(
+                    evidence_path,
+                    "azure_cloud",
+                    8 * HOUR,
+                    f"Azure Cloud full assessments list for account {account}",
+                )
+            ]
+        )
+        with raw_evidence(self.locker, evidence_path) as evidence:
+            if evidence:
+                evidence.set_content(json.dumps(assessments_by_account))

arboretum/azure_cloud/fetchers/fetch_assessments_metadata.py

@@ -0,0 +1,53 @@
+"""Azure Cloud assessment metadata fetcher API."""
+
+import json
+
+from arboretum.azure_cloud.fetchers.common import (
+    get_assessments_by_account,
+    get_azure_accounts,
+    get_credentials,
+)
+from arboretum.azure_cloud.fetchers.azure_constants import AZURE_MANAGEMENT_BASE_URL
+
+from compliance.evidence import HOUR, RawEvidence, raw_evidence
+from compliance.fetch import ComplianceFetcher
+
+from parameterized import parameterized
+
+
+class AssessmentsMetaDataListFetcher(ComplianceFetcher):
+    """Fetch assessments metadata in the specified subscription."""
+
+    @classmethod
+    def setUpClass(cls):
+        """Initialize the fetcher object with configuration settings."""
+        headers = {"Accept": "application/json"}
+        cls.session(AZURE_MANAGEMENT_BASE_URL, **headers)
+        return cls
+
+    @parameterized.expand(get_azure_accounts)
+    def fetch_full_assessment_metadata_list(self, account):
+        """Fetch Azure Cloud full assessment metadata."""
+        _, _, _, subscription_id = get_credentials(self.config, account)
+        api = (
+            f"subscriptions/{subscription_id}"
+            + "/providers/Microsoft.Security/"
+            + "assessmentMetadata?api-version=2020-01-01"
+        )
+        assessments_by_account = get_assessments_by_account(
+            account, self.session(), self.config, api
+        )
+        evidence_path = f"azure_cloud/assessment_metadata_{account}_list.json"
+        self.config.add_evidences(
+            [
+                RawEvidence(
+                    evidence_path,
+                    "azure_cloud",
+                    8 * HOUR,
+                    f"Azure Cloud assessments metadata list for account {account}",
+                )
+            ]
+        )
+        with raw_evidence(self.locker, evidence_path) as evidence:
+            if evidence:
+                evidence.set_content(json.dumps(assessments_by_account))