Merge pull request #217 from 0x41424142/patching

add pm.get_patch_catalog, CLI get_patch_catalog, sql.upload_pm_patch_catalog

docs/patch.md

@@ -25,9 +25,10 @@ You can use any of the endpoints currently supported:
 | ```lookup_cves``` | Returns a list of CVEs and other details associated with a QID. |
 | ```get_patches``` | Returns a list of patches. |
 | ```get_assets``` | Returns a list of assets. |
-| ```get_patch_count``` | Returns the number of patches for a given platform that match ```query``` and ```havingQuery``` |
-| ```get_asset_count``` | Returns the number of assets for a given platform that match ```query``` and ```havingQuery``` |
+| ```get_patch_count``` | Returns the number of patches for a given platform that match ```query``` and ```havingQuery```. |
+| ```get_asset_count``` | Returns the number of assets for a given platform that match ```query``` and ```havingQuery```. |
 | ```lookup_host_uuids``` | Returns a list of tuples, containing host UUIDs for a given list of asset IDs. |
+| ```get_patch_catalog``` | Returns the patch catalog for a given platform according to ```patchId```. |
 
 ## Get PM Version API
 
@@ -635,9 +636,56 @@ uuids = lookup_host_uuids(auth, ids)
 ]
 ```
 
-### Example 1 with GAV Query
+## Get Patch Catalog API
+
+```get_patch_catalog``` returns the patch catalog for a given platform according to ```patchId```.
+
+```patchId``` can be a list/BaseList of strings/integers, a single int/string, or a comma-separated string.
+
+>**Head's Up!:** You should only pass in one platform at a time. If you pass in a mix of both Windows and Linux patches, the function will only return the patches that match the ```platform``` parameter (default is Windows).
+
+|Parameter| Possible Values |Description| Required|
+|--|--|--|--|
+|```auth```|```qualysdk.auth.TokenAuth``` | Authentication object | ✅ |
+| ```patchId``` | ```Union[list[str, int], BaseList[str, int], str, int]``` | The IDs of the patches to look up | ✅ |
+| ```platform``` | ```Literal["windows", "linux"] = "windows"``` | The platform of the patches to return. Defaults to windows | ❌ |
+| ```attributes``` | ```str``` | The attributes to return in the response as a comma-separated string | ❌ |
 
 ```py
+from qualysdk.auth import TokenAuth
+from qualysdk.pm import get_patch_catalog, get_patches
+
+auth = TokenAuth(<username>, <password>, platform='qg1')
+
+# Get some Linux patches:
+patches = get_patches(
+  auth, 
+  platform='linux', 
+  attributes="id"
+)
+
+# Get the catalog entries:
+catalog = get_patch_catalog(
+  auth, 
+  [patch.id for patch in patches], 
+  platform='linux'
+)
+>>>[
+  CatalogPatch(
+    "patchId": "11111111-2222-3333-4444-555555555555",
+    "title": "My Patch",
+    "vendor": "My Vendor",
+    ...
+  ),
+  CatalogPatch(
+    "patchId": "22222222-3333-4444-5555-666666666666",
+    "title": "My Other Patch",
+    ...
+  ),
+  ...
+]
+```
+
 
 ## ```qualysdk-pm``` CLI tool
 
@@ -648,13 +696,12 @@ Use ```--help``` on each command to see the available options.
 ### Usage
 
 ```bash
-usage: qualysdk-pm [-h] -u USERNAME -p PASSWORD [-P {qg1,qg2,qg3,qg4}]
-                   {list_jobs,get_job_results,get_job_runs,lookup_cves,get_patches,get_patch_count,get_assets} ...
+usage: qualysdk-pm [-h] -u USERNAME -p PASSWORD [-P {qg1,qg2,qg3,qg4}] {list_jobs,get_job_results,get_job_runs,lookup_cves,get_patches,get_patch_count,get_assets,get_patch_catalog} ...
 
 CLI script to quickly perform Patch Management (PM) operations using qualysdk
 
 positional arguments:
-  {list_jobs,get_job_results,get_job_runs,lookup_cves,get_patches,get_patch_count,get_assets}
+  {list_jobs,get_job_results,get_job_runs,lookup_cves,get_patches,get_patch_count,get_assets,get_patch_catalog}
                         Action to perform
     list_jobs           Get a list of PM jobs.
     get_job_results     Get results for a PM job.
