def network_correlation()

in security_hub_correlation_cdk/lambdas/create_sh_finding/create_sh_finding.py [0:0]

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

def network_correlation(sh_resource, ddbtable):
    try:
        logger.info('CHECK#1: Security Hub exposed IP or unrestricted SG & GuardDuty Finding for Unusual Network port and Brute force attack for {}...'.format(sh_resource))
        check_sh_ec2_public_payload = check_sh_ec2_public(sh_resource, ddbtable)
        network_payload = check_network_unusual(sh_resource, ddbtable)
        ssh_brute_force_payload = check_ssh_brute_force(sh_resource, ddbtable)
        winrm_brute_forice_payload = check_winrm_brute_force(sh_resource, ddbtable)
        rdp_brute_force_payload = check_rdp_brute_force(sh_resource, ddbtable)
        if ((check_sh_ec2_public_payload) and network_payload and (ssh_brute_force_payload or rdp_brute_force_payload or winrm_brute_forice_payload)):
            logger.info('Match found for Security Hub exposed IP or unrestricted SG & GuardDuty Brute force and Unusual Network Port for {}.'.format(sh_resource))
            SH_title = {"SH_Title":'Unusual Network port and Brute force found for possibly exposed EC2 instance {}'.format(sh_resource)}
            SourceUrls = []
            SourceUrls.append(check_sh_ec2_public_payload['Items'][0]['SourceUrl'])
            SourceUrls.append(network_payload['Items'][0]['SourceUrl'])
            SourceUrlList = {"SourceUrlList": SourceUrls}
            network_payload.update(SourceUrlList)
            network_payload.update(SH_title)
        else:
            logger.info('No matches found for Security Hub exposed IP or SG & GuardDuty Unusual Network and RDS/SSH Brute force for {}.'.format(sh_resource))
    except ClientError as error_handle:
        logger.error(error_handle.dynamodb_match['Error']['Code'])