Merge pull request #66 from GitGuardian/sbaud/iac-diff-scan

Sbaud/iac diff scan

Author: sylvain-baud-gg

changelog.d/20230703_154225_sylvain.baud.ext_iac_diff_scan.md

@@ -0,0 +1,3 @@
+### Added
+
+- `iac_diff_scan` added to GGClient. This allows scanning two directories for IaC vulnerabilities and categorise incidents as new, unchanged or deleted.

pygitguardian/client.py

@@ -12,7 +12,12 @@
 from requests import Response, Session, codes
 
 from .config import DEFAULT_API_VERSION, DEFAULT_BASE_URI, DEFAULT_TIMEOUT
-from .iac_models import IaCScanParameters, IaCScanParametersSchema, IaCScanResult
+from .iac_models import (
+    IaCDiffScanResult,
+    IaCScanParameters,
+    IaCScanParametersSchema,
+    IaCScanResult,
+)
 from .models import (
     Detail,
     Document,
@@ -270,6 +275,7 @@ def post(
         **kwargs: Any,
     ) -> Response:
         # Be aware that self.iac_directory_scan bypass this method and calls self.request directly.
+        # self.iac_diff_scan also bypass this method
         return self.request(
             "post",
             endpoint=endpoint,
@@ -462,14 +468,23 @@ def create_honeytoken(
             result.status_code = resp.status_code
         return result
 
-    # For IaC Scans
     def iac_directory_scan(
         self,
         directory: Path,
         filenames: List[str],
         scan_parameters: IaCScanParameters,
         extra_headers: Optional[Dict[str, str]] = None,
     ) -> Union[Detail, IaCScanResult]:
+        """
+        iac_directory_scan handles the /iac_scan endpoint of the API.
+
+        :param directory: path to the directory to scan
+        :param filenames: filenames of the directory to include in the scan
+        :param scan_parameters: minimum severities wanted and policies to ignore
+            example: {"ignored_policies":["GG_IAC_0003"],"minimum_severity":"HIGH"}
+        :param extra_headers: optional extra headers to add to the request
+        :return: ScanResult response and status code
+        """
         tar = _create_tar(directory, filenames)
         result: Union[Detail, IaCScanResult]
         try:
@@ -498,6 +513,55 @@ def iac_directory_scan(
 
         return result
 
+    def iac_diff_scan(
+        self,
+        reference: bytes,
+        current: bytes,
+        scan_parameters: IaCScanParameters,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, IaCDiffScanResult]:
+        """
+        iac_diff_scan handles the /iac_diff_scan endpoint of the API.
+
+        Scan two directories and compare their vulnerabilities.
+        Vulnerabilities in reference but not in current are considered "new".
+        Vulnerabilities in both reference and current are considered "unchanged".
+        Vulnerabilities in current but not in reference are considered "deleted".
+
+        :param reference: tar file containing the reference directory. Usually an incoming commit
+        :param current: tar file of the current directory. Usually HEAD
+        :param scan_parameters: minimum severities wanted and policies to ignore
+            example: {"ignored_policies":["GG_IAC_0003"],"minimum_severity":"HIGH"}
+        :param extra_headers: optional extra headers to add to the request
+        :return: ScanResult response and status code
+        """
+        result: Union[Detail, IaCDiffScanResult]
+        try:
+            # bypass self.post because data argument is needed in self.request and self.post use it as json
+            resp = self.request(
+                "post",
+                endpoint="iac_diff_scan",
+                extra_headers=extra_headers,
+                files={
+                    "reference": reference,
+                    "current": current,
+                },
+                data={
+                    "scan_parameters": IaCScanParametersSchema().dumps(scan_parameters),
+                },
+            )
+        except requests.exceptions.ReadTimeout:
+            result = Detail("The request timed out.")
+            result.status_code = 504
+        else:
+            if is_ok(resp):
+                result = IaCDiffScanResult.from_dict(resp.json())
+            else:
+                result = load_detail(resp)
+
+            result.status_code = resp.status_code
+        return result
+
     def read_metadata(self) -> Optional[Detail]:
         """
         Fetch server preferences and store them in `self.secret_scan_preferences`.

pygitguardian/iac_models.py

@@ -63,3 +63,26 @@ class IaCScanResult(Base, FromDictMixin):
     Type[BaseSchema], marshmallow_dataclass.class_schema(IaCScanResult, BaseSchema)
 )
 IaCScanResult.SCHEMA = IaCScanResultSchema()
+
+
+@dataclass
+class IaCDiffScanEntities(Base):
+    unchanged: List[IaCFileResult] = field(default_factory=list)
+    new: List[IaCFileResult] = field(default_factory=list)
+    deleted: List[IaCFileResult] = field(default_factory=list)
+
+
+@dataclass
+class IaCDiffScanResult(Base, FromDictMixin):
+    id: str = ""
+    type: str = ""
+    iac_engine_version: str = ""
+    entities_with_incidents: IaCDiffScanEntities = field(
+        default_factory=IaCDiffScanEntities
+    )
+
+
+IaCDiffScanResultSchema = cast(
+    Type[BaseSchema], marshmallow_dataclass.class_schema(IaCDiffScanResult, BaseSchema)
+)
+IaCDiffScanResult.SCHEMA = IaCDiffScanResultSchema()