def scan_for_stage()

in src/helper.py [0:0]

• 31 lines of code
• 8 McCabe index (conditional complexity)

def scan_for_stage(dynamodb_item_stage, cp_stages):
    """This will compare the DynamoDB Stage Item against all CodePipeline Stages

    Args:
        dynamodb_item_stage (dict): Security / governance rule policy to scan against
        cp_stages (list): AWS CodePipeline Stages to be scanned

    Returns:
        :obj: Returns scan results
    """
    logger.info("Executing Stage Scan")
    dyn_item_actions = dynamodb_item_stage.get('Actions')
    dyn_item_actions_results = {}

    logger.debug(f"dynamodb_item_stage:{dynamodb_item_stage}")
    logger.debug(f"cp_stages:{cp_stages}")
    logger.debug(f"dyn_item_actions:{dyn_item_actions}")

    # Parse each stage in the CodePipeline Template
    for cp_stage in cp_stages:
        if cp_stage and (cp_stage["Name"] == dynamodb_item_stage.get("Name")):
            logger.info("Found Matching Stage Name, Checking Action Configuration")

            # Parse each action in the CodePipeline Template
            for cp_action in cp_stage['Actions']:

                # If DynamoDB doesn't have actions but Stage matched, PASS
                if not dyn_item_actions:
                    logger.info("DynamoDB Item has no actions")
                    return True

                else:
                    # Check each DynamoDB Action against the CodePipeline Actions, since we could have multiple actions
                    #  in the DynamoDB item we are setting the results of each action within a dictionary
                    #  (example {'Scan-CodePipeline': True, 'Update-CodePipeline': False}) if values are False, FAIL
                    for dyn_item_action in dyn_item_actions:
                        logger.info(f"dyn_item_action:{dyn_item_action}")
                        logger.info(f"cp_action:{cp_action}")

                        if dyn_item_action.get('Configuration', {}).items() <= cp_action['Configuration'].items() \
                                and dyn_item_action.get('ActionTypeId', {}).items() <= cp_action['ActionTypeId'].items() \
                                and dyn_item_action.get('Name', {}) == cp_action['Name']:
                            dyn_item_actions_results.update({dyn_item_action['Name']: True})

                        else:
                            if not dyn_item_actions_results.get(dyn_item_action['Name'], False):
                                dyn_item_actions_results.update({dyn_item_action['Name']: False})

            logger.debug(dyn_item_actions_results)

            if [_v for _k, _v in dyn_item_actions_results.items() if _v is False]:
                return False
            else:
                return True

    return False