<s> """<STR_LIT>""" <EOL> import json <EOL> from security_monkey . datastore import NetworkWhitelistEntry , Account <EOL> from security_monkey . tests import SecurityMonkeyTestCase <EOL> from security_monkey import db <EOL> from security_monkey . watchers . elasticsearch_service import ElasticSearchServiceItem <EOL> CONFIG_ONE = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_TWO = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_THREE = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_FOUR = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_FIVE = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_SIX = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_SEVEN = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_EIGHT = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> CONFIG_NINE = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : json . loads ( b"""<STR_LIT>""" ) <EOL> } <EOL> WHITELIST_CIDRS = [ <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , "<STR_LIT>" ) , <EOL> ] <EOL> class ElasticSearchServiceTestCase ( SecurityMonkeyTestCase ) : <EOL> def setUp ( self ) : <EOL> self . es_items = [ <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_ONE ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_TWO ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_THREE ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_FOUR ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_FIVE ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_SIX ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_SEVEN ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_EIGHT ) , <EOL> ElasticSearchServiceItem ( region = "<STR_LIT>" , account = "<STR_LIT>" , name = "<STR_LIT>" , config = CONFIG_NINE ) , <EOL> ] <EOL> test_account = Account ( ) <EOL> test_account . name = "<STR_LIT>" <EOL> test_account . notes = "<STR_LIT>" <EOL> test_account . s3_name = "<STR_LIT>" <EOL> test_account . number = "<STR_LIT>" <EOL> test_account . role_name = "<STR_LIT>" <EOL> db . session . add ( test_account ) <EOL> db . session . commit ( ) <EOL> def tearDown ( self ) : <EOL> test_account = Account . query . filter ( Account . number == "<STR_LIT>" ) . first ( ) <EOL> if test_account is not None : <EOL> db . session . delete ( test_account ) <EOL> db . session . commit ( ) <EOL> def test_es_auditor ( self ) : <EOL> from security_monkey . auditors . elasticsearch_service import ElasticSearchServiceAuditor <EOL> es_auditor = ElasticSearchServiceAuditor ( accounts = [ "<STR_LIT>" ] ) <EOL> es_auditor . network_whitelist = [ ] <EOL> for cidr in WHITELIST_CIDRS : <EOL> whitelist_cidr = NetworkWhitelistEntry ( ) <EOL> whitelist_cidr . cidr = cidr [ <NUM_LIT:1> ] <EOL> whitelist_cidr . name = cidr [ <NUM_LIT:0> ] <EOL> es_auditor . network_whitelist . append ( whitelist_cidr ) <EOL> for es_domain in self . es_items : <EOL> es_auditor . check_es_access_policy ( es_domain ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:0> ] . audit_issues ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:0> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:20> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:1> ] . audit_issues ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:1> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:20> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:2> ] . audit_issues ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:2> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:5> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:2> ] . audit_issues [ <NUM_LIT:1> ] . score , <NUM_LIT:7> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:3> ] . audit_issues ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:3> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:20> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:4> ] . audit_issues ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:5> ] . audit_issues ) , <NUM_LIT:0> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:6> ] . audit_issues ) , <NUM_LIT:3> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:6> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:5> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:6> ] . audit_issues [ <NUM_LIT:1> ] . score , <NUM_LIT:5> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:6> ] . audit_issues [ <NUM_LIT:2> ] . score , <NUM_LIT:7> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:7> ] . audit_issues ) , <NUM_LIT:1> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:7> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:20> ) <EOL> self . assertEquals ( len ( self . es_items [ <NUM_LIT:8> ] . audit_issues ) , <NUM_LIT:2> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:8> ] . audit_issues [ <NUM_LIT:0> ] . score , <NUM_LIT:6> ) <EOL> self . assertEquals ( self . es_items [ <NUM_LIT:8> ] . audit_issues [ <NUM_LIT:1> ] . score , <NUM_LIT:10> ) </s>