@@ -663,6 +710,7 @@ positional arguments:
     get_patches         Get patches for a given platform.
     get_patch_count     Get the number of patches available for a platform according to query and havingQuery.
     get_assets          Get assets for a given platform.
+    get_patch_catalog   Get patch catalog entries for a given platform.
 
 options:
   -h, --help            show this help message and exit

docs/sql.md

@@ -139,6 +139,7 @@ The final optional parameter is ```table_name```. If you want to specify a custo
 | ```upload_pm_patches``` | Patch Management | ```pm.get_patches()``` | ```pm_patches``` |
 | ```upload_pm_assets``` | Patch Management | ```pm.get_assets()``` | ```pm_assets``` |
 | ```upload_pm_assetids_to_uuids``` | Patch Management | ```pm.lookup_host_uuids()``` | ```pm_assetids_to_uuids``` |
+| ```upload_pm_patch_catalog``` | Patch Management | ```pm.get_patch_catalog()``` | ```pm_patch_catalog``` |
 | ```upload_cert_certs``` | Certificate View | ```cert.list_certs()``` | ```cert_certs``` for certificates and ```cert_assets``` for assets (key = certs.id -> assets.certId) |
 
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "qualysdk"
-version = "0.2.3"
+version = "0.2.4"
 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_schemas/pm_schema.py

@@ -170,6 +170,16 @@
                 "pagination": False,
                 "auth_type": "token",
             },
+            "get_patch_catalog": {
+                "endpoint": "/pm/v1/patchcatalog/patches",
+                "method": ["POST"],
+                "valid_params": ["attributes", "platform", "sort"],
+                "valid_POST_data": ["patchUuid"],
+                "use_requests_json_data": True,
+                "return_type": "json",
+                "pagination": False,
+                "auth_type": "token",
+            },
         },
     }
 )

qualysdk/cli/qualysdk_pm.py

@@ -34,6 +34,8 @@ def cli_fn(auth: TokenAuth, args: Namespace, endpoint: str) -> None:
             result = get_patch_count(auth, args.platform, **kwargs)
         case "get_assets":
             result = get_assets(auth, args.platform, **kwargs)
+        case "get_patch_catalog":
+            result = get_patch_catalog(auth, args.patch_id, args.os, **kwargs)
         case _:
             raise ValueError(f"Invalid endpoint: {endpoint}.")
 
@@ -267,6 +269,42 @@ def main():
         metavar=("key", "value"),
     )
 
+    get_patch_catalog_parser = subparsers.add_parser(
+        "get_patch_catalog", help="Get patch catalog entries for a given platform."
+    )
+
+    get_patch_catalog_parser.add_argument(
+        "--os",
+        help="Specify the platform to get patches for. Default is 'Windows'",
+        type=str,
+        default="windows",
+        choices=["windows", "linux"],
+    )
+
+    get_patch_catalog_parser.add_argument(
+        "-o",
+        "--output",
+        help="Output xlsx file to write results to",
+        type=str,
+        default="pm_patch_catalog.xlsx",
+    )
+
+    get_patch_catalog_parser.add_argument(
+        "--patch-id",
+        help="Specify the patch ID to get catalog entries for. Can be used multiple times",
+        type=str,
+        required=True,
+        action="append",
+    )
+
+    get_patch_catalog_parser.add_argument(
+        "--kwarg",
+        help="Specify a keyword argument to pass to the action. Can be used multiple times",
+        action="append",
+        nargs=2,
+        metavar=("key", "value"),
+    )
+
     args = parser.parse_args()
 
     # create TokenAuth object
@@ -306,6 +344,9 @@ def main():
             args.kwarg = {}  # No kwargs for this endpoint
             args.platform = args.os
             cli_fn(auth=auth, args=args, endpoint="get_patch_count")
+        case "get_patch_catalog":
+            args.platform = args.os
+            cli_fn(auth=auth, args=args, endpoint="get_patch_catalog")
         case _:
             parser.print_help()
             exit(1)

qualysdk/pm/__init__.py

@@ -16,3 +16,4 @@
 from .version import get_version
 from .patches import get_patches, get_patch_count
 from .assets import get_assets, lookup_host_uuids
+from .patchcatalog import get_patch_catalog

qualysdk/pm/assets.py

