| % Regression tests for Linux only |
| |
| # More information at http://www.secdev.org/projects/UTscapy/ |
| |
| |
| ############ |
| ############ |
| |
| + Linux only test |
| |
| = L3RawSocket |
| ~ netaccess IP TCP linux needs_root |
| |
| with no_debug_dissector(): |
| x = sr1(IP(dst="www.google.com")/TCP(sport=RandShort(), dport=80, flags="S"),timeout=3) |
| |
| x |
| assert x[IP].ottl() in [32, 64, 128, 255] |
| assert 0 <= x[IP].hops() <= 126 |
| |
| # TODO: fix this test (randomly stuck) |
| # ex: https://travis-ci.org/secdev/scapy/jobs/247473497 |
| |
| #= Supersocket _flush_fd |
| #~ needs_root linux |
| # |
| #import select |
| # |
| #from scapy.arch.linux import _flush_fd |
| #socket = conf.L2listen() |
| #select.select([socket],[],[],2) |
| #_flush_fd(socket.ins) |
| |
| = Interface aliases & sub-interfaces |
| ~ linux needs_root |
| |
| import os |
| exit_status = os.system("ip link add name scapy0 type dummy") |
| exit_status = os.system("ip addr add 192.0.2.1/24 dev scapy0") |
| exit_status = os.system("ip link set scapy0 up") |
| exit_status = os.system("ifconfig scapy0:0 inet 198.51.100.1/24 up") |
| if exit_status == 0: |
| exit_status = os.system("ip addr show scapy0") |
| print(get_if_list()) |
| conf.route.resync() |
| print(conf.route.routes) |
| assert conf.route.route("198.51.100.254") == ("scapy0", "198.51.100.1", "0.0.0.0") |
| route_alias = (3325256704, 4294967040, "0.0.0.0", "scapy0", "198.51.100.1", 0) |
| assert route_alias in conf.route.routes |
| exit_status = os.system("ip link add link scapy0 name scapy0.42 type vlan id 42") |
| exit_status = os.system("ip addr add 203.0.113.42/24 dev scapy0.42") |
| exit_status = os.system("ip link set scapy0.42 up") |
| exit_status = os.system("ip route add 192.0.2.43/32 via 203.0.113.41") |
| print(get_if_list()) |
| conf.route.resync() |
| print(conf.route.routes) |
| assert conf.route.route("192.0.2.43") == ("scapy0.42", "203.0.113.42", "203.0.113.41") |
| route_specific = (3221226027, 4294967295, "203.0.113.41", "scapy0.42", "0.0.0.0", 0) |
| assert route_specific in conf.route.routes |
| assert conf.route.route("203.0.113.42") == ('scapy0.42', '203.0.113.42', '0.0.0.0') |
| assert conf.route.route("203.0.113.43") == ('scapy0.42', '203.0.113.42', '0.0.0.0') |
| exit_status = os.system("ip link del name dev scapy0") |
| else: |
| assert True |
| |
| |
| = Test scoped interface addresses |
| ~ linux needs_root |
| |
| import os |
| exit_status = os.system("ip link add name scapy0 type dummy") |
| exit_status = os.system("ip link add name scapy1 type dummy") |
| exit_status |= os.system("ip addr add 192.0.2.1/24 dev scapy0") |
| exit_status |= os.system("ip addr add 192.0.3.1/24 dev scapy1") |
| exit_status |= os.system("ip link set scapy0 address 00:01:02:03:04:05 multicast on up") |
| exit_status |= os.system("ip link set scapy1 address 06:07:08:09:10:11 multicast on up") |
| assert exit_status == 0 |
| |
| conf.ifaces.reload() |
| conf.route.resync() |
| conf.route6.resync() |
| |
| conf.route6 |
| |
| try: |
| # IPv4 |
| a = Ether()/IP(dst="224.0.0.1%scapy0") |
| assert a[Ether].src == "00:01:02:03:04:05" |
| assert a[IP].src == "192.0.2.1" |
| b = Ether()/IP(dst="224.0.0.1%scapy1") |
| assert b[Ether].src == "06:07:08:09:10:11" |
| assert b[IP].src == "192.0.3.1" |
| c = Ether()/IP(dst="224.0.0.1/24%scapy1") |
| assert c[Ether].src == "06:07:08:09:10:11" |
| assert c[IP].src == "192.0.3.1" |
| # IPv6 |
| a = Ether()/IPv6(dst="ff02::fb%scapy0") |
| assert a[Ether].src == "00:01:02:03:04:05" |
| assert a[IPv6].src == "fe80::201:2ff:fe03:405" |
| b = Ether()/IPv6(dst="ff02::fb%scapy1") |
| assert b[Ether].src == "06:07:08:09:10:11" |
| assert b[IPv6].src == "fe80::407:8ff:fe09:1011" |
| c = Ether()/IPv6(dst="ff02::fb/30%scapy1") |
| assert c[Ether].src == "06:07:08:09:10:11" |
| assert c[IPv6].src == "fe80::407:8ff:fe09:1011" |
| finally: |
| exit_status = os.system("ip link del scapy0") |
| exit_status = os.system("ip link del scapy1") |
| conf.ifaces.reload() |
| conf.route.resync() |
| conf.route6.resync() |
| |
| = catch loopback device missing |
| ~ linux needs_root |
| |
| from unittest.mock import patch |
| |
| # can't remove the lo device (or its address without causing trouble) - use some pseudo dummy instead |
| |
| with patch('scapy.arch.linux.conf.loopback_name', 'scapy_lo_x'): |
| routes = read_routes() |
| |
| = catch loopback device no address assigned |
| ~ linux needs_root |
| |
| import os, socket |
| from unittest.mock import patch |
| |
| try: |
| exit_status = os.system("ip link add name scapy_lo type dummy") |
| assert exit_status == 0 |
| exit_status = os.system("ip link set dev scapy_lo up") |
| assert exit_status == 0 |
| |
| with patch('scapy.arch.linux.conf.loopback_name', 'scapy_lo'): |
| routes = read_routes() |
| |
| exit_status = os.system("ip addr add dev scapy_lo 10.10.0.1/24") |
| assert exit_status == 0 |
| |
| with patch('scapy.arch.linux.conf.loopback_name', 'scapy_lo'): |
| routes = read_routes() |
| |
| lo_routes = [ |
| (ltoa(dst_int), ltoa(msk_int), gw_str, if_name, if_addr, metric) |
| for dst_int, msk_int, gw_str, if_name, if_addr, metric in routes |
| if if_name == "scapy_lo" |
| ] |
| lo_routes.sort(key=lambda x: x[0]) |
| |
| expected_routes = [ |
| (168427520, 4294967040, '0.0.0.0', 'scapy_lo', '10.10.0.1', 0), |
| (168427521, 4294967295, '0.0.0.0', 'scapy_lo', '10.10.0.1', 0), |
| (168427775, 4294967295, '0.0.0.0', 'scapy_lo', '10.10.0.1', 0), |
| ] |
| print(lo_routes) |
| print(expected_routes) |
| finally: |
| exit_status = os.system("ip link del dev scapy_lo") |
| assert exit_status == 0 |
| |
| = IPv6 link-local address selection |
| |
| conf.ifaces._add_fake_iface("scapy0", 'e2:39:91:79:19:10') |
| |
| from unittest.mock import patch |
| conf.route6.routes = [('fe80::', 64, '::', 'scapy0', ['fe80::e039:91ff:fe79:1910'], 256)] |
| conf.route6.ipv6_ifaces = set(['scapy0']) |
| bck_conf_iface = conf.iface |
| conf.iface = "scapy0" |
| |
| p = Ether()/IPv6(dst="ff02::1")/ICMPv6NIQueryName(data="ff02::1") |
| print(p.sprintf("%Ether.src% > %Ether.dst%\n%IPv6.src% > %IPv6.dst%")) |
| ip6_ll_address = 'fe80::e039:91ff:fe79:1910' |
| print(p[IPv6].src, ip6_ll_address) |
| assert p[IPv6].src == ip6_ll_address |
| mac_address = 'e2:39:91:79:19:10' |
| print(p[Ether].src, mac_address) |
| assert p[Ether].src == mac_address |
| |
| conf.iface = bck_conf_iface |
| conf.route6.resync() |
| |
| = IPv6 - check OS routes |
| ~ linux ipv6 |
| |
| addrs = in6_getifaddr() |
| if addrs: |
| assert all(in6_isvalid(addr[0]) for addr in in6_getifaddr()), 'invalid ipv6 address' |
| ifaces6 = [addr[2] for addr in in6_getifaddr()] |
| assert all(iface in ifaces6 for iface in conf.route6.ipv6_ifaces), 'ipv6 interface has route but no real' |
| |
| |
| = veth interface error handling |
| ~ linux needs_root veth |
| |
| from scapy.arch.linux import VEthPair |
| |
| try: |
| veth = VEthPair('this_IF_name_is_to_long_and_will_cause_an_error', 'veth_scapy_1') |
| veth.setup() |
| assert False |
| except subprocess.CalledProcessError: |
| pass |
| except Exception: |
| assert False |
| |
| = veth interface usage - ctx manager |
| ~ linux needs_root veth |
| |
| from threading import Condition |
| |
| cond_started = Condition() |
| |
| def _sniffer_started(): |
| |
| global cond_started |
| cond_started.acquire() |
| cond_started.notify() |
| cond_started.release() |
| |
| cond_started.acquire() |
| |
| try: |
| with VEthPair('veth_scapy_0', 'veth_scapy_1') as veth: |
| if_list = get_if_list() |
| assert ('veth_scapy_0' in if_list) |
| assert ('veth_scapy_1' in if_list) |
| frm_count = 0 |
| def _sniffer(): |
| sniffed = sniff(iface='veth_scapy_1', |
| store=True, |
| count=2, |
| lfilter=lambda p: Ether in p and p[Ether].type == 0xbeef, |
| started_callback=_sniffer_started, |
| timeout=3) |
| global frm_count |
| frm_count = 2 |
| t_sniffer = Thread(target=_sniffer, name="linux.uts sniff veth_scapy_1") |
| t_sniffer.start() |
| cond_started.wait() |
| sendp(Ether(type=0xbeef)/Raw(b'0123456789'), |
| iface='veth_scapy_0', |
| count=2) |
| t_sniffer.join(1) |
| assert frm_count == 2 |
| |
| if_list = get_if_list() |
| assert ('veth_scapy_0' not in if_list) |
| assert ('veth_scapy_1' not in if_list) |
| except subprocess.CalledProcessError: |
| assert False |
| except Exception: |
| assert False |
| |
| = veth interface usage - manual interface handling |
| ~ linux needs_root veth |
| |
| from threading import Condition |
| |
| cond_started = Condition() |
| |
| def _sniffer_started(): |
| |
| global cond_started |
| cond_started.acquire() |
| cond_started.notify() |
| cond_started.release() |
| |
| cond_started.acquire() |
| |
| |
| veth = VEthPair('veth_scapy_0', 'veth_scapy_1') |
| try: |
| veth.setup() |
| veth.up() |
| except subprocess.CalledProcessError: |
| assert False |
| except Exception: |
| assert False |
| |
| conf.ifaces.reload() |
| if_list = get_if_list() |
| assert ('veth_scapy_0' in if_list) |
| assert ('veth_scapy_1' in if_list) |
| |
| frm_count = 0 |
| def _sniffer(): |
| sniffed = sniff(iface='veth_scapy_1', |
| store=True, |
| count=2, |
| lfilter=lambda p: Ether in p and p[Ether].type == 0xbeef, |
| started_callback=_sniffer_started, |
| timeout=3) |
| global frm_count |
| frm_count = 2 |
| |
| t_sniffer = Thread(target=_sniffer, name="linux.uts sniff veth_scapy_1 2") |
| t_sniffer.start() |
| cond_started.wait() |
| sendp(Ether(type=0xbeef)/Raw(b'0123456789'), |
| iface='veth_scapy_0', |
| count=2) |
| t_sniffer.join(1) |
| assert frm_count == 2 |
| |
| try: |
| veth.down() |
| veth.destroy() |
| |
| conf.ifaces.reload() |
| if_list = get_if_list() |
| assert ('veth_scapy_0' not in if_list) |
| assert ('veth_scapy_1' not in if_list) |
| except subprocess.CalledProcessError: |
| assert False |
| except Exception: |
| assert False |
| |
| |
| = Routing table, interface with no names |
| ~ linux |
| |
| from unittest.mock import patch |
| |
| @patch("scapy.arch.linux.ioctl") |
| def test_read_routes(mock_ioctl): |
| def raise_ioerror(*args, **kwargs): |
| if args[1] == 0x8912: |
| return args[2] |
| raise IOError |
| mock_ioctl.side_effect = raise_ioerror |
| read_routes() |
| |
| test_read_routes() |
| |
| |
| = L3PacketSocket sendto exception |
| ~ linux needs_root |
| |
| from scapy.arch.linux import L3PacketSocket |
| |
| import socket |
| from unittest import mock |
| |
| @mock.patch("scapy.arch.linux.socket.socket.sendto") |
| def test_L3PacketSocket_sendto_python3(mock_sendto): |
| mock_sendto.side_effect = OSError(22, 2807) |
| l3ps = L3PacketSocket() |
| l3ps.send(IP(dst="8.8.8.8")/ICMP()) |
| return True |
| |
| assert test_L3PacketSocket_sendto_python3() |
| |
| = Test _interface_selection |
| ~ netaccess linux needs_root |
| |
| import os |
| from scapy.sendrecv import _interface_selection |
| assert _interface_selection(IP(dst="8.8.8.8")/UDP()) == (conf.iface, False) |
| exit_status = os.system("ip link add name scapy0 type dummy") |
| exit_status = os.system("ip addr add 192.0.2.1/24 dev scapy0") |
| exit_status = os.system("ip addr add fc00::/24 dev scapy0") |
| exit_status = os.system("ip link set scapy0 up") |
| conf.ifaces.reload() |
| conf.route.resync() |
| conf.route6.resync() |
| assert _interface_selection(IP(dst="192.0.2.42")/UDP()) == ("scapy0", False) |
| assert _interface_selection(IPv6(dst="fc00::ae0d")/UDP()) == ("scapy0", True) |
| exit_status = os.system("ip link del name dev scapy0") |
| conf.ifaces.reload() |
| conf.route.resync() |
| conf.route6.resync() |
| |
| = Test 802.1Q sniffing |
| ~ linux needs_root veth |
| |
| from scapy.arch.linux import VEthPair |
| from threading import Thread, Condition |
| |
| def _send(): |
| sendp(Ether()/IP(dst="198.51.100.2")/ICMP(), iface='vlanleft0', count=2) |
| |
| |
| with VEthPair("left0", "right0") as veth: |
| exit_status = os.system("ip link add link right0 name vlanright0 type vlan id 42") |
| exit_status = os.system("ip link add link left0 name vlanleft0 type vlan id 42") |
| exit_status = os.system("ip link set vlanright0 up") |
| exit_status = os.system("ip link set vlanleft0 up") |
| exit_status = os.system("ip addr add 198.51.100.1/24 dev vlanleft0") |
| exit_status = os.system("ip addr add 198.51.100.2/24 dev vlanright0") |
| sniffer = AsyncSniffer( |
| iface="right0", |
| lfilter=lambda p: Dot1Q in p, |
| count=2, |
| timeout=5, |
| started_callback=_send, |
| ) |
| sniffer.start() |
| sniffer.join(1) |
| if sniffer.running: |
| sniffer.stop() |
| raise Scapy_Exception("Sniffer did not stop !") |
| else: |
| results = sniffer.results |
| |
| |
| assert len(results) == 2 |
| assert all(Dot1Q in x for x in results) |
| |
| |
| = Reload interfaces & routes |
| |
| conf.ifaces.reload() |
| conf.route.resync() |
| conf.route6.resync() |
