Merge pull request #142 from awslabs/v3/feat/add-aws-plugin-module

feat(metrics collection): added metrics collection and reworked AshAggregatedResults scanner results content for usability

Author: scrthq

.ash/.ash.yaml

@@ -1,6 +1,8 @@
 # yaml-language-server: $schema=../automated_security_helper/schemas/AshConfig.json
 project_name: automated-security-helper
 fail_on_findings: true
+ash_plugin_modules:
+  - automated_security_helper.plugin_modules.ash_aws_plugins
 global_settings:
   severity_threshold: MEDIUM
   ignore_paths:
@@ -22,11 +24,11 @@ build:
 reporters:
   asff:
     enabled: false
-  # cloudwatch-logs:
-  #   enabled: true
-  #   options:
-  #     aws_region: us-west-2
-  #     log_group_name: !ENV ASH_LOG_GROUP_NAME
+  cloudwatch-logs:
+    enabled: true
+    options:
+      aws_region: !ENV ${AWS_REGION:None}
+      log_group_name: !ENV ${ASH_CLOUDWATCH_LOG_GROUP:None}
   csv:
     enabled: true
   cyclonedx:
@@ -70,8 +72,7 @@ scanners:
     options:
       frameworks:
         - all
-      additional_formats:
-        - cyclonedx_json
+      additional_formats: []
       skip_path: []
       config_file: .ash/.checkov.yaml
   detect-secrets:

.vscode/launch.json

@@ -9,31 +9,27 @@
     },
     {
       "args": [
-        "run",
-        "--source-dir",
-        "${workspaceFolder}",
-        "--output-dir",
-        "${workspaceFolder}/ash_output",
-        "--config",
-        "${workspaceFolder}/.ash.yaml",
+        "scan",
         "--verbose",
         "--no-progress",
-        "--no-cleanup"
+        "--",
+        "--scanners=\"npm-audit\"",
+        "--exclude-scanners=\"cdk-nag\""
       ],
       "console": "integratedTerminal",
       "env": {
         "PATH": "${workspaceFolder}/.venv/bin:${env.PATH}"
       },
       "name": "ASH: Test CLI",
-      "program": "${workspaceFolder}/src/automated_security_helper/cli/main.py",
+      "program": "${workspaceFolder}/.venv/bin/ash",
       "request": "launch",
       "type": "debugpy"
     },
     {
       "args": [],
       "console": "integratedTerminal",
       "name": "ASH: Test CDK Nag Headless Wrapper",
-      "program": "./src/automated_security_helper/utils/cdk_nag_wrapper.py",
+      "program": "./automated_security_helper/utils/cdk_nag_wrapper.py",
       "request": "launch",
       "type": "debugpy"
     },

Dockerfile

@@ -1,4 +1,4 @@
-#checkov:skip=CKV_DOCKER_7: Base image is using a non-latest version tag by default, Checkov is unable to parse due to the use of ARG
+#checkov:skip=CKV_DOCKER_7:Base image is using a non-latest version tag by default, Checkov is unable to parse due to the use of ARG
 ARG BASE_IMAGE=public.ecr.aws/docker/library/python:3.10-bullseye
 
 # First stage: Build poetry requirements

automated_security_helper/base/converter_plugin.py

@@ -2,19 +2,22 @@
 
 from abc import abstractmethod
 from pathlib import Path
-from typing import Generic, List, TypeVar
+from typing import Annotated, Generic, List, TypeVar
 from typing_extensions import Self
 
-from pydantic import model_validator
+from pydantic import Field, model_validator
 
+from automated_security_helper.base.options import ConverterOptionsBase
 from automated_security_helper.base.plugin_base import PluginBase
 from automated_security_helper.base.plugin_config import PluginConfigBase
 from automated_security_helper.core.exceptions import ScannerError
 from automated_security_helper.utils.log import ASH_LOGGER
 
 
 class ConverterPluginConfigBase(PluginConfigBase):
-    pass
+    options: Annotated[ConverterOptionsBase, Field(description="Converter options")] = (
+        ConverterOptionsBase()
+    )
 
 
 T = TypeVar("T", bound=ConverterPluginConfigBase)

automated_security_helper/base/options.py

@@ -10,16 +10,18 @@ class BuilderOptionsBase(BaseModel):
     model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
 
 
-class ConverterOptionsBase(BaseModel):
-    """Base class for converter options."""
+class PluginOptionsBase(BaseModel):
+    """Base class for plugin options."""
 
     model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
 
 
-class ScannerOptionsBase(BaseModel):
-    """Base class for scanner options."""
+class ConverterOptionsBase(PluginOptionsBase):
+    """Base class for converter options."""
 
-    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)
+
+class ScannerOptionsBase(PluginOptionsBase):
+    """Base class for scanner options."""
 
     severity_threshold: Annotated[
         Literal["ALL", "LOW", "MEDIUM", "HIGH", "CRITICAL"] | None,
@@ -29,7 +31,5 @@ class ScannerOptionsBase(BaseModel):
     ] = None
 
 
