<s> import os <EOL> import socket <EOL> import stat <EOL> def validate_ssh_user ( ssh_user ) : <EOL> assert ssh_user , '<STR_LIT>' <EOL> assert isinstance ( ssh_user , str ) , '<STR_LIT>' <EOL> def validate_ssh_key_path ( ssh_key_path ) : <EOL> assert isinstance ( ssh_key_path , str ) , '<STR_LIT>' <EOL> assert ssh_key_path , '<STR_LIT>' <EOL> assert os . path . isfile ( ssh_key_path ) , '<STR_LIT>' . format ( ssh_key_path ) <EOL> assert stat . S_IMODE ( <EOL> os . stat ( ssh_key_path ) . st_mode ) & ( stat . S_IRWXG + stat . S_IRWXO ) == <NUM_LIT:0> , ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> with open ( ssh_key_path ) as fh : <EOL> assert '<STR_LIT>' not in fh . read ( ) , ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def validate_ssh_port ( ssh_port ) : <EOL> assert isinstance ( ssh_port , int ) , '<STR_LIT>' <EOL> assert <NUM_LIT:1> <= ssh_port <= <NUM_LIT> , '<STR_LIT>' <EOL> def validate_hosts_list ( agent_list ) : <EOL> assert agent_list , '<STR_LIT>' <EOL> assert isinstance ( agent_list , list ) , '<STR_LIT>' <EOL> assert len ( agent_list ) > <NUM_LIT:0> , '<STR_LIT>' <EOL> check_duplicates ( agent_list ) <EOL> check_ipv4_addrs ( agent_list ) <EOL> def validate_master_agent_lists ( master_list , agent_list ) : <EOL> validate_hosts_list ( master_list ) <EOL> assert isinstance ( agent_list , list ) , '<STR_LIT>' <EOL> assert len ( agent_list ) > <NUM_LIT:0> , '<STR_LIT>' <EOL> compare_lists ( master_list , agent_list ) <EOL> def check_duplicates ( arg_list ) : <EOL> assert isinstance ( arg_list , list ) , '<STR_LIT>' <EOL> dups = list ( filter ( lambda x : arg_list . count ( x ) > <NUM_LIT:1> , arg_list ) ) <EOL> assert len ( dups ) == <NUM_LIT:0> , '<STR_LIT>' . format ( '<STR_LIT:U+002CU+0020>' . join ( set ( dups ) ) ) <EOL> def compare_lists ( first_list , second_list ) : <EOL> assert isinstance ( first_list , list ) , '<STR_LIT>' <EOL> assert isinstance ( second_list , list ) , '<STR_LIT>' <EOL> dups = set ( first_list ) & set ( second_list ) <EOL> assert len ( dups ) == <NUM_LIT:0> , '<STR_LIT>' . format ( '<STR_LIT:U+002CU+0020>' . join ( dups ) ) <EOL> def check_ipv4_addrs ( ips ) : <EOL> assert isinstance ( ips , list ) <EOL> invalid_ips = [ ] <EOL> for ip in ips : <EOL> try : <EOL> socket . inet_pton ( socket . AF_INET , str ( ip ) ) <EOL> except OSError : <EOL> invalid_ips . append ( ip ) <EOL> assert not len ( invalid_ips ) , ( '<STR_LIT>' <EOL> '<STR_LIT:{}>' . format ( invalid_ips ) ) <EOL> def validate_config ( config ) : <EOL> assert isinstance ( config , dict ) <EOL> ssh_keys_checks_map = { <EOL> '<STR_LIT>' : validate_ssh_user , <EOL> '<STR_LIT>' : validate_ssh_port , <EOL> '<STR_LIT>' : validate_ssh_key_path , <EOL> '<STR_LIT>' : validate_hosts_list , <EOL> '<STR_LIT>' : lambda master_list : validate_master_agent_lists ( master_list , config . get ( '<STR_LIT>' ) ) <EOL> } <EOL> errors = { } <EOL> for ssh_key , validate_func in ssh_keys_checks_map . items ( ) : <EOL> input_value = config . get ( ssh_key ) <EOL> if not input_value : <EOL> errors [ ssh_key ] = '<STR_LIT>' . format ( ssh_key ) <EOL> continue <EOL> try : <EOL> validate_func ( input_value ) <EOL> except AssertionError as err : <EOL> errors [ ssh_key ] = str ( err ) <EOL> return errors </s>
