feat: output results to json file

Author: kirklandnuts

preflight_check/preflight_check/app.py

@@ -58,6 +58,7 @@ def configure(
                 assert isinstance(monitored_subscriptions_input, str)
                 assert isinstance(region_names, str)
                 assert isinstance(use_nat_gateway, bool)
+                assert isinstance(integration_type, models.IntegrationType)
                 scanning_subscription = self._subscriptions.get_subscription(
                     scanning_subscription_input.strip()
                 )
@@ -88,13 +89,16 @@ def configure(
                 use_nat_gateway=use_nat_gateway,
             )
 
-    def run(self) -> None:
+    def run(self, output_path: str) -> None:
         """Run the preflight check"""
+        if not self.deployment_config:
+            raise RuntimeError("Deployment config not set")
         usage_quota_limits = self._get_usage_quota_limits()
         permissions = self._get_permissions()
         preflight_check = PreflightCheck(
             self.deployment_config, usage_quota_limits, permissions)
         cli.print_preflight_check(preflight_check)
+        cli.output_preflight_check_results_file(preflight_check, path=output_path)
 
     def _prompt_deployment_config(self) -> None:
         available_subscriptions = self._subscriptions.get_subscriptions()
@@ -212,6 +216,10 @@ def main(
             help="Use NAT Gateway for optimized networking (recommended for 1000+ VMs)",
         ),
     ] = None,
