1 % Regression tests for Linux only 2 3 # More informations at http://www.secdev.org/projects/UTscapy/ 4 5 6 ############ 7 ############ 8 9 + Linux only test 10 11 = TCP client automaton 12 ~ automaton netaccess linux needs_root 13 * This test retries on failure because it often fails 14 15 from __future__ import print_function 16 import os 17 import time 18 import signal 19 20 from scapy.modules.six.moves import range 21 22 def handler(signum, stack_frame): 23 raise Exception("Timer expired !") 24 25 tmp = signal.signal(signal.SIGALRM, handler) 26 27 SECDEV_IP4 = "203.178.141.194" 28 IPTABLE_RULE = "iptables -%c INPUT -s %s -p tcp --sport 80 -j DROP" 29 30 # Drop packets from SECDEV_IP4 31 assert(os.system(IPTABLE_RULE % ('A', SECDEV_IP4)) == 0) 32 33 success = False 34 for i in range(10): 35 tmp = signal.alarm(5) 36 try: 37 r, w = os.pipe() 38 t = TCP_client(SECDEV_IP4, 80, external_fd={ "tcp": (r,w) }) 39 tmp = os.write(w, b"HEAD / HTTP/1.0\r\n\r\n") 40 t.runbg() 41 time.sleep(0.5) 42 response = os.read(r, 4096) 43 tmp = signal.alarm(0) # cancel the alarm 44 t.stop() 45 os.close(r) 46 os.close(w) 47 if response.startswith(b"HTTP/1.1 200 OK"): 48 success = True 49 break 50 else: 51 time.sleep(0.5) 52 except Exception as e: 53 print(e) 54 55 # Remove the iptables rule 56 assert(os.system(IPTABLE_RULE % ('D', SECDEV_IP4)) == 0) 57 58 assert(success) 59 60 = L3RawSocket 61 ~ netaccess IP ICMP linux needs_root 62 63 old_l3socket = conf.L3socket 64 old_debug_dissector = conf.debug_dissector 65 conf.debug_dissector = False 66 conf.L3socket = L3RawSocket 67 x = sr1(IP(dst="www.google.com")/ICMP(),timeout=3) 68 conf.debug_dissector = old_debug_dissector 69 conf.L3socket = old_l3socket 70 x 71 assert x[IP].ottl() in [32, 64, 128, 255] 72 assert 0 <= x[IP].hops() <= 126 73 x is not None and ICMP in x and x[ICMP].type == 0 74 75 # TODO: fix this test (randomly stuck) 76 # ex: https://travis-ci.org/secdev/scapy/jobs/247473497 77 78 #= Supersocket _flush_fd 79 #~ needs_root linux 80 # 81 #import select 82 # 83 #from scapy.arch.linux import _flush_fd 84 #socket = conf.L2listen() 85 #select.select([socket],[],[],2) 86 #_flush_fd(socket.ins) 87 88 = Test legacy attach_filter function 89 ~ linux needs_root 90 from scapy.arch.common import get_bpf_pointer 91 92 old_pypy = conf.use_pypy 93 conf.use_pypy = True 94 95 tcpdump_lines = ['12\n', '40 0 0 12\n', '21 0 5 34525\n', '48 0 0 20\n', '21 6 0 6\n', '21 0 6 44\n', '48 0 0 54\n', '21 3 4 6\n', '21 0 3 2048\n', '48 0 0 23\n', '21 0 1 6\n', '6 0 0 1600\n', '6 0 0 0\n'] 96 pointer = get_bpf_pointer(tcpdump_lines) 97 assert six.PY3 or isinstance(pointer, str) 98 assert six.PY3 or len(pointer) > 1 99 100 conf.use_pypy = old_pypy 101 102 = Interface aliases & sub-interfaces 103 ~ linux needs_root 104 105 import os 106 exit_status = os.system("ip link add name scapy0 type dummy") 107 exit_status = os.system("ip addr add 192.0.2.1/24 dev scapy0") 108 exit_status = os.system("ip link set scapy0 up") 109 exit_status = os.system("ifconfig scapy0:0 inet 198.51.100.1/24 up") 110 exit_status = os.system("ip addr show scapy0") 111 print(get_if_list()) 112 conf.route.resync() 113 print(conf.route.routes) 114 assert(conf.route.route("198.51.100.254") == ("scapy0", "198.51.100.1", "0.0.0.0")) 115 route_alias = (3325256704, 4294967040, "0.0.0.0", "scapy0", "198.51.100.1", 0) 116 assert(route_alias in conf.route.routes) 117 exit_status = os.system("ip link add link scapy0 name scapy0.42 type vlan id 42") 118 exit_status = os.system("ip addr add 203.0.113.42/24 dev scapy0.42") 119 exit_status = os.system("ip link set scapy0.42 up") 120 exit_status = os.system("ip route add 192.0.2.43/32 via 203.0.113.41") 121 print(get_if_list()) 122 conf.route.resync() 123 print(conf.route.routes) 124 assert(conf.route.route("192.0.2.43") == ("scapy0.42", "203.0.113.42", "203.0.113.41")) 125 route_specific = (3221226027, 4294967295, "203.0.113.41", "scapy0.42", "203.0.113.42", 0) 126 assert(route_specific in conf.route.routes) 127 exit_status = os.system("ip link del name dev scapy0") 128