@@ -89,13 +89,6 @@ def lookup_host_uuids(
     if not isinstance(assetIds, (list, BaseList)):
         assetIds = [assetIds]
 
-    # If assetIds is longer than 1K, chunk it into smaller lists:
-    if len(assetIds) > 1_000:
-        print(
-            "[!]: len(assetIds) > 1K. Expect longer processing time as the SDK will make multiple API calls."
-        )
-
-    no_results = []
     mapped_results = BaseList()
 
     while True:
@@ -118,16 +111,7 @@ def lookup_host_uuids(
             for asset in j["mappedAssets"]:
                 mapped_results.append((asset["assetId"], asset["assetUuid"]))
 
-        if j.get("unmappedAssets") and j.get("unmappedAssets") not in ["[]", "[null]"]:
-            for asset in j["unmappedAssets"]:
-                no_results.append(asset)
-
         if len(j["mappedAssets"]) == 0:
             break
 
-    if no_results:
-        print(
-            f"Warning: {len(no_results)} assetId(s) could not be found. Please check returned list."
-        )
-
     return mapped_results

qualysdk/pm/data_classes/CatalogPatch.py

@@ -0,0 +1,111 @@
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Union
+
+from ...base.base_class import BaseClass
+from ...base.base_list import BaseList
+
+DT_FIELDS = [
+    "syncDateTime",
+    "modifiedDate",
+    "publishedDate",
+]
+
+BL_STR_FIELDS = [
+    "appFamily",
+    "product",
+    "qid",
+    "supersedes",
+    "cve",
+    "architecture",
+    "supersededBy",
+    "supportedOs",
+]
+
+
+@dataclass
+class PackageDetail(BaseClass):
+    """
+    Linux package details.
+    """
+
+    packageName: str = None
+    architecture: str = None
+
+    def __str__(self):
+        return f"{self.packageName} ({self.architecture})"
+
+
+@dataclass
+class CatalogPatch(BaseClass):
+    """
+    A data class representing a patch in the Qualys Patch Management catalog.
+    """
+
+    patchId: str = None
+    id: str = None
+    title: str = None
+    type: str = None
+    appFamily: BaseList[str] = None
+    vendor: str = None
+    product: BaseList[str] = None
+    platform: str = None
+    kb: str = None
+    isSuperseded: bool = None
+    isSecurity: bool = None
+    isRollback: bool = None
+    servicePack: bool = None
+    advisory: str = None
+    vendorlink: str = None
+    osIdentifier: str = None
+    advisoryLink: str = None
+    deleted: bool = None
+    rebootRequired: bool = None
+    vendorSeverity: str = None
+    description: str = None
+    qid: BaseList[int] = None
+    enabled: bool = None
+    downloadMethod: str = None
+    supportedOs: str = None
+    supersedes: BaseList[str] = None
+    notification: str = None
+    cve: BaseList[str] = None
+    architecture: BaseList[str] = None
+    packageDetails: BaseList[PackageDetail] = None
+    patchFeedProviderId: int = None
+    syncDateTime: Union[int, datetime] = None
+    vendorPatchId: Union[str, int] = None
+    modifiedDate: Union[int, datetime] = None
+    publishedDate: Union[int, datetime] = None
+    category: str = None
+    isEsuPatch: bool = None
+    supersededBy: BaseList[str] = None
+    bulletin: str = None
+
+    def __post_init__(self):
+        for field in DT_FIELDS:
+            if getattr(self, field):
+                try:
+                    setattr(
+                        self,
+                        field,
+                        datetime.fromtimestamp(getattr(self, field) / 1000),
+                    )
+                except (TypeError, OSError):
+                    setattr(self, field, None)
+
+        for field in BL_STR_FIELDS:
+            if getattr(self, field):
+                setattr(self, field, BaseList(getattr(self, field)))
+
+        if self.notification:
+            print(
+                "CatalogPatch's notification attribute is currently not parsed and is set to a string. Please submit a PR adding the functionality to parse this attribute."
+            )
+            setattr(self, "notification", str(self.notification))
+
+        if self.packageDetails:
+            bl = BaseList()
+            for pd in self.packageDetails:
+                bl.append(PackageDetail(**pd))
+            setattr(self, "packageDetails", bl)

qualysdk/pm/data_classes/__init__.py

@@ -7,3 +7,4 @@
 from .PMAsset import PMAssetJobView, Asset
 from .PMRun import PMRun
 from .Patch import Patch
+from .CatalogPatch import CatalogPatch