Merge pull request #857 from Ostorlab/feature/-Add-CI/CD-parameters-t…

…o-scan-creation-API-and-CLI

 Add CI/CD parameters to scan creation API and CLI

src/ostorlab/apis/scan_create.py

@@ -3,10 +3,21 @@
 import enum
 import io
 import json
+import dataclasses
 from typing import Dict, Optional, BinaryIO, List
 
 from . import request
 
+
+@dataclasses.dataclass
+class ScanSource:
+    """Dataclass holding scan source related parameters."""
+
+    source: str
+    repository: Optional[str] = None
+    pr_number: Optional[str] = None
+
+
 SCAN_PROFILES = {
     "fast_scan": "Fast Scan",
     "full_scan": "Full Scan",
@@ -34,13 +45,15 @@ def __init__(
         application: BinaryIO,
         test_credential_ids: Optional[List[int]] = None,
         sboms: list[io.FileIO] = None,
+        scan_source: Optional[ScanSource] = None,
     ):
         self._title = title
         self._asset_type = asset_type
         self._scan_profile = scan_profile
         self._application = application
         self._test_credential_ids = test_credential_ids
         self._sboms = sboms
+        self._scan_source = scan_source
 
     @property
     def query(self) -> Optional[str]:
@@ -51,8 +64,8 @@ def query(self) -> Optional[str]:
         """
 
         return """
-mutation MobileScan($title: String!, $assetType: String!, $application: Upload!, $sboms: [Upload!], $scanProfile: String!, $credentialIds: [Int]) {
-  createMobileScan(title: $title, assetType: $assetType, application: $application, sboms: $sboms, scanProfile: $scanProfile, credentialIds: $credentialIds) {
+mutation MobileScan($title: String!, $assetType: String!, $application: Upload!, $sboms: [Upload!], $scanProfile: String!, $credentialIds: [Int], $scanSource: ScanSourceInputType) {
+  createMobileScan(title: $title, assetType: $assetType, application: $application, sboms: $sboms, scanProfile: $scanProfile, credentialIds: $credentialIds, scanSource: $scanSource) {
     scan {
       id
     }
@@ -83,6 +96,17 @@ def data(self) -> Optional[Dict]:
                         "scanProfile": self._scan_profile,
                         "credentialIds": self._test_credential_ids,
                         "sboms": [None for _ in self._sboms],
+                        "scanSource": {
+                            "source": self._scan_source.source
+                            if self._scan_source is not None
+                            else None,
+                            "repository": self._scan_source.repository
+                            if self._scan_source is not None
+                            else None,
+                            "prNumber": self._scan_source.pr_number
+                            if self._scan_source is not None
+                            else None,
+                        },
                     },
                 }
             ),

src/ostorlab/cli/ci_scan/run/assets/mobile.py

@@ -6,7 +6,7 @@
 import click
 import itertools
 
-from typing import List
+from typing import List, Optional
 
 from ostorlab.cli.ci_scan.run import run
 from ostorlab.apis.runners import authenticated_runner
@@ -63,6 +63,9 @@ def run_mobile_scan(
         break_on_risk_rating = ctx.obj["break_on_risk_rating"]
         max_wait_minutes = ctx.obj["max_wait_minutes"]
         sboms = ctx.obj["sboms"]
+        repository = ctx.obj["repository"]
+        source = ctx.obj["source"]
+        pr_number = ctx.obj["pr_number"]
         runner = authenticated_runner.AuthenticatedAPIRunner(
             api_key=ctx.obj.get("api_key")
         )
@@ -79,6 +82,12 @@ def run_mobile_scan(
             ci_logger.info(
                 f"creating scan `{title}` with profile `{scan_profile}` for `{asset_type}`"
             )
+            scan_source = None
+            if source is not None:
+                scan_source = scan_create_api.ScanSource(
+                    source=source, repository=repository, pr_number=pr_number
+                )
+
             scan_id = _create_scan(
                 title,
                 scan_profile,
@@ -87,6 +96,7 @@ def run_mobile_scan(
                 credential_ids,
                 runner,
                 sboms,
+                scan_source,
             )
 
             ci_logger.output(name="scan_id", value=scan_id)
@@ -108,7 +118,16 @@ def run_mobile_scan(
         raise click.exceptions.Exit(2) from None
 
 
-def _create_scan(title, scan_profile, asset_type, file, credential_ids, runner, sboms):
+def _create_scan(
+    title: str,
+    scan_profile: str,
+    asset_type: scan_create_api.MobileAssetType,
+    file: io.FileIO,
+    credential_ids: List[int],
+    runner: authenticated_runner.AuthenticatedAPIRunner,
+    sboms: List[io.FileIO],
+    scan_source: Optional[scan_create_api.ScanSource] = None,
+) -> int:
     scan_result = runner.execute(
         scan_create_api.CreateMobileScanAPIRequest(
             title=title,
@@ -117,6 +136,7 @@ def _create_scan(title, scan_profile, asset_type, file, credential_ids, runner,
             application=file,
             test_credential_ids=credential_ids,
             sboms=sboms,
+            scan_source=scan_source,
         )
     )
     scan_id = scan_result.get("data").get("createMobileScan").get("scan").get("id")

src/ostorlab/cli/ci_scan/run/run.py

@@ -8,7 +8,7 @@
 import multiprocessing
 import click
 import time
-from typing import List
+from typing import List, Optional
 
 from ostorlab.cli.ci_scan.ci_scan import ci_scan
 from ostorlab.apis import scan_create as scan_create_api
@@ -127,6 +127,24 @@
     required=False,
     default=None,
 )
+@click.option(
+    "--source",
+    help="CI/CD source.",
+    required=False,
+    default=None,
+)
+@click.option(
+    "--repository",
+    help="Repository name.",
+    required=False,
+    default=None,
+)
+@click.option(
+    "--pr-number",
+    help="Pull request number.",
+    required=False,
+    default=None,
+)
 @click.pass_context
 def run(
     ctx: click.core.Context,
@@ -146,6 +164,9 @@ def run(
     filtered_url_regexes: List[str],
     proxy: str,
     qps: int,
+    source: Optional[str] = None,
+    repository: Optional[str] = None,
+    pr_number: Optional[str] = None,
 ) -> None:
     """Start a scan based on a scan profile in the CI.\n"""
 
@@ -195,6 +216,9 @@ def run(
     ctx.obj["filtered_url_regexes"] = filtered_url_regexes
     ctx.obj["proxy"] = proxy
     ctx.obj["qps"] = qps
+    ctx.obj["source"] = source
+    ctx.obj["repository"] = repository
+    ctx.obj["pr_number"] = pr_number
 
 
 def apply_break_scan_risk_rating(

tests/cli/ci_scan/run/run_test.py

@@ -424,3 +424,50 @@ def testRunWebScanCLI_withsboms_callApi(
     )
 
     assert api_caller_mock.call_count == 2
+
+
+def testRunScanCLI_withSourceGithub_callApi(mocker: plugin.MockerFixture) -> None:
+    """Test ostorlab ci_scan with invalid break_on_risk_rating. it should exit with error exit_code = 2."""
+    scan_create_dict = {"data": {"createMobileScan": {"scan": {"id": "1"}}}}
+
+    scan_info_dict = {
+        "data": {
+            "scan": {
+                "progress": "done",
+                "riskRating": "high",
+            }
+        }
+    }
+
+    api_caller_mock = mocker.patch(
+        "ostorlab.apis.runners.authenticated_runner.AuthenticatedAPIRunner.execute",
+        side_effect=[scan_create_dict, scan_info_dict, scan_info_dict],
+    )
+    mocker.patch.object(run.run, "SLEEP_CHECKS", 1)
+
+    runner = CliRunner()
+    runner.invoke(
+        rootcli.rootcli,
+        [
+            "--api-key=12",
+            "ci-scan",
+            "run",
+            "--scan-profile=full_scan",
+            "--break-on-risk-rating=medium",
+            "--max-wait-minutes=10",
+            "--title=scan1",
+            "--log-flavor=github",
+            "--source=github",
+            "--repository=org/repo",
+            "--pr-number=123456",
+            "ios-ipa",
+            TEST_FILE_PATH,
+        ],
+    )
+
+    assert api_caller_mock.call_count == 2
+    assert api_caller_mock.call_args_list[0].args[0]._scan_source.source == "github"
+    assert (
+        api_caller_mock.call_args_list[0].args[0]._scan_source.repository == "org/repo"
+    )
+    assert api_caller_mock.call_args_list[0].args[0]._scan_source.pr_number == "123456"