<s> """<STR_LIT>""" <EOL> from security_monkey . auditor import Auditor <EOL> from security_monkey . watchers . rds_security_group import RDSSecurityGroup <EOL> from security_monkey . datastore import NetworkWhitelistEntry <EOL> from security_monkey . auditors . security_group import _check_rfc_1918 <EOL> import ipaddr <EOL> class RDSSecurityGroupAuditor ( Auditor ) : <EOL> index = RDSSecurityGroup . index <EOL> i_am_singular = RDSSecurityGroup . i_am_singular <EOL> i_am_plural = RDSSecurityGroup . i_am_plural <EOL> network_whitelist = [ ] <EOL> def __init__ ( self , accounts = None , debug = False ) : <EOL> super ( RDSSecurityGroupAuditor , self ) . __init__ ( accounts = accounts , debug = debug ) <EOL> def prep_for_audit ( self ) : <EOL> self . network_whitelist = NetworkWhitelistEntry . query . all ( ) <EOL> def _check_inclusion_in_network_whitelist ( self , cidr ) : <EOL> for entry in self . network_whitelist : <EOL> if ipaddr . IPNetwork ( cidr ) in ipaddr . IPNetwork ( str ( entry . cidr ) ) : <EOL> return True <EOL> return False <EOL> def check_rds_ec2_rfc1918 ( self , sg_item ) : <EOL> """<STR_LIT>""" <EOL> tag = "<STR_LIT>" <EOL> severity = <NUM_LIT:8> <EOL> if sg_item . config . get ( "<STR_LIT>" , None ) : <EOL> return <EOL> for ipr in sg_item . config . get ( "<STR_LIT>" , [ ] ) : <EOL> cidr = ipr . get ( "<STR_LIT>" , None ) <EOL> if cidr and _check_rfc_1918 ( cidr ) : <EOL> self . add_issue ( severity , tag , sg_item , notes = cidr ) <EOL> def check_securitygroup_large_subnet ( self , sg_item ) : <EOL> """<STR_LIT>""" <EOL> tag = "<STR_LIT>" <EOL> severity = <NUM_LIT:3> <EOL> for ipr in sg_item . config . get ( "<STR_LIT>" , [ ] ) : <EOL> cidr = ipr . get ( "<STR_LIT>" , None ) <EOL> if cidr and not self . _check_inclusion_in_network_whitelist ( cidr ) : <EOL> if '<STR_LIT:/>' in cidr and not cidr == "<STR_LIT>" and not cidr == "<STR_LIT>" : <EOL> mask = int ( cidr . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] ) <EOL> if mask < <NUM_LIT> and mask > <NUM_LIT:0> : <EOL> self . add_issue ( severity , tag , sg_item , notes = cidr ) <EOL> def check_securitygroup_zero_subnet ( self , sg_item ) : <EOL> """<STR_LIT>""" <EOL> tag = "<STR_LIT>" <EOL> severity = <NUM_LIT:10> <EOL> for ipr in sg_item . config . get ( "<STR_LIT>" , [ ] ) : <EOL> cidr = ipr . get ( "<STR_LIT>" , None ) <EOL> if cidr and '<STR_LIT:/>' in cidr and not cidr == "<STR_LIT>" and not cidr == "<STR_LIT>" : <EOL> mask = int ( cidr . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] ) <EOL> if mask == <NUM_LIT:0> : <EOL> self . add_issue ( severity , tag , sg_item , notes = cidr ) <EOL> def check_securitygroup_any ( self , sg_item ) : <EOL> """<STR_LIT>""" <EOL> tag = "<STR_LIT>" <EOL> severity = <NUM_LIT:5> <EOL> for ipr in sg_item . config . get ( "<STR_LIT>" , [ ] ) : <EOL> cidr = ipr . get ( "<STR_LIT>" ) <EOL> if "<STR_LIT>" == cidr : <EOL> self . add_issue ( severity , tag , sg_item , notes = cidr ) <EOL> return <EOL> def check_securitygroup_10net ( self , sg_item ) : <EOL> """<STR_LIT>""" <EOL> tag = "<STR_LIT>" <EOL> severity = <NUM_LIT:5> <EOL> for ipr in sg_item . config . get ( "<STR_LIT>" , [ ] ) : <EOL> cidr = ipr . get ( "<STR_LIT>" ) <EOL> if "<STR_LIT>" == cidr : <EOL> self . add_issue ( severity , tag , sg_item , notes = cidr ) <EOL> return </s>