-class ReporterOptionsBase(BaseModel):
+class ReporterOptionsBase(PluginOptionsBase):
     """Base class for reporter options."""
-
-    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

automated_security_helper/base/plugin_config.py

@@ -1,4 +1,4 @@
-from automated_security_helper.base.options import ScannerOptionsBase
+from automated_security_helper.base.options import PluginOptionsBase
 from pydantic import BaseModel, ConfigDict, Field
 from typing import Annotated
 
@@ -23,6 +23,6 @@ class PluginConfigBase(BaseModel):
     enabled: Annotated[bool, Field(description="Whether the component is enabled")] = (
         True
     )
-    options: Annotated[ScannerOptionsBase, Field(description="Scanner options")] = (
-        ScannerOptionsBase()
+    options: Annotated[PluginOptionsBase, Field(description="Scanner options")] = (
+        PluginOptionsBase()
     )

automated_security_helper/base/reporter_plugin.py

@@ -2,11 +2,12 @@
 
 from abc import abstractmethod
 from datetime import datetime, timezone
-from typing import Generic, TypeVar
+from typing import Annotated, Generic, TypeVar
 from typing_extensions import Self
 
-from pydantic import model_validator
+from pydantic import Field, model_validator
 
+from automated_security_helper.base.options import ReporterOptionsBase
 from automated_security_helper.base.plugin_base import PluginBase
 from automated_security_helper.base.plugin_config import PluginConfigBase
 from automated_security_helper.core.exceptions import ScannerError
@@ -15,6 +16,9 @@
 
 
 class ReporterPluginConfigBase(PluginConfigBase):
+    options: Annotated[ReporterOptionsBase, Field(description="Reporter options")] = (
+        ReporterOptionsBase()
+    )
     extension: str | None = None
 
 

automated_security_helper/base/scanner_plugin.py

@@ -1,5 +1,6 @@
 from datetime import datetime, timezone
 import logging
+from automated_security_helper.base.options import ScannerOptionsBase
 from automated_security_helper.base.plugin_base import PluginBase
 from automated_security_helper.base.plugin_config import PluginConfigBase
 from automated_security_helper.core.enums import ScannerToolType
@@ -16,7 +17,9 @@
 
 
 class ScannerPluginConfigBase(PluginConfigBase):
-    pass
+    options: Annotated[ScannerOptionsBase, Field(description="Scanner options")] = (
+        ScannerOptionsBase()
+    )
 
 
 T = TypeVar("T", bound=ScannerPluginConfigBase)
@@ -221,6 +224,8 @@ def _post_scan(
             options: Optional scanner-specific options
         """
         self.end_time = datetime.now(timezone.utc)
+        if not self.start_time:
+            self.start_time = self.end_time
 
         # ec_color = "bold green" if self.exit_code == 0 else "bold red"
         self._plugin_log(

automated_security_helper/cli/config.py

@@ -1,6 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import logging
 import os
 from pathlib import Path
 import sys
@@ -12,6 +13,7 @@
 from automated_security_helper.config.resolve_config import resolve_config
 from automated_security_helper.core.constants import ASH_CONFIG_FILE_NAMES
 from automated_security_helper.core.exceptions import ASHConfigValidationError
+from automated_security_helper.utils.log import get_logger
 
 config_app = typer.Typer(
     name="config",
@@ -42,7 +44,15 @@ def init(
             help="Overwrite the config file if it already exists at the target path.",
         ),
     ] = False,
+    verbose: Annotated[bool, typer.Option(help="Enable verbose logging")] = False,
+    debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False,
+    color: Annotated[bool, typer.Option(help="Enable/disable colorized output")] = True,
 ):
+    get_logger(
+        level=(logging.DEBUG if debug else 15 if verbose else logging.INFO),
+        show_progress=False,
+        use_color=color,
+    )
     config_path_path = Path(config_path)
     if config_path_path.absolute().exists() and not force:
         typer.secho(
@@ -73,7 +83,15 @@ def get(
             help=f"The name of the config file to get. By default, ASH looks for the following config file names in the source directory of a scan: {ASH_CONFIG_FILE_NAMES}. If  a different filename is specified, it must be provided when running ASH via the `--config` option or by setting the `ASH_CONFIG` environment variable.",
         ),
     ] = None,
+    verbose: Annotated[bool, typer.Option(help="Enable verbose logging")] = False,
+    debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False,
+    color: Annotated[bool, typer.Option(help="Enable/disable colorized output")] = True,
 ):
+    get_logger(
+        level=(logging.DEBUG if debug else 15 if verbose else logging.INFO),
+        show_progress=False,
+        use_color=color,
+    )
     if config_path is not None and not Path(config_path).exists():
         typer.secho(f"Config file does not exist at {config_path}", fg=typer.colors.RED)
         raise typer.Exit(1)
@@ -100,7 +118,15 @@ def validate(
             help=f"The name of the config file to create. By default, ASH looks for the following config file names in the source directory of a scan: {ASH_CONFIG_FILE_NAMES}. If  a different filename is specified, it must be provided when running ASH via the `--config` option or by setting the `ASH_CONFIG` environment variable.",
         ),
     ] = None,
+    verbose: Annotated[bool, typer.Option(help="Enable verbose logging")] = False,
+    debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False,
+    color: Annotated[bool, typer.Option(help="Enable/disable colorized output")] = True,
 ):
+    get_logger(
+        level=(logging.DEBUG if debug else 15 if verbose else logging.INFO),
+        show_progress=False,
+        use_color=color,
+    )
     if config_path is not None and not Path(config_path).exists():
         typer.secho(f"Config file does not exist at {config_path}", fg=typer.colors.RED)
         raise typer.Exit(1)

automated_security_helper/cli/dependencies.py

@@ -7,7 +7,7 @@
 import platform
 import sys
 from pathlib import Path
-from typing import List, Optional
+from typing import Annotated, List, Optional
 
 import typer
 from rich import print
@@ -19,7 +19,7 @@
 from automated_security_helper.config.resolve_config import resolve_config
 from automated_security_helper.core.constants import ASH_BIN_PATH, ASH_WORK_DIR_NAME
 from automated_security_helper.plugins import ash_plugin_manager
-from automated_security_helper.utils.log import ASH_LOGGER
+from automated_security_helper.utils.log import get_logger
 
 dependencies_app = typer.Typer(
     name="dependencies",
@@ -69,16 +69,17 @@ def install_dependencies(
         "--bin-path",
         "-b",
         help="Path to install binaries to.",
+        envvar="ASH_BIN_PATH",
     ),
     plugin_types: List[str] = typer.Option(
         ["converter", "scanner", "reporter"],
         "--plugin-type",
         "-t",
         help="Plugin types to install dependencies for",
     ),
-    verbose: bool = typer.Option(
-        False, "--verbose", "-v", help="Enable verbose output"
-    ),
+    verbose: Annotated[bool, typer.Option(help="Enable verbose logging")] = False,
+    debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False,
+    color: Annotated[bool, typer.Option(help="Enable/disable colorized output")] = True,
 ) -> int:
     """Install dependencies for ASH plugins.
 
@@ -87,22 +88,20 @@ def install_dependencies(
 
     Binary tools will be installed to the specified bin path (defaults to ~/.ash/bin).
     """
+    # Set the ASH_BIN_PATH environment variable to override the default
+    import os
+
     # Set up logging
-    if verbose:
-        ASH_LOGGER.setLevel(logging.DEBUG)
-        handler = logging.StreamHandler()
-        handler.setLevel(logging.DEBUG)
-        ASH_LOGGER.addHandler(handler)
+    get_logger(
+        level=(logging.DEBUG if debug else 15 if verbose else logging.INFO),
+        show_progress=False,
+        use_color=color,
+    )
 
-    # Use provided bin path or default
+    # Use provided bin path or default if bin_path was null
     target_bin_path = bin_path or ASH_BIN_PATH
-
-    # Create bin directory if it doesn't exist
+    # Create target_bin_path directory if it doesn't exist
     target_bin_path.mkdir(parents=True, exist_ok=True)
-
-    # Set the ASH_BIN_PATH environment variable to override the default
-    import os
-
     os.environ["ASH_BIN_PATH"] = str(target_bin_path)
 
     console.print(

automated_security_helper/cli/plugin.py

@@ -9,7 +9,7 @@
 import logging
 import os
 from pathlib import Path
-from typing import Annotated
+from typing import Annotated, List
 import typer
 from rich.console import Console
 from rich.table import Table
@@ -49,6 +49,13 @@ def list_plugins(
         bool,
         typer.Option(help="Whether to include the plugin config in the response table"),
     ] = False,
+    ash_plugin_modules: Annotated[
+        List[str],
+        typer.Option(
+            help="List of Python modules to import containing ASH plugins and/or event subscribers. These are loaded in addition to the default modules.",
+            envvar="ASH_PLUGIN_MODULES",
+        ),
+    ] = [],
     verbose: Annotated[bool, typer.Option(help="Enable verbose logging")] = False,
     debug: Annotated[bool, typer.Option(help="Enable debug logging")] = False,
     color: Annotated[bool, typer.Option(help="Enable/disable colorized output")] = True,
@@ -83,10 +90,12 @@ def list_plugins(
 
     try:
         console = Console()
+        ash_config = resolve_config(config_path=config)
+        ash_config.ash_plugin_modules.extend(ash_plugin_modules)
         plugin_context = PluginContext(
             source_dir=Path.cwd(),
             output_dir=Path.cwd().joinpath(".ash", "ash_output"),
-            config=resolve_config(config_path=config),
+            config=ash_config,
         )
 
         # Load all plugins