Merge pull request #66 from iriusrisk/release/1.1.0

[release/1.1.0] to main

Author: dfernandezvigo

startleft/api/controllers/iac/file_type.py

@@ -34,7 +34,7 @@
 
 @router.post(URL, status_code=RESPONSE_STATUS_CODE, description="Generates an OTM threat model from an IaC file")
 def iac(iac_file: UploadFile = File(..., description="File that contains the Iac definition"),
-        iac_type: IacType = Form(..., description="Type of IaC File: CLOUDFORMATION"),
+        iac_type: IacType = Form(..., description="Type of IaC File: CLOUDFORMATION, TERRAFORM"),
         id: str = Form(..., description="ID of the new project"),
         name: str = Form(..., description="Name of the new project"),
         mapping_file: UploadFile = File(..., description="File that contains the mapping between IaC "

startleft/api/controllers/iac/iac_create_otm_controller.py

@@ -1,6 +1,16 @@
 from enum import Enum
 
+from startleft.config import paths
+
 
 class IacType(str, Enum):
-    JSON = 'CLOUDFORMATION'
+    CLOUDFORMATION = ("CLOUDFORMATION", "CloudFormation", paths.default_cf_mapping_file)
+    TERRAFORM = ("TERRAFORM", "Terraform", paths.default_tf_aws_mapping_file)
+
+    def __new__(cls, value, description: str = None, def_map_file=None):
+        obj = str.__new__(cls, [value])
+        obj._value_ = value
+        obj.description = description
+        obj.def_map_file = def_map_file
 
+        return obj

startleft/api/controllers/iac/iac_type.py

@@ -1,10 +1,13 @@
 import logging
+import re
 import sys
 
 import click
 
+from startleft.api.controllers.iac.iac_type import IacType
 from startleft.api.errors import CommonError
 from startleft.iac_to_otm import IacToOtm
+from startleft.messages import messages
 from startleft.project.iriusrisk_project_repository import IriusriskProjectRepository
 from startleft.project.otm_project import OtmProject
 from startleft.project.otm_project_service import OtmProjectService
@@ -14,18 +17,18 @@
 __version__ = "0.1.0"
 
 
-def validate_logging(ctx, param, value):
+def get_log_level(ctx, param, value):
     levels = {
-        'none': 100,
-        'crit': logging.CRITICAL,
-        'error': logging.ERROR,
-        'warn': logging.WARNING,
-        'info': logging.INFO,
-        'debug': logging.DEBUG
+        'NONE': 100,
+        'CRIT': logging.CRITICAL,
+        'ERROR': logging.ERROR,
+        'WARN': logging.WARNING,
+        'INFO': logging.INFO,
+        'DEBUG': logging.DEBUG
     }
 
-    if value.lower() in levels:
-        return levels[value.lower()]
+    if value.upper() in levels:
+        return levels[value.upper()]
     raise click.BadParameter(f"Log level must be one of: {', '.join(levels.keys())}")
 
 
@@ -37,6 +40,16 @@ def configure_logger(level, verbose):
         logging.basicConfig(format='%(message)s', level=level)
 
 
+def validate_server(ctx, param, value):
+    regex = "((http|https)://)(www.)?[a-zA-Z0-9@:%._\\+~#?&//=]{2,256}(\\.[a-z]{2,6}\\b([-a-zA-Z0-9@:%._\\+~#?&//=]*))?(:[0-9]+)?"
+    p = re.compile(regex)
+
+    if value is None or not re.search(p, value):
+        raise click.BadParameter("IriusRisk host must follow the convention 'proto://server[:port]'")
+
+    return value
+
+
 class CatchAllExceptions(click.Group):
     def __call__(self, *args, **kwargs):
         try:
@@ -49,9 +62,10 @@ def __call__(self, *args, **kwargs):
 
 
 @click.group(cls=CatchAllExceptions)
-@click.option('--log-level', '-l', callback=validate_logging, default='info',
-              help='Set the log level. Must be one of: crit, error, warn, info, debug, none.')
-@click.option('--verbose/--no-verbose', default=False, help='Makes logging more verbose.')
+@click.option('--log-level', '-l',
+              type=click.Choice(['CRIT', 'ERROR', 'WARN', 'INFO', 'DEBUG', 'NONE'], case_sensitive=False),
+              callback=get_log_level, default='info', help='Set the log level.')
+@click.option('--verbose/--no-verbose', '-v/-nv', default=False, help='Makes logging more verbose.')
 @click.version_option(__version__)
 def cli(log_level, verbose):
     """
@@ -61,100 +75,90 @@ def cli(log_level, verbose):
 
 
 @cli.command()
-@click.option('--type', '-t',
-              type=click.Choice(['JSON', 'YAML', 'CloudFormation', 'HCL2', 'Terraform', 'XML'], case_sensitive=False),
-              default="JSON",
-              help='Specify the source file type.')
-@click.option('--map', '-m', help='Map file to use when parsing source files')
-@click.option('--otm', '-o', default='threatmodel.otm', help='OTM output file name')
-@click.option('--name', '-n', help='Project name')
-@click.option('--id', help='Project ID')
-@click.option('--recreate/--no-recreate', default=False, help='Delete and recreate the project each time')
-@click.option('--irius-server', default='', envvar='IRIUS_SERVER',
-              help='IriusRisk server to connect to (proto://server[:port])')
-@click.option('--api-token', default='', envvar='IRIUS_API_TOKEN', help='IriusRisk API token')
-@click.argument('filename', nargs=-1)
-def run(type, map, otm, name, id, recreate, irius_server, api_token, filename):
+@click.option(messages.IAC_TYPE_NAME, messages.IAC_TYPE_SHORTNAME, type=click.Choice(messages.IAC_TYPE_SUPPORTED, case_sensitive=False), required=True, help=messages.IAC_TYPE_DESC)
+@click.option(messages.MAPPING_FILE_NAME, messages.MAPPING_FILE_SHORTNAME, help=messages.MAPPING_FILE_DESC)
+@click.option(messages.OUTPUT_FILE_NAME, messages.OUTPUT_FILE_SHORTNAME, default=messages.OUTPUT_FILE, help=messages.OUTPUT_FILE_DESC)
+@click.option(messages.PROJECT_NAME_NAME, messages.PROJECT_NAME_SHORTNAME, required=True, help=messages.PROJECT_NAME_DESC)
+@click.option(messages.PROJECT_ID_NAME, messages.PROJECT_ID_SHORTNAME, required=True, help=messages.PROJECT_ID_DESC)
+@click.option(messages.RECREATE_NAME, messages.RECREATE_SHORTNAME, default=False, help=messages.RECREATE_DESC)
+@click.option(messages.IRIUS_SERVER_NAME, messages.IRIUS_SERVER_SHORTNAME, default='', envvar=messages.IRIUS_SERVER_ENVAR, callback=validate_server, help=messages.IRIUS_SERVER_DESC)
+@click.option(messages.API_TOKEN_NAME, messages.API_TOKEN_SHORTNAME, default='', envvar=messages.API_TOKEN_ENVAR, help=messages.API_TOKEN_DESC)
+@click.argument(messages.IAC_FILE_NAME, required=True, nargs=-1)
+def run(iac_type, mapping_file, output_file, project_name, project_id, recreate, irius_server, api_token,
+        iac_file):
     """
-    Parses IaC source files into Open Threat Model and immediately uploads threat model to IriusRisk
+    Parses an IaC file into an Open Threat Model (OTM) and uploads it to IriusRisk
     """
     logger.info("Parsing IaC source files into OTM")
-    otm_project = OtmProject.from_iac_file(id, name, type, filename, map, otm)
+    otm_project = OtmProject.from_iac_file(project_id, project_name, IacType(iac_type.upper()), iac_file,
+                                           mapping_file, output_file)
     otm_service = OtmProjectService(IriusriskProjectRepository(api_token, irius_server))
 
     __create_or_recreate_otm_project(recreate, otm_service, otm_project)
 
 
 @cli.command()
-@click.option('--type', '-t',
-              type=click.Choice(['JSON', 'YAML', 'CloudFormation', 'HCL2', 'Terraform', 'XML'], case_sensitive=False),
-              default="JSON",
-              help='Specify the source file type.')
-@click.option('--map', '-m', help='Map file to use when parsing source files')
-@click.option('--otm', '-o', default='threatmodel.otm', help='OTM output file name')
-@click.option('--name', '-n', help='Project name')
-@click.option('--id', help='Project ID')
-@click.argument('filename', nargs=-1)
-def parse(type, map, otm, name, id, filename):
+@click.option(messages.IAC_TYPE_NAME, messages.IAC_TYPE_SHORTNAME, type=click.Choice(messages.IAC_TYPE_SUPPORTED, case_sensitive=False), required=True, help=messages.IAC_TYPE_DESC)
+@click.option(messages.MAPPING_FILE_NAME, messages.MAPPING_FILE_SHORTNAME, help=messages.MAPPING_FILE_DESC)
+@click.option(messages.OUTPUT_FILE_NAME, messages.OUTPUT_FILE_SHORTNAME, default=messages.OUTPUT_FILE, help=messages.OUTPUT_FILE_DESC)
+@click.option(messages.PROJECT_NAME_NAME, messages.PROJECT_NAME_SHORTNAME, required=True, help=messages.PROJECT_NAME_DESC)
+@click.option(messages.PROJECT_ID_NAME, messages.PROJECT_ID_SHORTNAME, required=True, help=messages.PROJECT_ID_DESC)
+@click.argument(messages.IAC_FILE_NAME, required=True, nargs=-1)
+def parse(iac_type, mapping_file, output_file, project_name, project_id, iac_file):
     """
     Parses IaC source files into Open Threat Model
     """
     logger.info("Parsing IaC source files into OTM")
-    OtmProject.from_iac_file(id, name, type, filename, map, otm)
+    OtmProject.from_iac_file(project_id, project_name, IacType(iac_type.upper()), iac_file, mapping_file, output_file)
 
 
 @cli.command()
-@click.option('--recreate/--no-recreate', default=False, help='Delete and recreate the project each time')
-@click.option('--irius-server', default='', envvar='IRIUS_SERVER',
-              help='IriusRisk server to connect to (proto://server[:port])')
-@click.option('--api-token', default='', envvar='IRIUS_API_TOKEN', help='IriusRisk API token')
-@click.argument('filename', nargs=-1)
-def threatmodel(recreate, irius_server, api_token, filename):
+@click.option(messages.RECREATE_NAME, messages.RECREATE_SHORTNAME, default=False, help=messages.RECREATE_DESC)
+@click.option(messages.IRIUS_SERVER_NAME, messages.IRIUS_SERVER_SHORTNAME, default='', envvar=messages.IRIUS_SERVER_ENVAR, callback=validate_server, help=messages.IRIUS_SERVER_DESC)
+@click.option(messages.API_TOKEN_NAME, messages.API_TOKEN_SHORTNAME, default='', envvar=messages.API_TOKEN_ENVAR, help=messages.API_TOKEN_DESC)
+@click.argument(messages.IAC_FILE_NAME, required=True, nargs=-1)
+def threatmodel(recreate, irius_server, api_token, iac_file):
     """
     Uploads an OTM file to IriusRisk
     """
-    otm_project = OtmProject.from_otm_file(filename)
+    otm_project = OtmProject.from_otm_file(iac_file)
     otm_service = OtmProjectService(IriusriskProjectRepository(api_token, irius_server))
 
     __create_or_recreate_otm_project(recreate, otm_service, otm_project)
 
 
 @cli.command()
-@click.option('--map', '-m', multiple=True, help='Map file to validate')
-@click.option('--otm', '-o', multiple=True, help='OTM file to validate')
-def validate(map, otm):
+@click.option(messages.MAPPING_FILE_NAME, messages.MAPPING_FILE_SHORTNAME, help=messages.MAPPING_FILE_DESC)
+@click.option(messages.OUTPUT_FILE_NAME, messages.OUTPUT_FILE_SHORTNAME, default=messages.OUTPUT_FILE, help=messages.OUTPUT_FILE_DESC)
+def validate(mapping_file, output_file):
     """
     Validates a mapping or OTM file
     """
-    if map:
+    if mapping_file:
         logger.info("Validating IaC mapping files")
-        OtmProject.validate_iac_mappings_file(map)
+        OtmProject.validate_iac_mappings_file(mapping_file)
 
-    if otm:
+    if output_file:
         logger.info("Validating OTM file")
-        OtmProject.load_and_validate_otm_file(otm)
+        OtmProject.load_and_validate_otm_file(output_file)
 
 
 @cli.command()
-@click.option('--type', '-t',
-              type=click.Choice(['JSON', 'YAML', 'CloudFormation', 'HCL2', 'Terraform', 'XML'], case_sensitive=False),
-              default="JSON",
-              help='Specify the source file type.')
-@click.option('--query', help='JMESPath query to run against source files')
-@click.argument('filename', nargs=-1)
-def search(type, query, filename):
+@click.option(messages.IAC_TYPE_NAME, messages.IAC_TYPE_SHORTNAME, type=click.Choice(messages.IAC_TYPE_SUPPORTED, case_sensitive=False), required=True, help=messages.IAC_TYPE_DESC)
+@click.option('--query', '-q', help='JMESPath query to run against the IaC file.')
+@click.argument(messages.IAC_FILE_NAME, required=True, nargs=-1)
+def search(iac_type, query, iac_file):
     """
     Searches source files for the given query
     """
 
-    iac_to_otm = IacToOtm(None, None)
-    logger.info("Running JMESPath search query against source files")
-    iac_to_otm.search(type, query, filename)
+    iac_to_otm = IacToOtm(None, None, IacType(iac_type.upper()))
+    logger.info("Running JMESPath search query against the IaC file")
+    iac_to_otm.search(IacType(iac_type.upper()), query, iac_file)
 
 
-@click.option('--port', default=5000, envvar='application port',
-              help='The port to deploy this application to')
 @cli.command()
+@click.option('--port', '-p', default=5000, envvar='STARTLEFT_PORT', help='Startleft deployment port.')
 def server(port: int):
     """
     Launches the REST server to generate OTMs from requests