feat: iac diff scan

Author: sylvain-baud-gg

pygitguardian/client.py

@@ -12,7 +12,13 @@
 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,
+    IaCDiffScanResultSchema,
+    IaCScanParameters,
+    IaCScanParametersSchema,
+    IaCScanResult,
+)
 from .models import (
     Detail,
     Document,
@@ -270,6 +276,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,
@@ -498,6 +505,41 @@ def iac_directory_scan(
 
         return result
 
+    # For IaC diff Scans
+    def iac_diff_scan(
+        self,
+        reference: bytes,
+        current: bytes,
+        scan_parameters: IaCScanParameters,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Union[Detail, IaCDiffScanResult]:
+        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 = IaCDiffScanResultSchema.from_dict(resp.json())  # type: ignore
+            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,25 @@ 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):
+    id: str = ""
+    type: str = ""
+    iac_engine_version: str = ""
+    entities_with_incidents: IaCDiffScanEntities = field(
+        default_factory=IaCDiffScanEntities
+    )
+
+
+IaCDiffScanResultSchema = marshmallow_dataclass.class_schema(
+    IaCDiffScanResult, BaseSchema
+)