def check_s3_exfil()

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

• 20 lines of code
• 4 McCabe index (conditional complexity)

def check_s3_exfil(sh_resource, ddbtable):
    s3_exfil_unusual = 'TTPs/Exfiltration:S3-ObjectRead.Unusual'
    s3_exfil_malicious_ip = 'TTPs/Exfiltration:S3-MaliciousIPCaller'
    try:
        s3_exfil_payload = ddbtable.query(
            IndexName= DYNAMODB_GSI_TYPE,
            KeyConditionExpression=Key('ResourceId').eq(sh_resource) & Key('Types').eq(s3_exfil_unusual)
        )
        s3_exfil_malicious_ip_payload = ddbtable.query(
            IndexName= DYNAMODB_GSI_TYPE,
            KeyConditionExpression=Key('ResourceId').eq(sh_resource) & Key('Types').eq(s3_exfil_malicious_ip)
        )
        if s3_exfil_payload ['Count'] >= 1:
            logger.info('Found GuardDuty finding with unusual reads on S3 bucket {}.'.format(sh_resource))
            return s3_exfil_payload
        elif s3_exfil_malicious_ip_payload ['Count'] >= 1:
            logger.info('Found GuardDuty finding for actions from malicious IPs on S3 bucket {}.'.format(sh_resource))
            return s3_exfil_malicious_ip_payload
    except ClientError as error_handle:
        logger.error(error_handle.dynamodb_match['Error']['Code'])