Merge pull request #147 from 0x41424142/totalcloud

Added totalcloud.get_remediation_activities and sql.upload_totalcloud_remediation_activities

docs/sql.md

@@ -124,6 +124,7 @@ The final optional parameter is ```table_name```. If you want to specify a custo
 | ```upload_totalcloud_aws_iamrole``` | TotalCloud | ```totalcloud.get_inventory(provider='aws', resourceType='iam role')``` | ```totalcloud_aws_iamrole_inventory``` |
 | ```upload_totalcloud_aws_sagemakernotebook``` | TotalCloud | ```totalcloud.get_inventory(provider='aws', resourceType='sagemaker notebook')``` | ```totalcloud_aws_sagemakernotebook_inventory``` |
 | ```upload_totalcloud_aws_cloudfrontdistribution``` | TotalCloud | ```totalcloud.get_inventory(provider='aws', resourceType='cloudfront distribution')``` | ```totalcloud_aws_cloudfrontdistribution_inventory``` |
+| ```upload_totalcloud_remediation_activities``` | TotalCloud | ```totalcloud.get_remediation_activities()``` | ```totalcloud_remediation_activities``` |
 | ```upload_cs_containers``` | Container Security | ```cs.list_containers()``` | ```cs_containers``` |
 | ```upload_was_webapps``` | WAS | ```was.get_webapps()``` or ```was.get_webapps_verbose()``` (```get_webapps_verbose()``` is recommended!) | ```was_webapps``` |
 

docs/totalcloud.md

@@ -21,6 +21,7 @@ You can use any of the endpoints currently supported:
 | ```get_evaluation``` | Get statistics for a specific control on a specific resource ID. |
 | ```get_account_evaluation``` | Get statistics for a list of controls for a specific cloud account. |
 | ```get_resources_evaluated_by_control``` | Get resources evaluated by a specific control. |
+| ```get_remediation_activities``` | Get a list of remediation activities for a specific cloud provider. |
 
 
 
