def create()

in aws_sra_examples/solutions/guardduty/guardduty_org/lambda/src/app.py [0:0]

• 44 lines of code
• 7 McCabe index (conditional complexity)

def create(event, context):
    """
    CloudFormation Create Event.
    :param event: event data
    :param context: runtime information
    :return: GuardDutyResourceId
    """
    request_type = event["RequestType"]
    logger.info(f"{request_type} Event")

    try:
        check_parameters(event)
        params = event.get("ResourceProperties")

        # Required to enable GuardDuty in the Org Management account from the delegated admin
        create_service_linked_role(SERVICE_ROLE_NAME, SERVICE_NAME)

        available_regions = get_available_service_regions(params.get("ENABLED_REGIONS", ""), "guardduty")

        enable_organization_admin_account(params.get("DELEGATED_ADMIN_ACCOUNT_ID", ""), available_regions)
        session = assume_role(
            params.get("DELEGATED_ADMIN_ACCOUNT_ID", ""),
            params.get("AWS_PARTITION", "aws"),
            params.get("CONFIGURATION_ROLE_NAME", ""),
            "CreateGuardDuty"
        )
        detectors_exist = False
        run_count = 0

        while not detectors_exist and run_count < MAX_RUN_COUNT:
            run_count += 1
            detectors_exist = check_for_detectors(session, available_regions)
            logger.info(f"All Detectors Exist: {detectors_exist} Count: {run_count}")
            if not detectors_exist:
                time.sleep(SLEEP_SECONDS)

        if detectors_exist:
            auto_enable_s3_logs = (params.get("AUTO_ENABLE_S3_LOGS", "false")).lower() in "true"

            configure_guardduty(
                session,
                params.get("DELEGATED_ADMIN_ACCOUNT_ID", ""),
                auto_enable_s3_logs,
                available_regions,
                params.get("FINDING_PUBLISHING_FREQUENCY", "FIFTEEN_MINUTES"),
                params.get("KMS_KEY_ARN", ""),
                params.get("PUBLISHING_DESTINATION_BUCKET_ARN", "")
            )
        else:
            raise ValueError(
                "GuardDuty Detectors did not get created in the allowed time. "
                "Check the Org Management delegated admin setup."
            )
    except Exception as exc:
        logger.error(f"Unexpected error {exc}")
        raise ValueError("Unexpected error. Review logs for details.")

    if request_type == "Create":
        return "GuardDutyResourceId"