+    output_path: Annotated[
+        str,
+        typer.Option("--output-path", "-o", help="Path to output the preflight check results"),
+    ] = "./preflight_report.json",
 ) -> None:
     """
     Preflight check for Azure Agentless Scanner deployment.
@@ -220,7 +228,7 @@ def main(
         app = App()
         app.configure(integration_type, scanning_subscription,
                       monitored_subscriptions, regions, use_nat_gateway)
-        app.run()
+        app.run(output_path)
     except typer.Exit:
         raise
     except Exception as e:

preflight_check/preflight_check/cli.py

@@ -1,3 +1,5 @@
+import json
+from pathlib import Path
 
 from rich.box import HEAVY_EDGE
 from rich.console import Console
@@ -16,42 +18,38 @@ def prompt_scanning_subscription(available_subs: list[Subscription]) -> Subscrip
     print_subscriptions(available_subs)
     console.print("\n[bold]Scanning Subscription[/bold]")
     console.print(
-        "[dim]What subscription do you want to deploy the AWLS scanner resources to?[/dim]")
-    sub_id = Prompt.ask("Subscription ID",
-                        choices=[sub.id for sub in available_subs],
-                        show_choices=False)
+        "[dim]What subscription do you want to deploy the AWLS scanner resources to?[/dim]"
+    )
+    sub_id = Prompt.ask(
+        "Subscription ID", choices=[sub.id for sub in available_subs], show_choices=False
+    )
 
     # Find the subscription in our list of available subscriptions
     for sub in available_subs:
         if sub.id == sub_id:
-            console.print(
-                f"[green]Selected scanning subscription: {sub.name}[/green]")
+            console.print(f"[green]Selected scanning subscription: {sub.name}[/green]")
             return sub
     raise ValueError(f"Subscription {sub_id} not found")
 
 
-def prompt_monitored_subscriptions(available_subs: list[Subscription]) -> tuple[list[Subscription], IntegrationType]:
+def prompt_monitored_subscriptions(
+    available_subs: list[Subscription],
+) -> tuple[list[Subscription], IntegrationType]:
     """Prompt user to choose which subscriptions to monitor"""
     console.print("\n[bold]Monitored Subscriptions[/bold]")
     console.print("[dim]Choose which subscriptions to be monitored.[/dim]")
 
-    scan_scope = Prompt.ask(
-        "Scan scope",
-        choices=["all", "exclude", "specify"],
-        default="all"
-    )
+    scan_scope = Prompt.ask("Scan scope", choices=["all", "exclude", "specify"], default="all")
 
     if scan_scope == "all":
         console.print("[dim]All subscriptions will be monitored.[/dim]")
         return (available_subs, IntegrationType.TENANT)
 
     if scan_scope == "exclude":
-        console.print(
-            "[dim]Select subscriptions to exclude from monitoring.[/dim]")
+        console.print("[dim]Select subscriptions to exclude from monitoring.[/dim]")
         print_subscriptions(available_subs)
 
-        excluded_ids = Prompt.ask(
-            "Excluded Subscription IDs (comma-separated)")
+        excluded_ids = Prompt.ask("Excluded Subscription IDs (comma-separated)")
         excluded_subs = []
 
         for sub_id in [s.strip() for s in excluded_ids.split(",")]:
@@ -60,21 +58,18 @@ def prompt_monitored_subscriptions(available_subs: list[Subscription]) -> tuple[
                     excluded_subs.append(sub)
                     break
             else:
-                console.print(
-                    f"[yellow]Warning: Subscription {sub_id} not found[/yellow]")
+                console.print(f"[yellow]Warning: Subscription {sub_id} not found[/yellow]")
 
         monitored_subs = [s for s in available_subs if s not in excluded_subs]
-        console.print(
-            "\n[bold]The following subscriptions will be excluded:[/bold]")
+        console.print("\n[bold]The following subscriptions will be excluded:[/bold]")
         print_subscriptions(excluded_subs)
         return (monitored_subs, IntegrationType.TENANT)
 
     # specify
     console.print("[dim]Select specific subscriptions to monitor.[/dim]")
     print_subscriptions(available_subs)
 
-    specified_ids = Prompt.ask(
-        "Specified Subscription IDs (comma-separated)")
+    specified_ids = Prompt.ask("Specified Subscription IDs (comma-separated)")
     specified_subs = []
 
     for sub_id in [s.strip() for s in specified_ids.split(",")]:
@@ -83,11 +78,9 @@ def prompt_monitored_subscriptions(available_subs: list[Subscription]) -> tuple[
                 specified_subs.append(sub)
                 break
         else:
-            console.print(
-                f"[yellow]Warning: Subscription {sub_id} not found[/yellow]")
+            console.print(f"[yellow]Warning: Subscription {sub_id} not found[/yellow]")
 
-    console.print(
-        "\n[bold]Only the following subscriptions will be monitored:[/bold]")
+    console.print("\n[bold]Only the following subscriptions will be monitored:[/bold]")
     print_subscriptions(specified_subs)
     return (specified_subs, IntegrationType.SUBSCRIPTION)
 
@@ -96,9 +89,11 @@ def prompt_nat_gateway() -> bool:
     """Prompt user about NAT Gateway usage"""
     console.print("\n[bold]Network Configuration[/bold]")
     console.print(
-        "[dim]We recommend deploying AWLS with a NAT Gateway, but you can opt out if you prefer.[/dim]")
+        "[dim]We recommend deploying AWLS with a NAT Gateway, but you can opt out if you prefer.[/dim]"
+    )
     console.print(
-        "[dim]For more details on deploying with/without a NAT Gateway, please refer to the Lacework FortiCNAPP docs:\nhttps://docs.fortinet.com/document/lacework-forticnapp/24.4.0/new-features/13869/integrating-agentless-workload-security-with-azure?preview_token=44849b8a6bf658c3c2b3#:~:text=deploy%20the%20Agentless%20scanning%20integration%20with%20a%20NAT%20gateway.[/dim]")
+        "[dim]For more details on deploying with/without a NAT Gateway, please refer to the Lacework FortiCNAPP docs:\nhttps://docs.fortinet.com/document/lacework-forticnapp/24.4.0/new-features/13869/integrating-agentless-workload-security-with-azure?preview_token=44849b8a6bf658c3c2b3#:~:text=deploy%20the%20Agentless%20scanning%20integration%20with%20a%20NAT%20gateway.[/dim]"
+    )
     return Confirm.ask("Use NAT Gateway?", default=True)
 
 
@@ -164,8 +159,7 @@ def print_vm_counts(subscriptions: list[Subscription]) -> None:
     console.print(table)
 
     if subscriptions_with_no_vms:
-        console.print(
-            "\n[yellow]Warning: The following subscriptions have no VMs:[/yellow]")
+        console.print("\n[yellow]Warning: The following subscriptions have no VMs:[/yellow]")
         print_subscriptions(subscriptions_with_no_vms)
 
 
@@ -178,26 +172,21 @@ def prompt_regions(subscriptions: list[Subscription]) -> list[str]:
     # Get unique regions across all subscriptions
     detected_regions = {region_name for sub in subscriptions for region_name in sub.regions}
     console.print("\n[bold]Deployment Regions[/bold]")
-    console.print(
-        "[dim]Enter the Azure regions that you'd like to monitor.[/dim]")
+    console.print("[dim]Enter the Azure regions that you'd like to monitor.[/dim]")
     console.print("[dim]The scanner will be deployed in these regions.[/dim]")
 
-    regions_default = ",".join(
-        sorted(detected_regions)) if detected_regions else None
-    regions_input = Prompt.ask(
-        "Azure regions (comma-separated)",
-        default=regions_default
-    )
+    regions_default = ",".join(sorted(detected_regions)) if detected_regions else None
+    regions_input = Prompt.ask("Azure regions (comma-separated)", default=regions_default)
     assert isinstance(regions_input, str)
     selected_regions = []
     for region_input in regions_input.strip().split(","):
         if region_input.strip() in detected_regions:
             selected_regions.append(region_input.strip())
         else:
-            console.print(
-                f"[yellow]Warning: Region {region_input.strip()} not found[/yellow]")
+            console.print(f"[yellow]Warning: Region {region_input.strip()} not found[/yellow]")
     console.print(
-        f"[green]The following regions will be monitored:[/green] {', '.join(selected_regions)}")
+        f"[green]The following regions will be monitored:[/green] {', '.join(selected_regions)}"
+    )
     return selected_regions
 
 
@@ -208,7 +197,8 @@ def print_quota_checks(quota_checks: QuotaChecks) -> None:
 
     console.print("\n[bold]Usage Quota Limits[/bold]")
     console.print(
-        f"Here are the configured and required usage quota limits for subscription [bold]{scanning_subscription.name}[/bold]:\n")
+        f"Here are the configured and required usage quota limits for subscription [bold]{scanning_subscription.name}[/bold]:\n"
+    )
 
     # Create quota requirements table
     table = Table(show_header=True, header_style="bold", box=HEAVY_EDGE)
@@ -246,7 +236,7 @@ def print_quota_checks(quota_checks: QuotaChecks) -> None:
                 str(check.configured_limit),
                 str(check.current_usage),
                 str(check.required_quota),
-                status
+                status,
             )
         # Add a blank row between regions (unless it's the last region)
         if region_name != region_names[-1]:
@@ -256,13 +246,12 @@ def print_quota_checks(quota_checks: QuotaChecks) -> None:
 
     # Print summary and help message
     if quota_checks.all_checks_pass():
-        console.print(
-            "\n[green]✅ All configured usage quota limits are sufficient![/green]")
+        console.print("\n[green]✅ All configured usage quota limits are sufficient![/green]")
     else:
+        console.print("\n[yellow]⚠️ Some configured usage quota limits are not sufficient.[/yellow]")
         console.print(
-            "\n[yellow]⚠️ Some configured usage quota limits are not sufficient.[/yellow]")
-        console.print(
-            "Please request quota increases at: https://portal.azure.com/#blade/Microsoft_Azure_Capacity/QuotaRequestBlade")
+            "Please request quota increases at: https://portal.azure.com/#blade/Microsoft_Azure_Capacity/QuotaRequestBlade"
+        )
 
 
 def print_auth_checks(auth_checks: AuthChecks) -> None:
@@ -271,12 +260,11 @@ def print_auth_checks(auth_checks: AuthChecks) -> None:
     monitored_subscriptions_auth_checks = auth_checks.monitored_subscriptions
 
     # print scanning subscription auth check
-    console.print("\n[bold]Scanning Subscription Authorization Checks[/bold]")
+    console.print("\n[bold]Scanning Subscription Permission Checks[/bold]")
     print_auth_check_table([scanning_subscription_auth_check])
 
     # print monitored subscriptions auth checks
-    console.print(
-        "\n[bold]Monitored Subscriptions Authorization Checks[/bold]")
+    console.print("\n[bold]Monitored Subscriptions Permission Checks[/bold]")
     print_auth_check_table(monitored_subscriptions_auth_checks)
 
 
@@ -304,8 +292,113 @@ def print_auth_check_table(auth_checks: list[AuthCheck]) -> None:
     console.print(table)
 
 
-
 def print_preflight_check(preflight_check: PreflightCheck) -> None:
     """Print the preflight check results"""
     print_quota_checks(preflight_check.usage_quota_checks)
     print_auth_checks(preflight_check.auth_checks)
+    # summarize results
+    console.print("\n[bold]Preflight Check Summary[/bold]")
+    if preflight_check.usage_quota_checks.all_checks_pass():
+        console.print("[green bold]✅ All usage quota limits are sufficient![/green bold]")
+    else:
+        console.print("\n[yellow bold]Some usage quota limits are not sufficient.[/yellow bold]")
+        for region_name, quota_checks in preflight_check.usage_quota_checks.quota_checks.items():
+            console.print(f"\t[yellow]{region_name}[/yellow]")
+            for quota_check in quota_checks:
+                if not quota_check.success:
+                    console.print(
+                        f"\t\t[yellow]- {quota_check.display_name} (Configured: {quota_check.configured_limit}, Current: {quota_check.current_usage}, Required: {quota_check.required_quota})[/yellow]"
+                    )
+    # print auth checks summary
+    if preflight_check.auth_checks.all_checks_pass():
+        console.print("[green bold]✅ All permission checks passed![/green bold]")
+    else:
+        console.print("\n[yellow bold]Some permission checks did not pass.[/yellow bold]")
+        console.print(
+            "[yellow]The authenticated principal is missing the following permissions on the following subscriptions:[/yellow]"
+        )
+        for auth_check in [
+            preflight_check.auth_checks.scanning_subscription,
+            *preflight_check.auth_checks.monitored_subscriptions,
+        ]:
+            if not auth_check.success:
+                console.print(f"\t[bold]{auth_check.subscription.name}[/bold]:")
+                for missing_permission in auth_check.missing_permissions:
+                    console.print(
+                        f"\t\t[yellow]- {missing_permission.required_permission}[/yellow]"
+                    )
+
+
+def output_preflight_check_results_file(
+    preflight_check: PreflightCheck, path_str: str = "./preflight_report.json"
+) -> None:
+    """Output the preflight check results to a file"""
+    path = Path(path_str)
+    results = {
+        "deployment_config": {
+            "integration_level": preflight_check.deployment_config.integration_type.value,
+            "scanning_subscription": f"/subscriptions/{preflight_check.deployment_config.scanning_subscription.id}",
+            "monitored_subscriptions": [
+                f"/subscriptions/{sub.id}"
+                for sub in preflight_check.deployment_config.monitored_subscriptions
+            ],
+            "regions": preflight_check.deployment_config.regions,
+            "use_nat_gateway": preflight_check.deployment_config.use_nat_gateway,
+        },
+        "vm_count": {
+            f"/subscriptions/{sub.id}": {
+                region_name: region.vm_count for region_name, region in sub.regions.items()
+            }
+            for sub in preflight_check.deployment_config.monitored_subscriptions
+        },
+        "success": preflight_check.usage_quota_checks.all_checks_pass()
+        and preflight_check.auth_checks.all_checks_pass(),
+        "permissions_check": {
+            "success": preflight_check.auth_checks.all_checks_pass(),
+            "scanning_subscription": {
+                "scope": f"/subscriptions/{preflight_check.auth_checks.scanning_subscription.subscription.id}",
+                "success": preflight_check.auth_checks.scanning_subscription.success,
+                "missing_permissions": [
+                    check.required_permission
+                    for check in preflight_check.auth_checks.scanning_subscription.missing_permissions
+                ],
+            },
+            "monitored_subscriptions": [
+                {
+                    "scope": f"/subscriptions/{auth_check.subscription.id}",
+                    "success": auth_check.success,
+                    "missing_permissions": [
+                        check.required_permission for check in auth_check.missing_permissions
+                    ],
+                }
+                for auth_check in preflight_check.auth_checks.monitored_subscriptions
+            ],
+        },
+        "usage_quota_check": {
+            "success": preflight_check.usage_quota_checks.all_checks_pass(),
+            "subscription": f"/subscriptions/{preflight_check.usage_quota_checks.subscription.id}",
+            "quota_checks": [
+                {
+                    "region": region_name,
+                    "quotas": [
+                        {
+                            "name": check.name,
+                            "display_name": check.display_name,
+                            "required_limit": check.required_quota,
+                            "current_usage": check.current_usage,
+                            "configured_limit": check.configured_limit,
+                            "success": check.success,
+                        }
+                        for check in checks
+                    ],
+                    "success": all(check.success for check in checks),
+                }
+                for region_name, checks in preflight_check.usage_quota_checks.quota_checks.items()
+            ],
+        },
+    }
+
+    with open(path, "w") as f:
+        json.dump(results, f, indent=2)
+
+    console.print(f"\nPreflight check results written to [bold]{path}[/bold]\n")

preflight_check/preflight_check/core/auth_check.py

@@ -51,6 +51,10 @@ def required_permissions(self) -> list[str]:
     def success(self) -> bool:
         return all(check.is_granted for check in self.checked_permissions)
 
+    @property
+    def missing_permissions(self) -> list[RequiredPermissionCheck]:
+        return [check for check in self.checked_permissions if not check.is_granted]
+
 
 class MonitoredSubscriptionAuthCheck(AuthCheck):
     """Defines the permissions required on a monitored subscription"""
@@ -64,6 +68,12 @@ def required_permissions(self) -> list[str]:
             "Microsoft.Authorization/roleDefinitions/write",
             "Microsoft.Authorization/roleDefinitions/delete",
             "Microsoft.Authorization/roleDefinitions/read",
+            "Microsoft.Storage/storageAccounts/blobServices/read",
+            "Microsoft.Storage/storageAccounts/blobServices/write",
+            "Microsoft.Storage/storageAccounts/blobServices/delete",
+            "Microsoft.Storage/storageAccounts/blobServices/containers/read",
+            "Microsoft.Storage/storageAccounts/blobServices/containers/write",
+            "Microsoft.Storage/storageAccounts/blobServices/containers/delete",
         ]