@@ -374,6 +375,36 @@ resources_evaluated[0]
 )
 ```
 
+## Get Remediation Activities API
+
+```get_remediation_activities``` returns a list of remediation activities for a specific cloud provider.
+
+|Parameter| Possible Values |Description| Required|
+|--|--|--|--|
+|```auth```|```qualysdk.auth.BasicAuth``` | Authentication object | ✅ |
+| ```provider``` | ```Literal['aws', 'azure']``` | Cloud Provider | ✅ |
+| ```page_count``` | ```Union[int>=1, 'all'] = 'all'``` | Number of pages to pull | ❌ |
+| ```filter``` | ```str``` | Filter results by Qualys Totalcloud [QQL](https://docs.qualys.com/en/cloudview/latest/search_tips/search_remediation_activity.htm?rhhlterm=search%20remediation%20activity%20activities&rhsearch=Search%20for%20Remediation%20Activity) | ❌ |
+| ```pageNo``` | ```int``` | Page number to start pulling from, or page to pull if ```page_count``` is set to 1 | ❌ |
+| ```pageSize``` | ```int``` | Number of records to pull per page | ❌ |
+
+```py
+from qualysdk.auth import BasicAuth
+from qualysdk.totalcloud import get_remediation_activities
+
+auth = BasicAuth(<username>, <password>, platform='qg1')
+# Get all remediation activities for AWS:
+get_remediation_activities(auth, provider='aws')
+>>>[RemediationActivity(
+    resourceId='123456789123', 
+    region=None, 
+    accountId='246802456', 
+    remediationAction='Disable Access Analyzer',
+    connectorName='myConnector',
+    ...), ...
+]
+```
+
 
 ## ```resourceType``` Values
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "qualysdk"
-version = "0.1.46"
+version = "0.1.47"
 description = "SDK for interacting with Qualys APIs, across most modules the platform offers."
 authors = ["0x41424142 <jake@jakelindsay.uk>", "0x4A616B65 <jake.lindsay@thermofisher.com>"]
 maintainers = ["Jake Lindsay <jake@jakelindsay.uk>"]

qualysdk/base/call_schema.py

@@ -1012,6 +1012,16 @@
                 "pagination": False,
                 "auth_type": "basic",
             },
+            "get_remediation_activities": {
+                "endpoint": "/cloudview-api/rest/v1/remediation/activity",
+                "method": ["GET"],
+                "valid_params": ["pageNo", "pageSize", "filter", "cloudType"],
+                "valid_POST_data": [],
+                "use_requests_json_data": False,
+                "return_type": "json",
+                "pagination": True,
+                "auth_type": "basic",
+            },
         },
         "containersecurity": {
             "url_type": "gateway",

qualysdk/sql/__init__.py

@@ -52,6 +52,7 @@
     upload_totalcloud_aws_iamrole,
     upload_totalcloud_aws_sagemakernotebook,
     upload_totalcloud_aws_cloudfrontdistribution,
+    upload_totalcloud_remediation_activities,
 )
 
 from .cloud_agent import upload_cloud_agents

qualysdk/sql/totalcloud.py

@@ -2282,3 +2282,78 @@ def upload_totalcloud_aws_cloudfrontdistribution(
         dtype=COLS,
         override_import_dt=override_import_dt,
     )
+
+
+def upload_totalcloud_remediation_activities(
+    remediation_activities: BaseList,
+    cnxn: Connection,
+    table_name: str = "totalcloud_remediation_activities",
+    override_import_dt: datetime = None,
+) -> int:
+    """
+    Upload data from totalcloud.get_remediation_activities() to SQL.
+
+    Args:
+        remediation_activities (BaseList): The BaseList of Remediation Activities to upload.
+        cnxn (Connection): The Connection object to the SQL database.
+        table_name (str): The name of the table to upload to. Defaults to 'totalcloud_remediation_activities'.
+        override_import_dt (datetime): Use the passed datetime instead of generating one to upload to the database.
+
+    Returns:
+        int: The number of rows uploaded.
+    """
+
+    COLS = {
+        "resourceId": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "controlId": types.Integer(),
+        "cloudType": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "accountId": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "region": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "status": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "resourceType": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "remediationAction": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "connectorName": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "policyNames": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "controlName": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "triggeredOn": types.DateTime(),
+        "Errors": types.String().with_variant(TEXT(charset="utf8"), "mysql", "mariadb"),
+        "triggeredBy": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+        "remediationReason": types.String().with_variant(
+            TEXT(charset="utf8"), "mysql", "mariadb"
+        ),
+    }
+
+    # Convert the BaseList to a DataFrame:
+    df = DataFrame(
+        [
+            prepare_dataclass(remediation_activity)
+            for remediation_activity in remediation_activities
+        ]
+    )
+
+    # Upload the data:
+    return upload_data(
+        df,
+        table_name,
+        cnxn,
+        dtype=COLS,
+        override_import_dt=override_import_dt,
+    )

qualysdk/totalcloud/__init__.py

@@ -18,3 +18,4 @@
 )
 from .get_inventory import get_inventory
 from .get_resource_details import get_resource_details
+from .remediation_log import get_remediation_activities

qualysdk/totalcloud/data_classes/RemediationActivity.py

@@ -0,0 +1,106 @@
+"""
+Contains the RemediationActivity dataclass.
+"""
+
+from dataclasses import dataclass, asdict
+from typing import Union, Literal
+from datetime import datetime
+
+from ...base.base_list import BaseList
+from ...exceptions.Exceptions import *
+
+
+@dataclass
+class RemediationActivity:
+    """
+    Represents a remediation activity in TotalCloud.
+    """
+
+    resourceId: str = None
+    controlId: int = None
+    cloudType: Literal["AWS", "AZURE"] = None
+    # accountId depends on the cloudType
+    accountId: Union[str, int] = None
+    region: str = None
+    status: str = None
+    resourceType: str = None
+    remediationAction: str = None
+    connectorName: str = None
+    policyNames: BaseList[str] = None
+    controlName: str = None
+    triggeredOn: Union[str, datetime] = None
+    Errors: str = None
+    triggeredBy: str = None
+    remediationReason: str = None
+
+    def __post_init__(self):
+        DT_FIELDS = ["triggeredOn"]
+
+        for field in DT_FIELDS:
+            if not isinstance(getattr(self, field), datetime):
+                setattr(self, field, datetime.fromisoformat(getattr(self, field)))
+
+        if self.cloudType:
+            if self.cloudType.upper() == "AWS":
+                self.accountId = int(self.accountId)
+
+        if self.Errors:
+            setattr(self, "errors", str(self.Errors))
+
+        if self.policyNames:
+            bl = BaseList()
+            for policy in self.policyNames:
+                bl.append(policy)
+            setattr(self, "policyNames", bl)
+
+    def to_dict(self) -> dict:
+        """
+        Converts the object to a dictionary.
+
+        Returns:
+            dict: The object as a dictionary.
+        """
+        return asdict(self)
+
+    def __dict__(self) -> dict:
+        return self.to_dict()
+
+    def keys(self) -> list:
+        """
+        Returns the keys of the object.
+
+        Returns:
+            list: The keys of the object.
+        """
+        return self.to_dict().keys()
+
+    def values(self) -> list:
+        """
+        Returns the values of the object.
+
+        Returns:
+            list: The values of the object.
+        """
+        return self.to_dict().values()
+
+    def items(self) -> list:
+        """
+        Returns the items of the object.
+
+        Returns:
+            list: The items of the object.
+        """
+        return self.to_dict().items()
+
+    @classmethod
+    def from_dict(cls, data: dict):
+        """
+        Creates a RemediationActivity object from a dictionary.
+
+        Args:
+            data (dict): The dictionary to create the object from.
+
+        Returns:
+            RemediationActivity: The object created from the dictionary.
+        """
+        return cls(**data)

qualysdk/totalcloud/data_classes/__init__.py

@@ -8,3 +8,4 @@
 from .Controls import Control, AccountLevelControl
 from .Evaluation import Evaluation, AccountLevelEvaluation
 from .AWSResources import AWSBucket, AWSNetworkACL, AWSRDS
+from .RemediationActivity import RemediationActivity

qualysdk/totalcloud/remediation_log.py

@@ -0,0 +1,91 @@
+"""
+Contains the get_remediation_activities function for Totalcloud
+"""
+
+from typing import Union, Literal
+
+from ..base.call_api import call_api
+from ..base.base_list import BaseList
+from ..auth.token import BasicAuth
+from ..exceptions.Exceptions import *
+from .data_classes.RemediationActivity import RemediationActivity
+
+
+def get_remediation_activities(
+    auth: BasicAuth,
+    provider: Literal["aws", "azure"],
+    page_count: Union[int, "all"] = "all",
+    **kwargs,
+) -> BaseList[dict]:
+    """
+    Pull remediation activity by cloud provider
+
+    Args:
+        auth (BasicAuth): The authentication object.
+        provider (Literal['aws', 'azure']): The cloud provider of the resource.
+        page_count (Union[int, 'all']): The number of pages to retrieve. If 'all', all pages are retrieved.
+
+    ## Kwargs:
+        - filter (str): Filter resources by providing a QQL [query](https://docs.qualys.com/en/cloudview/latest/search_tips/search_remediation_activity.htm?rhhlterm=search%20remediation%20activity%20activities&rhsearch=Search%20for%20Remediation%20Activity)
+        - pageNo (int): The ordered page to start retrieving resources from, or if page_count is 1, the page to retrieve.
+        - pageSize (int): The number of resources to get per page.
+
+    Returns:
+        BaseList[dict]: A list of dictionaries containing the remediation activities. WARNING: I have not seen the response from this API, so I am not sure what the response will look like.
+    """
+
+    if (page_count != "all" and (not isinstance(page_count, int))) or (
+        isinstance(page_count, int) and page_count < 1
+    ):
+        raise ValueError("page_count must be an integer >=1 or 'all'.")
+
+    if provider.lower() not in ["aws", "azure"]:
+        raise ValueError("Invalid provider. Must be 'aws' or 'azure'.")
+
+    params = {
+        "cloudType": provider.upper(),
+    }
+
+    # Add in kwargs to the params
+    if kwargs:
+        params.update(kwargs)
+
+    pulled_pages = 0
+    bl = BaseList()
+
+    while True:
+        # Call the API
+        response = call_api(
+            auth=auth,
+            module="cloudview",
+            endpoint="get_remediation_activities",
+            params=params,
+        )
+
+        # Check the response
+        if response.status_code != 200:
+            raise QualysAPIError(f"{response.status_code}: {response.text}")
+
+        j = response.json()
+
+        if "errorCode" in j:
+            raise QualysAPIError(f"{j['errorCode']}, {j['message']}")
+
+        if "content" not in j.keys() or not j["pageable"].get("empty"):
+            print("No content in response")
+            break
+
+        data = j["content"]
+        if isinstance(data, dict):
+            data = [data]
+        for item in data:
+            bl.append(RemediationActivity.from_dict(item))
+
+        # Check if we need to pull more pages
+        if page_count != "all":
+            pulled_pages += 1
+            if pulled_pages >= page_count:
+                print("Reached page limit of", page_count)
+                break
+
+    return bl