File: regression.uts

package info (click to toggle)
scapy 2.5.0%2Bdfsg-2
links: PTS, VCS
area: main
in suites: bookworm
size: 12,136 kB
sloc: python: 127,773; sh: 133; makefile: 11
file content (4787 lines) | stat: -rw-r--r-- 156,835 bytes
% Regression tests for Scapy

# More information at http://www.secdev.org/projects/UTscapy/

############
############
+ Information on Scapy

= Setup
def expect_exception(e, c):
    try:
        c()
        return False
    except e:
        return True

= Get conf
~ conf command
* Dump the current configuration
conf

IP().src
conf.loopback_name

= Test module version detection
~ conf

class FakeModule(object):
    __version__ = "v1.12"

class FakeModule2(object):
    __version__ = "5.143.3.12"

class FakeModule3(object):
    __version__ = "v2.4.2.dev42"

from scapy.config import _version_checker

assert _version_checker(FakeModule, (1,11,5))
assert not _version_checker(FakeModule, (1,13))

assert _version_checker(FakeModule2, (5, 1))
assert not _version_checker(FakeModule2, (5, 143, 4))

assert _version_checker(FakeModule3, (2, 4, 2))

= List layers
~ conf command
ls()

= List layers - advanced
~ conf command

with ContextManagerCaptureOutput() as cmco:
    ls("IP", case_sensitive=True)
    result_ls = cmco.get_output().split("\n")

assert all("IP" in x for x in result_ls if x.strip())
assert len(result_ls) >= 3

= List packet fields - ls
~ command

with ContextManagerCaptureOutput() as cmco:
    ls(ARP(hwsrc="aa:aa:aa:aa:aa:aa", psrc="1.1.1.1"))
    result_ls = cmco.get_output().split("\n")

result_ls
assert result_ls[5] == "hwsrc      : MultipleTypeField (SourceMACField, StrFixedLenField) = 'aa:aa:aa:aa:aa:aa' ('None')"
assert result_ls[6] == "psrc       : MultipleTypeField (SourceIPField, SourceIP6Field, StrFixedLenField) = '1.1.1.1'       ('None')"

= List commands
~ conf command
lsc()

= List contribs
~ command
def test_list_contrib():
    with ContextManagerCaptureOutput() as cmco:
        list_contrib()
        result_list_contrib = cmco.get_output()
    assert "http2               : HTTP/2 (RFC 7540, RFC 7541)              status=loads" in result_list_contrib
    assert len(result_list_contrib.split('\n')) > 40

test_list_contrib()

= Test automatic doc generation
~ command

dat = rfc(IP, ret=True).split("\n")
assert dat[0].replace(" ", "").strip() == "0123"
assert "0123456789" in dat[1].replace(" ", "")
for l in dat:
    # only upper case and +-
    assert re.match(r"[A-Z+-]*", l)

# Add a space before each + to avoid conflicts with UTscapy !
# They will be stripped below
result = """
 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |VERSION|  IHL  |      TOS      |              LEN              |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |               ID              |FLAGS|           FRAG          |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |      TTL      |     PROTO     |             CHKSUM            |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                              SRC                              |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                              DST                              |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |            OPTIONS            |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

                             Fig. IP
""".strip()
result = [x.strip() for x in result.split("\n")]
output = [x.strip() for x in rfc(IP, ret=True).strip().split("\n")]
assert result == output

result = """
  0                   1                   2                   3
  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |      CODE     |       ID      |              LEN              |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |      TYPE     |L|M|S|RES|VERSI|          MESSAGE LEN          |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                               |              DATA             |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

                           Fig. EAP_TTLS
""".strip()
result = [x.strip() for x in result.split("\n")]
output = [x.strip() for x in rfc(EAP_TTLS, ret=True).strip().split("\n")]
assert result == output


result = """
  0                   1                   2                   3
  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |VERSION|       TC      |                   FL                  |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |              PLEN             |       NH      |      HLIM     |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                              SRC                              |
 +                                                               +
 |                                                               |
 +                                                               +
 |                                                               |
 +                                                               +
 |                                                               |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 |                              DST                              |
 +                                                               +
 |                                                               |
 +                                                               +
 |                                                               |
 +                                                               +
 |                                                               |
 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

                             Fig. IPv6
""".strip()
result = [x.strip() for x in result.split("\n")]
output = [x.strip() for x in rfc(IPv6, ret=True).strip().split("\n")]
assert result == output

= Check that all contrib modules are well-configured
~ command
list_contrib(_debug=True)

= Configuration
~ conf
conf.debug_dissector = True

= Configuration conf.use_* LINUX
~ linux

try:
    conf.use_bpf = True
    assert False
except:
    True

assert not conf.use_bpf

= Configuration conf.use_* WINDOWS
~ windows

try:
    conf.use_bpf = True
    assert False
except:
    True

assert not conf.use_bpf

= Configuration conf.use_pcap
~ linux

if not conf.use_pcap:
    assert not conf.iface.provider.libpcap
    conf.use_pcap = True
    assert conf.iface.provider.libpcap
    for iface in conf.ifaces.values():
        assert iface.provider.libpcap or iface.is_valid() == False
    conf.use_pcap = False
    assert not conf.iface.provider.libpcap

= Test layer filtering
~ filter

pkt = NetflowHeader()/NetflowHeaderV5()/NetflowRecordV5()

conf.layers.filter([NetflowHeader, NetflowHeaderV5])
assert NetflowRecordV5 not in NetflowHeader(bytes(pkt))

conf.layers.unfilter()
assert NetflowRecordV5 in NetflowHeader(bytes(pkt))


###########
###########
= UTscapy route check
* Check that UTscapy has correctly replaced the routes. Many tests won't work otherwise

p = IP().src
p
assert p == "127.0.0.1"

############
############
+ Scapy functions tests

= Interface related functions

import mock

conf.iface

get_if_raw_hwaddr(conf.iface)

bytes_hex(get_if_raw_addr(conf.iface))

def get_dummy_interface():
    """Returns a dummy network interface"""
    IFACES._add_fake_iface("dummy0")
    return "dummy0"

get_if_raw_addr(get_dummy_interface())

get_if_list()

get_working_if()

get_if_raw_addr6(conf.iface)

if conf.use_bpf:
    addr = u"lladdr 29:0b:c2:ff:fe:53:21:e9\n"
    b = Bunch(returncode=0, communicate=lambda *args, **kargs: (addr, None))
    with mock.patch('scapy.arch.bpf.core.subprocess.Popen', return_value=b) as popen:
        try:
            scapy.arch.bpf.core.get_if_raw_hwaddr("fw0")
            assert False
        except Scapy_Exception:
            assert True

= More Interfaces related functions

# Test name resolution
old = conf.iface
conf.iface = conf.iface.name
assert conf.iface == old

assert isinstance(conf.iface, NetworkInterface)
assert conf.iface.is_valid()

import mock
@mock.patch("scapy.interfaces.conf.route.routes", [])
@mock.patch("scapy.interfaces.conf.ifaces.values")
def _test_get_working_if(rou):
    rou.side_effect = lambda: []
    assert get_working_if() == conf.loopback_name

assert conf.iface + "a"  # left +
assert "hey! are you, ready to go ? %s" % conf.iface  # format
assert "cuz you know the way to go" + conf.iface  # right +

_test_get_working_if()

= Test conf.ifaces

conf.iface
conf.ifaces

assert conf.iface in conf.ifaces.values()
assert conf.ifaces.dev_from_index(conf.iface.index) == conf.iface
assert conf.ifaces.dev_from_networkname(conf.iface.network_name) == conf.iface

conf.ifaces.data = {'a': NetworkInterface(InterfaceProvider(), {"name": 'a', "network_name": 'a', "description": 'a', "ips": ["127.0.0.1", "::1", "::2", "127.0.0.2"], "mac": 'aa:aa:aa:aa:aa:aa'})}

with ContextManagerCaptureOutput() as cmco:
    conf.ifaces.show()
    output = cmco.get_output()

data = """
Source   Index  Name  MAC                IPv4       IPv6
Unknown  0      a     aa:aa:aa:aa:aa:aa  127.0.0.1  ::1
                                         127.0.0.2  ::2
""".strip()

output = [x.strip() for x in output.strip().split("\n")]
data = [x.strip() for x in data.strip().split("\n")]

assert output == data

conf.ifaces.reload()

= Test read_routes6() - default output

routes6 = read_routes6()
if WINDOWS:
    from scapy.arch.windows import _route_add_loopback
    _route_add_loopback(routes6, True)

routes6

# Expected results:
# - one route if there is only the loopback interface
# - one route if IPv6 is supported but disabled on network interfaces
# - three routes if there is a network interface
# - on OpenBSD, only two routes on lo0 are expected

if routes6:
    iflist = get_if_list()
    if WINDOWS:
        from scapy.arch.windows import _route_add_loopback
        _route_add_loopback(ipv6=True, iflist=iflist)
    if OPENBSD:
        len(routes6) >= 2
    elif iflist == [conf.loopback_name]:
        len(routes6) == 1
    elif len(iflist) >= 2:
        len(routes6) >= 1
    else:
        False
else:
    # IPv6 seems disabled. Force a route to ::1
    conf.route6.routes.append(("::1", 128, "::", conf.loopback_name, ["::1"], 1))
    conf.route6.ipv6_ifaces = set([conf.loopback_name])
    True

= Build HBHOptUnknown for IPv6ExtHdrHopByHop with disabled autopad
~ ipv6 hbh opt
* Build the HBHOptUnknown of IPv6ExtHdrHopByHop with autopad=0
v6Opt = HBHOptUnknown(otype=3, optlen=7, optdata="Beijing")
pkt = Ether()/IPv6()/IPv6ExtHdrHopByHop(autopad=0, options=[v6Opt, ])
pkt.build()

= Build HBHOptUnknown for IPv6ExtHdrDestOpt with disabled autopad
~ ipv6 hbh opt
* Build the HBHOptUnknown of IPv6ExtHdrDestOpt with autopad=0
v6Opt = HBHOptUnknown(otype=3, optlen=6, optdata="Haikou")
pkt = Ether()/IPv6()/IPv6ExtHdrDestOpt(autopad=0, options=[v6Opt, ])
pkt.build()


= Test read_routes6() - check mandatory routes

conf.route6

# Doesn't pass on Travis Bionic XXX
if len(routes6) > 2 and not WINDOWS:
    # Identify routes to fe80::/64
    assert sum(1 for r in routes6 if r[0] == "::1" and r[4] == ["::1"]) >= 1
    if not OPENBSD and len(iflist) >= 2:
        assert sum(1 for r in routes6 if r[0] == "fe80::" and r[1] == 64) >= 1
        try:
            # Identify a route to a node IPv6 link-local address
            assert sum(1 for r in routes6 if in6_islladdr(r[0]) and r[1] == 128) >= 1
        except:
            # IPv6 is not available, but we still check the loopback
            assert conf.route6.route("::/0") == (conf.loopback_name, "::", "::")
            assert sum(1 for r in routes6 if r[1] == 128 and r[4] == ["::1"]) >= 1
else:
    True

= Test ifchange()
conf.route6.ifchange(conf.loopback_name, "::1/128")
if WINDOWS:
    conf.netcache.in6_neighbor["::1"] = "ff:ff:ff:ff:ff:ff"  # Restore fake cache

True

= Packet.route()
assert (Ether() / ARP()).route()[0] is not None
assert (Ether() / ARP()).payload.route()[0] is not None
assert (ARP(ptype=0, pdst="hello. this isn't a valid IP")).route()[0] is None


= plain_str test

data = b"\xffsweet\xef celestia\xab"
if not six.PY2:
    # Only Python 3 has to deal with Str/Bytes conversion,
    # as we don't use Python 2's unicode
    if sys.version_info[0:2] <= (3, 4):
        # Python3.4 can only ignore unknown special characters
        assert plain_str(data) == "sweet celestia"
    else:
        # Python >3.4 can replace them with a backslash representation
        assert plain_str(data) == "\\xffsweet\\xef celestia\\xab"
else:
    assert plain_str(data) == "\xffsweet\xef celestia\xab"

############
############
+ compat.py

= test bytes_hex/hex_bytes

monty_data = b"Stop! Who approaches the Bridge of Death must answer me these questions three, 'ere the other side he see."
hex_data = bytes_hex(monty_data)
assert hex_data == b'53746f70212057686f20617070726f61636865732074686520427269646765206f66204465617468206d75737420616e73776572206d65207468657365207175657374696f6e732074687265652c202765726520746865206f746865722073696465206865207365652e'
assert hex_bytes(hex_data) == monty_data

= test gzip_decompress/gzip_compress

from scapy.compat import gzip_compress, gzip_decompress

gziped_data = b"\x1f\x8b\x08\x00N\xf5\xd7\\\x02\xff\x1d\x8b9\x0e\x800\x0c\x04\xbf\xb2T4\x88G ~@A\x1d\x91\x85\xa4H\x1cl#\xbe\xcf\xd1\xac4\x9a\xd9\xc5\xa5uX\x93 \xb4\xa6\x12\xb6D\x83'b\xd2\x1c\x0fBv\xcc\x0c\x9eP.s\x84j7\x15\x85_b\xc4y\xd1<K\xfd.J\x0e\xe8\xa9\xbf\x83\xbc\xa3\xb0\x1c\x89\x97\x8c\x1c\x1fc\xbeS\\j\x00\x00\x00"
assert gzip_decompress(gziped_data) == monty_data

data = b"sweet celestia"
assert gzip_decompress(gzip_compress(data)) == data

= orb/chb

assert orb(b"\x01"[0]) == 1
assert chb(1) == b"\x01"

############
############
+ Main.py tests

= Pickle and unpickle a packet

import scapy.libs.six as six

a = IP(dst="192.168.0.1")/UDP()

b = six.moves.cPickle.dumps(a)
c = six.moves.cPickle.loads(b)

assert c[IP].dst == "192.168.0.1"
assert raw(c) == raw(a)

= Usage test

from scapy.main import _usage
try:
    _usage()
    assert False
except SystemExit:
    assert True

= Session test

# This is automatic when using the console
def get_var(var):
    return six.moves.builtins.__dict__["scapy_session"][var]

def set_var(var, value):
    six.moves.builtins.__dict__["scapy_session"][var] = value

def del_var(var):
    del six.moves.builtins.__dict__["scapy_session"][var]

init_session(None, {"init_value": 123})
set_var("test_value", "8.8.8.8") # test_value = "8.8.8.8"
save_session()
del_var("test_value")
load_session()
update_session()
assert get_var("test_value") == "8.8.8.8" #test_value == "8.8.8.8"
assert get_var("init_value") == 123
 
= Session test with fname

session_name = tempfile.mktemp()
init_session(session_name)
set_var("test_value", IP(dst="192.168.0.1")) # test_value = IP(dst="192.168.0.1")
save_session(fname="%s.dat" % session_name)
del_var("test_value")

set_var("z", True) #z = True
load_session(fname="%s.dat" % session_name)
try:
    get_var("z")
    assert False
except:
    pass

set_var("z", False) #z = False
update_session(fname="%s.dat" % session_name)
assert get_var("test_value").dst == "192.168.0.1" #test_value.dst == "192.168.0.1"
assert not get_var("z")

= Clear session files

os.remove("%s.dat" % session_name)

= Test temporary file creation
~ ci_only

scapy_delete_temp_files()

tmpfile = get_temp_file(autoext=".ut")
tmpfile
if WINDOWS:
    assert "scapy" in tmpfile and tmpfile.lower().startswith('c:\\users\\appveyor\\appdata\\local\\temp')
else:
    import platform
    BYPASS_TMP = platform.python_implementation().lower() == "pypy" or DARWIN
    assert "scapy" in tmpfile and (BYPASS_TMP == True or "/tmp/" in tmpfile)

assert conf.temp_files[0].endswith(".ut")
scapy_delete_temp_files()
assert len(conf.temp_files) == 0

= Emulate interact()
import mock, sys
from scapy.main import interact
# Detect IPython
try:
    import IPython
except:
    code_interact_import = "scapy.main.code.interact"
else:
    code_interact_import = "IPython.start_ipython"

@mock.patch(code_interact_import)
def interact_emulator(code_int, extra_args=[]):
    try:
        code_int.side_effect = lambda *args, **kwargs: lambda *args, **kwargs: None
        interact(argv=["-s scapy1"] + extra_args, mybanner="What a test")
    finally:
        sys.ps1 = ">>> "

interact_emulator()  # Default

try:
    interact_emulator(extra_args=["-?"])  # Failing
    assert False
except:
    pass

interact_emulator(extra_args=["-d"])  # Extended

= Test explore() with GUI mode
~ command

import mock

def test_explore_gui(is_layer, layer):
    prompt_toolkit_mocked_module = Bunch(
                                       shortcuts=Bunch(
                                           dialogs=Bunch(
                                               radiolist_dialog=(lambda *args, **kargs: layer),
                                               button_dialog=(lambda *args, **kargs: "layers" if is_layer else "contribs")
                                           )
                                       ),
                                       formatted_text=Bunch(HTML=lambda x: x),
                                       __version__="2.0.0"
                                   )
    # a mock.patch isn't enough to mock a module. Let's roll sys.modules
    modules_patched = {
        "prompt_toolkit": prompt_toolkit_mocked_module,
        "prompt_toolkit.shortcuts": prompt_toolkit_mocked_module.shortcuts,
        "prompt_toolkit.shortcuts.dialogs": prompt_toolkit_mocked_module.shortcuts.dialogs,
        "prompt_toolkit.formatted_text": prompt_toolkit_mocked_module.formatted_text,
    }
    with mock.patch.dict("sys.modules", modules_patched):
        with ContextManagerCaptureOutput() as cmco:
            explore()
            result_explore = cmco.get_output()
        return result_explore

conf.interactive = True
explore_dns = test_explore_gui(True, "scapy.layers.dns")
assert "DNS" in explore_dns
assert "DNS Question Record" in explore_dns
assert "DNSRRNSEC3" in explore_dns
assert "DNS TSIG Resource Record" in explore_dns

explore_avs = test_explore_gui(False, "avs")
assert "AVSWLANHeader" in explore_avs
assert "AVS WLAN Monitor Header" in explore_avs

= Test explore() with non-GUI mode
~ command

def test_explore_non_gui(layer):
    with ContextManagerCaptureOutput() as cmco:
        explore(layer)
        result_explore = cmco.get_output()
    return result_explore

explore_dns = test_explore_non_gui("scapy.layers.dns")
assert "DNS" in explore_dns
assert "DNS Question Record" in explore_dns
assert "DNSRRNSEC3" in explore_dns
assert "DNS TSIG Resource Record" in explore_dns

explore_avs = test_explore_non_gui("avs")
assert "AVSWLANHeader" in explore_avs
assert "AVS WLAN Monitor Header" in explore_avs

assert test_explore_non_gui("scapy.layers.dns") == test_explore_non_gui("dns")
assert test_explore_non_gui("scapy.contrib.avs") == test_explore_non_gui("avs")

try:
    explore("unknown_module")
    assert False  # The previous should have raised an exception
except Scapy_Exception:
    pass

= Test load_contrib overwrite
load_contrib("gtp")
assert GTPHeader.__module__ == "scapy.contrib.gtp"

load_contrib("gtp_v2")
assert GTPHeader.__module__ == "scapy.contrib.gtp_v2"

load_contrib("gtp")
assert GTPHeader.__module__ == "scapy.contrib.gtp"

= Test load_contrib failure
try:
    load_contrib("doesnotexist")
    assert False
except:
    pass

= Test sane function
sane("A\x00\xFFB") == "A..B"

= Test lhex function
assert lhex(42) == "0x2a"
assert lhex((28,7)) == "(0x1c, 0x7)"
assert lhex([28,7]) == "[0x1c, 0x7]"

= Test restart function
import mock
conf.interactive = True

try:
    from scapy.utils import restart
    import os
    @mock.patch("os.execv")
    @mock.patch("subprocess.call")
    @mock.patch("os._exit")
    def _test(e, m, m2):
        def check(x, y=[]):
            z = [x] + y if not isinstance(x, list) else x + y
            assert os.path.isfile(z[0])
            assert os.path.isfile(z[1])
            return 0
        m2.side_effect = check
        m.side_effect = check
        e.side_effect = lambda x: None
        restart()
    _test()
finally:
    conf.interactive = False

= Test linehexdump function
conf_color_theme = conf.color_theme
conf.color_theme = BlackAndWhite()
assert linehexdump(Ether(src="00:01:02:03:04:05"), dump=True) == 'FF FF FF FF FF FF 00 01 02 03 04 05 90 00  ..............'
conf.color_theme = conf_color_theme

= Test chexdump function
chexdump(Ether(src="00:01:02:02:04:05"), dump=True) == "0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, 0x01, 0x02, 0x02, 0x04, 0x05, 0x90, 0x00"

= Test repr_hex function
repr_hex("scapy") == "7363617079"

= Test hexstr function
hexstr(b"A\x00\xFFB") == "41 00 FF 42  A..B"

= Test fletcher16 functions
assert fletcher16_checksum(b"\x28\x07") == 22319
assert fletcher16_checkbytes(b"\x28\x07", 1) == b"\xaf("

= Test hexdiff function
~ not_pypy
def test_hexdiff(a, b, autojunk=False):
    conf_color_theme = conf.color_theme
    conf.color_theme = BlackAndWhite()
    with ContextManagerCaptureOutput() as cmco:
        hexdiff(a, b, autojunk=autojunk)
        result_hexdiff = cmco.get_output()
    conf.interactive = True
    conf.color_theme = conf_color_theme
    return result_hexdiff

# Basic string test

result_hexdiff = test_hexdiff("abcde", "abCde")
expected  = "0000        61 62 63 64 65                                     abcde\n"
expected += "     0000   61 62 43 64 65                                     abCde\n"
assert result_hexdiff == expected

# More advanced string test

result_hexdiff = test_hexdiff("add_common_", "_common_removed")
expected  = "0000        61 64 64 5F 63 6F 6D 6D  6F 6E 5F                  add_common_     \n"
expected += "     -003            5F 63 6F 6D 6D  6F 6E 5F 72 65 6D 6F 76      _common_remov\n"
expected += "     000d   65 64                                              ed\n"
assert result_hexdiff == expected

# Compare packets

result_hexdiff = test_hexdiff(IP(dst="127.0.0.1", src="127.0.0.1"), IP(dst="127.0.0.2", src="127.0.0.1"))
expected  = "0000        45 00 00 14 00 01 00 00  40 00 7C E7 7F 00 00 01   E.......@.|.....\n"
expected += "     0000   45 00 00 14 00 01 00 00  40 00 7C E6 7F 00 00 01   E.......@.|.....\n"
expected += "0010        7F 00 00 01                                        ....\n"
expected += "     0010   7F 00 00 02                                        ....\n"
assert result_hexdiff == expected

# Compare using autojunk

a = "A" * 1000 + "findme" + "B" * 1000
b = "A" * 1000 + "B" * 1000
ret1 = test_hexdiff(a, b)
ret2 = test_hexdiff(a, b, autojunk=True)

expected_ret1 = """
03d0 03d0   41 41 41 41 41 41 41 41  41 41 41 41 41 41 41 41   AAAAAAAAAAAAAAAA
03e0        41 41 41 41 41 41 41 41  66 69 6E 64 6D 65 42 42   AAAAAAAAfindmeBB
     03e0   41 41 41 41 41 41 41 41                    42 42   AAAAAAAA      BB
03ea 03ea   42 42 42 42 42 42 42 42  42 42 42 42 42 42 42 42   BBBBBBBBBBBBBBBB
"""
expected_ret2 = """
03d0 03d0   41 41 41 41 41 41 41 41  41 41 41 41 41 41 41 41   AAAAAAAAAAAAAAAA
03e0        41 41 41 41 41 41 41 41  66 69 6E 64 6D 65 42 42   AAAAAAAAfindmeBB
     03e0   41 41 41 41 41 41 41 41  42 42 42 42 42 42 42 42   AAAAAAAABBBBBBBB
03f0 03f0   42 42 42 42 42 42 42 42  42 42 42 42 42 42 42 42   BBBBBBBBBBBBBBBB
"""

assert ret1 != ret2
assert expected_ret1 in ret1
assert expected_ret2 in ret2

# Test corner cases that should not crash

hexdiff(b"abc", IP() / TCP())
hexdiff(IP() / TCP(), b"abc")

= Test mysummary functions - Ether

p = Ether(dst="ff:ff:ff:ff:ff:ff", src="ff:ff:ff:ff:ff:ff", type=0x9000)
p
assert p.mysummary() in ['ff:ff:ff:ff:ff:ff > ff:ff:ff:ff:ff:ff (%s)' % loop
                         for loop in ['0x9000', 'LOOP']]

= Test zerofree_randstring function
random.seed(0x2807)
zerofree_randstring(4) in [b"\xd2\x12\xe4\x5b", b'\xd3\x8b\x13\x12']

= Test strand function
assert strand("AC", "BC") == b'@C'

= Test export_object and import_object functions
import mock
def test_export_import_object():
    with ContextManagerCaptureOutput() as cmco:
        export_object(2807)
        result_export_object = cmco.get_output(eval_bytes=True)
    assert result_export_object.startswith("eNprYPL9zqUHAAdrAf8=")
    assert import_object(result_export_object) == 2807

test_export_import_object()

= Test tex_escape function
tex_escape("$#_") == "\\$\\#\\_"

= Test colgen function
f = colgen(range(3))
assert len([next(f) for i in range(2)]) == 2

= Test incremental_label function
f = incremental_label()
assert [next(f) for i in range(2)] == ["tag00000", "tag00001"]

= Test corrupt_* functions
import random
random.seed(0x2807)
assert corrupt_bytes("ABCDE") in [b"ABCDW", b"ABCDX"]
assert sane(corrupt_bytes("ABCDE", n=3)) in ["A.8D4", ".2.DE"]

assert corrupt_bits("ABCDE") in [b"EBCDE", b"ABCDG"]
assert sane(corrupt_bits("ABCDE", n=3)) in ["AF.EE", "QB.TE"]

= Test save_object and load_object functions
import tempfile
fd, fname = tempfile.mkstemp()
save_object(fname, 2807)
assert load_object(fname) == 2807

= Test whois function
~ netaccess

if not WINDOWS:
    result = whois("193.0.6.139")
    assert b"inetnum" in result and b"Amsterdam" in result

= Test manuf DB methods
~ manufdb
assert conf.manufdb._resolve_MAC("00:00:0F:01:02:03") == "Next:01:02:03"
assert conf.manufdb._get_short_manuf("00:00:0F:01:02:03") == "Next"
assert in6_addrtovendor("fe80::0200:0fff:fe01:0203").lower().startswith("next")

assert conf.manufdb.lookup("00:00:0F:01:02:03") == ('Next', 'Next, Inc.')
assert "00:00:0F" in conf.manufdb.reverse_lookup("Next")

= Test multiple wireshark's manuf formats
~ manufdb

new_format = """
# comment
00:00:00    JokyIsland    Joky Insland Corp SA
00:01:12    SecdevCorp    Secdev Corporation SA LLC
EE:05:01    Scapy         Scapy CO LTD & CIE
FF:00:11    NoName
"""
old_format = """
# comment
00:00:00    JokyIsland  #  Joky Insland Corp SA
00:01:12    SecdevCorp  #  Secdev Corporation SA LLC
EE:05:01    Scapy       #  Scapy CO LTD & CIE
FF:00:11    NoName
"""

manuf1 = get_temp_file()
manuf2 = get_temp_file()

with open(manuf1, "w") as w:
    w.write(old_format)

with open(manuf2, "w") as w:
    w.write(new_format)

a = load_manuf(manuf1)
b = load_manuf(manuf2)

a.lookup("00:00:00") == ('JokyIsland', 'Joky Insland Corp SA'),
a.lookup("FF:00:11:00:00:00") == ('NoName', 'NoName')
a.reverse_lookup("Scapy") == {'EE:05:01': ('Scapy', 'Scapy CO LTD & CIE')}
a.reverse_lookup("Secdevcorp") == {'00:01:12': ('SecdevCorp', 'Secdev Corporation SA LLC')}


b.lookup("00:00:00") == ('JokyIsland', 'Joky Insland Corp SA'),
b.lookup("FF:00:11:00:00:00") == ('NoName', 'NoName')
b.reverse_lookup("Scapy") == {'EE:05:01': ('Scapy', 'Scapy CO LTD & CIE')}
b.reverse_lookup("Secdevcorp") == {'00:01:12': ('SecdevCorp', 'Secdev Corporation SA LLC')}

scapy_delete_temp_files()

= Test load_services

data_services = """
itu-bicc-stc	3097/sctp
cvsup		5999/udp			# CVSup
x11		6000-6063/tcp			# X Window System
x11		6000-6063/udp			# X Window System
ndl-ahp-svc	6064/tcp			# NDL-AHP-SVC
"""

services = get_temp_file()
with open(services, "w") as w:
    w.write(data_services)

tcp, udp, sctp = load_services(services)
assert tcp[6002] == "x11"
assert tcp.ndl_ahp_svc == 6064
assert tcp.x11 in range(6000, 6093)
assert udp[6002] == "x11"
assert udp.x11 in range(6000, 6093)
assert udp.cvsup == 5999
assert sctp[3097] == "itu_bicc_stc"
assert sctp.itu_bicc_stc == 3097

scapy_delete_temp_files()

= Test utility functions - network related
~ netaccess

atol("www.secdev.org") == 3642339845

= Test autorun functions
~ autorun

ret = autorun_get_text_interactive_session("IP().src")
ret
assert ret == (">>> IP().src\n'127.0.0.1'\n", '127.0.0.1')

ret = autorun_get_html_interactive_session("IP().src")
ret
assert ret == ("<span class=prompt>&gt;&gt;&gt; </span>IP().src\n'127.0.0.1'\n", '127.0.0.1')

ret = autorun_get_latex_interactive_session("IP().src")
ret
assert ret == ("\\textcolor{blue}{{\\tt\\char62}{\\tt\\char62}{\\tt\\char62} }IP().src\n'127.0.0.1'\n", '127.0.0.1')

ret = autorun_get_text_interactive_session("scapy_undefined")
assert "NameError" in ret[0]

= Test autorun with logging

cmds = """log_runtime.info(hex_bytes("446166742050756e6b"))\n"""
ret = autorun_get_text_interactive_session(cmds)
assert "Daft Punk" in ret[0]

= Test utility TEX functions

assert tex_escape("{scapy}\\^$~#_&%|><") == "{\\tt\\char123}scapy{\\tt\\char125}{\\tt\\char92}\\^{}\\${\\tt\\char126}\\#\\_\\&\\%{\\tt\\char124}{\\tt\\char62}{\\tt\\char60}"

a = colgen(1, 2, 3)
assert next(a) == (1, 2, 2)
assert next(a) == (1, 3, 3)
assert next(a) == (2, 2, 1)
assert next(a) == (2, 3, 2)
assert next(a) == (2, 1, 3)
assert next(a) == (3, 3, 1)
assert next(a) == (3, 1, 2)
assert next(a) == (3, 2, 3)

= Test config file functions

saved_conf_verb = conf.verb
fd, fname = tempfile.mkstemp()
os.write(fd, b"conf.verb = 42\n")
os.close(fd)
from scapy.main import _read_config_file
_read_config_file(fname, globals(), locals())
assert conf.verb == 42
conf.verb = saved_conf_verb

= Test config file functions failures

from scapy.main import _probe_config_file
assert _probe_config_file("filethatdoesnotexistnorwillever.tsppajfsrdrr") is None

= Test CacheInstance repr

conf.netcache

= Test pyx detection functions

from mock import patch

def _r(*args, **kwargs):
    raise OSError

with patch("scapy.libs.test_pyx.subprocess.check_call", _r):
    from scapy.libs.test_pyx import _test_pyx
    assert _test_pyx() == False

= Test matplotlib detection functions

from mock import MagicMock, patch

bck_scapy_libs_matplot = sys.modules.get("scapy.libs.matplot", None)
if bck_scapy_libs_matplot:
    del sys.modules["scapy.libs.matplot"]

mock_matplotlib = MagicMock()
mock_matplotlib.get_backend.return_value = "inline"
mock_matplotlib.pyplot = MagicMock()
mock_matplotlib.pyplot.plt = None
with patch.dict("sys.modules", **{ "matplotlib": mock_matplotlib, "matplotlib.lines": mock_matplotlib}):
    from scapy.libs.matplot import MATPLOTLIB, MATPLOTLIB_INLINED, MATPLOTLIB_DEFAULT_PLOT_KARGS, Line2D
    assert MATPLOTLIB == 1
    assert MATPLOTLIB_INLINED == 1
    assert "marker" in MATPLOTLIB_DEFAULT_PLOT_KARGS

mock_matplotlib.get_backend.return_value = "ko"
with patch.dict("sys.modules", **{ "matplotlib": mock_matplotlib, "matplotlib.lines": mock_matplotlib}):
    from scapy.libs.matplot import MATPLOTLIB, MATPLOTLIB_INLINED, MATPLOTLIB_DEFAULT_PLOT_KARGS
    assert MATPLOTLIB == 1
    assert MATPLOTLIB_INLINED == 0
    assert "marker" in MATPLOTLIB_DEFAULT_PLOT_KARGS

if bck_scapy_libs_matplot:
    sys.modules["scapy.libs.matplot"] = bck_scapy_libs_matplot


############
############
+ Basic tests

* Those test are here mainly to check nothing has been broken
* and to catch Exceptions

= Packet class methods
p = IP()/ICMP()
ret = p.do_build_ps()                                                                                                                             
assert ret[0] == b"@\x00\x00\x00\x00\x01\x00\x00@\x01\x00\x00\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\x00\x00\x00\x00\x00\x00"
assert len(ret[1]) == 2

assert p[ICMP].firstlayer() == p

assert p.command() == "IP()/ICMP()"

p.decode_payload_as(UDP)
assert p.sport == 2048 and p.dport == 63487

= hide_defaults
conf_color_theme = conf.color_theme
conf.color_theme = BlackAndWhite()
p = IP(ttl=64)/ICMP()
assert repr(p) in ["<IP  frag=0 ttl=64 proto=icmp |<ICMP  |>>", "<IP  frag=0 ttl=64 proto=1 |<ICMP  |>>"]
p.hide_defaults()
assert repr(p) in ["<IP  frag=0 proto=icmp |<ICMP  |>>", "<IP  frag=0 proto=1 |<ICMP  |>>"]
conf.color_theme = conf_color_theme

= split_layers
p = IP()/ICMP()
s = raw(p)
split_layers(IP, ICMP, proto=1)
assert Raw in IP(s)
bind_layers(IP, ICMP, frag=0, proto=1)

= fuzz

r = fuzz(IP(tos=2)/ICMP())
assert r.tos == 2
z = r.ttl
assert r.ttl != z
assert r.ttl != z


= fuzz a Packet with MultipleTypeField

fuzz(ARP(pdst="127.0.0.1"))
fuzz(IP()/ARP(pdst='10.0.0.254'))

= fuzz on packets with advanced RandNum

x = IP(dst="8.8.8.8")/fuzz(UDP()/NTP(version=4))
x.show2()
x = IP(raw(x))
assert NTP in x

= fuzz on packets with FlagsField
assert isinstance(fuzz(TCP()).flags, VolatileValue)

= Building some packets
~ basic IP TCP UDP NTP LLC SNAP Dot11
IP()/TCP()
Ether()/IP()/UDP()/NTP()
Dot11()/LLC()/SNAP()/IP()/TCP()/"XXX"
IP(ttl=25)/TCP(sport=12, dport=42)
IP().summary()

= Manipulating some packets
~ basic IP TCP
a=IP(ttl=4)/TCP()
a.ttl
a.ttl=10
del a.ttl
a.ttl
TCP in a
a[TCP]
a[TCP].dport=[80,443]
a
assert a.copy().time == a.time
a=3

= Bind string array as payload
~ basic
assert bytes(Raw("sca")/"py") == b"scapy"
assert bytes(Raw("sca")/b"py") == b"scapy"
assert bytes(Raw("sca")/bytearray(b"py")) == b"scapy"
assert bytes("sca"/Raw("py")) == b"scapy"
assert bytes(b"sca"/Raw("py")) == b"scapy"
assert bytes(bytearray(b"sca")/Raw("py")) == b"scapy"
a=Raw("sca")
a.add_payload("py")
assert bytes(a) == b"scapy"
a=Raw("sca")
a.add_payload(b"py")
assert bytes(a) == b"scapy"
a=Raw("sca")
a.add_payload(bytearray(b"py"))
assert bytes(a) == b"scapy"

= Checking overloads
~ basic IP TCP Ether
a=Ether()/IP()/TCP()
r = a.proto
r
r == 6


= sprintf() function
~ basic sprintf Ether IP UDP NTP
a=Ether()/IP()/IP(ttl=4)/UDP()/NTP()
r = a.sprintf("%type% %IP.ttl% %#05xr,UDP.sport% %IP:2.ttl%")
r
r in ['0x800 64 0x07b 4', 'IPv4 64 0x07b 4']


= sprintf() function 
~ basic sprintf IP TCP SNAP LLC Dot11
* This test is on the conditional substring feature of <tt>sprintf()</tt>
a=Dot11()/LLC()/SNAP()/IP()/TCP()
r = a.sprintf("{IP:{TCP:flags=%TCP.flags%}{UDP:port=%UDP.ports%} %IP.src%}")
r
r == 'flags=S 127.0.0.1'


= haslayer function
~ basic haslayer IP TCP ICMP ISAKMP
x=IP(id=1)/ISAKMP_payload_SA(prop=ISAKMP_payload_SA(prop=IP()/ICMP()))/TCP()
r = (TCP in x, ICMP in x, IP in x, UDP in x)
r
r == (True,True,True,False)

= getlayer function
~ basic getlayer IP ISAKMP UDP
x=IP(id=1)/ISAKMP_payload_SA(prop=IP(id=2)/UDP(dport=1))/IP(id=3)/UDP(dport=2)
x[IP]
x[IP:2]
x[IP:3]
x.getlayer(IP,3)
x.getlayer(IP,4)
x[UDP]
x[UDP:1]
x[UDP:2]
assert(x[IP].id == 1 and x[IP:2].id == 2 and x[IP:3].id == 3 and 
       x.getlayer(IP).id == 1 and x.getlayer(IP,3).id == 3 and
       x.getlayer(IP,4) == None and
       x[UDP].dport == 1 and x[UDP:2].dport == 2)
try:
    x[IP:4]
except IndexError:
    True
else:
    False

= getlayer / haslayer with name
~ basic getlayer IP ICMP IPerror TCPerror
x = IP() / ICMP() / IPerror()
assert x.getlayer(ICMP) is not None
assert x.getlayer(IPerror) is not None
assert x.getlayer("IP in ICMP") is not None
assert x.getlayer(TCPerror) is None
assert x.getlayer("TCP in ICMP") is None
assert x.haslayer(ICMP)
assert x.haslayer(IPerror)
assert x.haslayer("IP in ICMP")
assert not x.haslayer(TCPerror)
assert not x.haslayer("TCP in ICMP")

= getlayer with a filter
~ getlayer IP
pkt = IP() / IP(ttl=3) / IP()
assert pkt[IP::{"ttl":3}].ttl == 3
assert pkt.getlayer(IP, ttl=3).ttl == 3
assert IPv6ExtHdrHopByHop(options=[HBHOptUnknown()]).getlayer(HBHOptUnknown, otype=42) is None

= specific haslayer and getlayer implementations for EAP
~ haslayer getlayer EAP
pkt = Ether() / EAPOL() / EAP_MD5()
assert EAP in pkt
assert pkt.haslayer(EAP)
assert isinstance(pkt[EAP], EAP_MD5)
assert isinstance(pkt.getlayer(EAP), EAP_MD5)

= specific haslayer and getlayer implementations for RadiusAttribute
~ haslayer getlayer RadiusAttribute
pkt = RadiusAttr_EAP_Message()
assert RadiusAttribute in pkt
assert pkt.haslayer(RadiusAttribute)
assert isinstance(pkt[RadiusAttribute], RadiusAttr_EAP_Message)
assert isinstance(pkt.getlayer(RadiusAttribute), RadiusAttr_EAP_Message)


= equality
~ basic
w=Ether()/IP()/UDP(dport=53)
x=Ether()/IP(version=4)/UDP()
y=Ether()/IP()/UDP(dport=4)
z=Ether()/IP()/UDP()/NTP()
t=Ether()/IP()/TCP()
assert x != y and x != z and x != t and y != z and y != t and z != t and w == x

= answers
~ basic
a1, a2 = "1.2.3.4", "5.6.7.8"
p1 = IP(src=a1, dst=a2)/ICMP(type=8)
p2 = IP(src=a2, dst=a1)/ICMP(type=0)
assert p1.hashret() == p2.hashret()
assert not p1.answers(p2)
assert p2.answers(p1)
assert p1 > p2
assert p2 < p1
assert p1 == p1
conf_back = conf.checkIPinIP
conf.checkIPinIP = True
px = [IP()/p1, IPv6()/p1]
assert not any(p.hashret() == p2.hashret() for p in px)
assert not any(p.answers(p2) for p in px)
assert not any(p2.answers(p) for p in px)
conf.checkIPinIP = False
assert all(p.hashret() == p2.hashret() for p in px)
assert not any(p.answers(p2) for p in px)
assert all(p2.answers(p) for p in px)
conf.checkIPinIP = conf_back

= answers - Net
~ netaccess

a1, a2 = Net("www.google.com"), Net("www.secdev.org")
prt1, prt2 = 12345, 54321
s1, s2 = 2767216324, 3845532842
p1 = IP(src=a1, dst=a2)/TCP(flags='SA', seq=s1, ack=s2, sport=prt1, dport=prt2)
p2 = IP(src=a2, dst=a1)/TCP(flags='R', seq=s2, ack=0, sport=prt2, dport=prt1)
assert p2.answers(p1)
assert not p1.answers(p2)
# Not available yet because of IPv6
# a1, a2 = Net6("www.google.com"), Net6("www.secdev.org")
p1 = IP(src=a1, dst=a2)/TCP(flags='S', seq=s1, ack=0, sport=prt1, dport=prt2)
p2 = IP(src=a2, dst=a1)/TCP(flags='RA', seq=0, ack=s1+1, sport=prt2, dport=prt1)
assert p2.answers(p1)
assert not p1.answers(p2)
p1 = IP(src=a1, dst=a2)/TCP(flags='S', seq=s1, ack=0, sport=prt1, dport=prt2)
p2 = IP(src=a2, dst=a1)/TCP(flags='SA', seq=s2, ack=s1+1, sport=prt2, dport=prt1)
assert p2.answers(p1)
assert not p1.answers(p2)
p1 = IP(src=a1, dst=a2)/TCP(flags='A', seq=s1, ack=s2+1, sport=prt1, dport=prt2)
assert not p2.answers(p1)
assert p1.answers(p2)
p1 = IP(src=a1, dst=a2)/TCP(flags='S', seq=s1, ack=0, sport=prt1, dport=prt2)
p2 = IP(src=a2, dst=a1)/TCP(flags='SA', seq=s2, ack=s1+10, sport=prt2, dport=prt1)
assert not p2.answers(p1)
assert not p1.answers(p2)
p1 = IP(src=a1, dst=a2)/TCP(flags='A', seq=s1, ack=s2+1, sport=prt1, dport=prt2)
assert not p2.answers(p1)
assert not p1.answers(p2)
p1 = IP(src=a1, dst=a2)/TCP(flags='A', seq=s1+9, ack=s2+10, sport=prt1, dport=prt2)
assert not p2.answers(p1)
assert not p1.answers(p2)

= conf.checkIPsrc

conf_checkIPsrc = conf.checkIPsrc
conf.checkIPsrc = 0
query = IP(id=42676, src='10.128.0.7', dst='192.168.0.1')/ICMP(id=26)
answer = IP(src='192.168.48.19', dst='10.128.0.7')/ICMP(type=11)/IPerror(id=42676, src='192.168.51.23', dst='192.168.0.1')/ICMPerror(id=26)
assert answer.answers(query)
conf.checkIPsrc = conf_checkIPsrc


############
############
+ Tests on padding

= Padding assembly
r = raw(Padding("abc"))
r
assert r == b"abc" 
r = raw(Padding("abc")/Padding("def"))
r
assert r == b"abcdef" 
r = raw(Raw("ABC")/Padding("abc")/Padding("def"))
r
assert r == b"ABCabcdef" 
r = raw(Raw("ABC")/Padding("abc")/Raw("DEF")/Padding("def"))
r
assert r == b"ABCDEFabcdef"

= Padding and length computation
p = IP(raw(IP()/Padding("abc")))
p
assert p.len == 20 and len(p) == 23
p = IP(raw(IP()/Raw("ABC")/Padding("abc")))
p
assert p.len == 23 and len(p) == 26
p = IP(raw(IP()/Raw("ABC")/Padding("abc")/Padding("def")))
p
assert p.len == 23 and len(p) == 29

= PadField test
~ PadField padding

class TestPad(Packet):
    fields_desc = [ PadField(StrNullField("st", b""),4), StrField("id", b"")]

TestPad() == TestPad(raw(TestPad()))

= ReversePadField
~ PadField padding

class TestReversePad(Packet):
    fields_desc = [ ByteField("a", 0),
                    ReversePadField(IntField("b", 0), 4)]

assert raw(TestReversePad(a=1, b=0xffffffff)) == b'\x01\x00\x00\x00\xff\xff\xff\xff'
assert TestReversePad(raw(TestReversePad(a=1, b=0xffffffff))).b == 0xffffffff

############
############
+ Tests on default value changes mechanism

= Creation of an IPv3 class from IP class with different default values
class IPv3(IP):
    version = 3
    ttl = 32

= Test of IPv3 class
a = IPv3()
v,t = a.version, a.ttl
v,t
assert (v,t) == (3,32)
r = raw(a)
r
assert r == b'5\x00\x00\x14\x00\x01\x00\x00 \x00\xac\xe7\x7f\x00\x00\x01\x7f\x00\x00\x01'

############
############
+ ASN.1 tests

= ASN1 - ASN1_Object
assert ASN1_Object(1) == ASN1_Object(1)
assert ASN1_Object(1) > ASN1_Object(0)
assert ASN1_Object(1) >= ASN1_Object(1)
assert ASN1_Object(0) < ASN1_Object(1)
assert ASN1_Object(1) <= ASN1_Object(2)
assert ASN1_Object(1) != ASN1_Object(2)
ASN1_Object(2).show()

= ASN1 - RandASN1Object
a = RandASN1Object()
random.seed(0x2807)
o = bytes(a)
o
assert o in [
    b'\x1e\x023V',  # PyPy 2.7
    b'A\x02\x07q',  # Python 2.7
    b'B\x02\xfe\x92',  # python 3.7-3.9
]

= ASN1 - ASN1_BIT_STRING
a = ASN1_BIT_STRING("test", readable=True)
a
assert a.val == '01110100011001010111001101110100'
assert raw(a) == b'\x03\x05\x00test'

a = ASN1_BIT_STRING(b"\xff"*16, readable=True)
a
assert a.val == "1" * 128
assert raw(a) == b'\x03\x11\x00\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff\xff'

= ASN1 - ASN1_SEQUENCE
a = ASN1_SEQUENCE([ASN1_Object(1), ASN1_Object(0)])
assert a.strshow() == '# ASN1_SEQUENCE:\n  <ASN1_Object[1]>\n  <ASN1_Object[0]>\n'

= ASN1 - ASN1_DECODING_ERROR
a = ASN1_DECODING_ERROR("error", exc=OSError(1))
assert repr(a) == "<ASN1_DECODING_ERROR['error']{{1}}>"
b = ASN1_DECODING_ERROR("error", exc=OSError(ASN1_BIT_STRING("0")))
assert repr(b) in ["<ASN1_DECODING_ERROR['error']{{<ASN1_BIT_STRING[0]=b'\\x00' (7 unused bits)>}}>",
                   "<ASN1_DECODING_ERROR['error']{{<ASN1_BIT_STRING[0]='\\x00' (7 unused bits)>}}>"]

= ASN1 - ASN1_INTEGER
a = ASN1_INTEGER(int("1"*23))
assert repr(a) in ["0x25a55a46e5da99c71c7 <ASN1_INTEGER[1111111111...1111111111]>",
                   "0x25a55a46e5da99c71c7 <ASN1_INTEGER[1111111111...111111111L]>"]

= ASN1 - ASN1_OID
assert raw(ASN1_OID("")) == b"\x06\x00"

= RandASN1Object(), specific crashes

import random

# ASN1F_NUMERIC_STRING
random.seed(1514315682)
raw(RandASN1Object())

# ASN1F_VIDEOTEX_STRING
random.seed(1240186058)
raw(RandASN1Object())

# ASN1F_UTC_TIME & ASN1F_GENERALIZED_TIME
random.seed(1873503288)
raw(RandASN1Object())

= SSID is parsed properly even with the presence of RSN Information
~ SSID RSN Information
# A regression test for https://github.com/secdev/scapy/pull/2685.
# https://github.com/secdev/scapy/issues/2683 describes a packet with
# RSN Information that isn't parsed properly,
# causing the SSID to be overridden.
# This test checks the SSID is parsed properly.
filename = scapy_path("/test/pcaps/bad_rsn_parsing_overrides_ssid.pcap")
frame = rdpcap(filename)[0]
beacon = frame.getlayer(5)
ssid = beacon.network_stats()['ssid']
assert ssid == "ROUTE-821E295"


############
############
+ Network tests

* Those tests need network access

= Sending and receiving an ICMP
~ netaccess needs_root IP ICMP icmp_firewall
def _test():
    with no_debug_dissector():
    	x = sr1(IP(dst="www.google.com")/ICMP(),timeout=3)
    x
    assert x[IP].ottl() in [32, 64, 128, 255]
    assert 0 <= x[IP].hops() <= 126
    x is not None and ICMP in x and x[ICMP].type == 0

retry_test(_test)

= Sending a TCP syn message at layer 2 and layer 3
~ netaccess needs_root IP
def _test():
    tmp = send(IP(dst="8.8.8.8")/TCP(sport=RandShort(), dport=53, flags="S"), return_packets=True, realtime=True)
    assert len(tmp) == 1
    
    tmp = sendp(Ether()/IP(dst="8.8.8.8")/TCP(sport=RandShort(), dport=53, flags="S"), return_packets=True, realtime=True)
    assert len(tmp) == 1
    
    p = Ether()/IP(dst="8.8.8.8")/TCP(sport=RandShort(), dport=53, flags="S")
    from decimal import Decimal
    p.time = Decimal(p.time)
    tmp = sendp(p, return_packets=True, realtime=True)
    assert len(tmp) == 1

retry_test(_test)

= Latency check: localhost ICMP
~ netaccess needs_root linux latency

sock = conf.L3socket
conf.L3socket = L3RawSocket

def _test():
    req = IP(dst="127.0.0.1")/ICMP()
    ans = sr1(req, timeout=3)
    assert (ans.time - req.sent_time) >= 0
    assert (ans.time - req.sent_time) <= 1e-3

try:
    retry_test(_test)
finally:
    conf.L3socket = sock

= Test sniffing on multiple sockets
~ netaccess needs_root sniff

# This test sniffs on the same interface twice at the same time, to
# simulate sniffing on multiple interfaces.

def _test():
	iface = conf.route.route("www.google.com")[0]
	port = int(RandShort())
	pkt = IP(dst="www.google.com")/TCP(sport=port, dport=80, flags="S")
	def cb():
	    sr1(pkt, timeout=3)
	sniffer = AsyncSniffer(started_callback=cb,
			       iface=[iface, iface],
			       lfilter=lambda x: TCP in x and x[TCP].dport == port,
			       prn=lambda x: x.summary(),
			       count=2)
	sniffer.start()
	sniffer.join(timeout=3)
	assert len(sniffer.results) == 2
	for pkt in sniffer.results:
	    assert pkt.sniffed_on == iface

retry_test(_test)

= Sending a TCP syn 'forever' at layer 2 and layer 3
~ netaccess needs_root IP
def _test():
    tmp = srloop(IP(dst="8.8.8.8")/TCP(sport=RandShort(), dport=53, flags="S"), count=1, timeout=3)
    assert type(tmp) == tuple and len(tmp[0]) == 1
    
    tmp = srploop(Ether()/IP(dst="8.8.8.8")/TCP(sport=RandShort(), dport=53, flags="S"), count=1, timeout=3)
    assert type(tmp) == tuple and len(tmp[0]) == 1

retry_test(_test)

= Sending and receiving an TCP syn with flooding methods
~ netaccess needs_root IP flood
from functools import partial
# flooding methods do not support timeout. Packing the test for security
def _test_flood(ip, flood_function, add_ether=False):
    with no_debug_dissector():
        p = IP(dst=ip)/TCP(sport=RandShort(), dport=80, flags="S")
        if add_ether:
            p = Ether()/p
        p.show2()
        x = flood_function(p, timeout=0.5, maxretries=10)
    if type(x) == tuple:
        x = x[0][0][1]
    x
    assert x[IP].ottl() in [32, 64, 128, 255]
    assert 0 <= x[IP].hops() <= 126

_test_srflood = partial(_test_flood, "www.google.com", srflood)
retry_test(_test_srflood)

_test_sr1flood = partial(_test_flood, "www.google.fr", sr1flood)
retry_test(_test_sr1flood)

_test_srpflood = partial(_test_flood, "www.google.net", srpflood, True)
retry_test(_test_srpflood)

_test_srp1flood = partial(_test_flood, "www.google.co.uk", srp1flood, True)
retry_test(_test_srp1flood)

= test chainEX
~ netaccess

import socket
sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ssck = StreamSocket(sck)

try:
    r = ssck.sr1(ICMP(type='echo-request'), timeout=0.1, chainEX=True)
    assert False
except Exception:
    assert True
finally:
    sck.close()

= Sending and receiving an ICMPv6EchoRequest
~ netaccess ipv6
def _test():
    with no_debug_dissector():
        x = sr1(IPv6(dst="www.google.com")/ICMPv6EchoRequest(),timeout=3)
    x
    assert x[IPv6].ottl() in [32, 64, 128, 255]
    assert 0 <= x[IPv6].hops() <= 126
    x is not None and ICMPv6EchoReply in x and x[ICMPv6EchoReply].type == 129

retry_test(_test)

= Whois request
~ netaccess IP
* This test retries on failure because it often fails
def _test():
    IP(src="8.8.8.8").whois()

retry_test(_test)

= AS resolvers
~ netaccess IP as_resolvers
* This test retries on failure because it often fails

def _test():
    ret = conf.AS_resolver.resolve("8.8.8.8", "8.8.4.4")
    assert (len(ret) == 2)
    all(x[1] == "AS15169" for x in ret)

retry_test(_test)

riswhois_data = b"route:      8.8.8.0/24\ndescr:      Google\norigin:     AS15169\nnotify:     radb-contact@google.com\nmnt-by:     MAINT-AS15169\nchanged:    radb-contact@google.com 20150728\nsource:     RADB\n\nroute:         8.0.0.0/9\ndescr:         Proxy-registered route object\norigin:        AS3356\nremarks:       auto-generated route object\nremarks:       this next line gives the robot something to recognize\nremarks:       L'enfer, c'est les autres\nremarks:       \nremarks:       This route object is for a Level 3 customer route\nremarks:       which is being exported under this origin AS.\nremarks:       \nremarks:       This route object was created because no existing\nremarks:       route object with the same origin was found, and\nremarks:       since some Level 3 peers filter based on these objects\nremarks:       this route may be rejected if this object is not created.\nremarks:       \nremarks:       Please contact routing@Level3.net if you have any\nremarks:       questions regarding this object.\nmnt-by:        LEVEL3-MNT\nchanged:       roy@Level3.net 20060203\nsource:        LEVEL3\n\n\n"

ret = AS_resolver_riswhois()._parse_whois(riswhois_data)
assert ret == ('AS15169', 'Google')

retry_test(_test)

# This test is too buggy, and is simulated below
#def _test():
#    ret = AS_resolver_cymru().resolve("8.8.8.8")
#    assert (len(ret) == 1)
#    all(x[1] == "AS15169" for x in ret)
#
#retry_test(_test)

cymru_bulk_data = """
Bulk mode; whois.cymru.com [2017-10-03 08:38:08 +0000]
24776   | 217.25.178.5     | INFOCLIP-AS, FR
36459   | 192.30.253.112   | GITHUB - GitHub, Inc., US
26496   | 68.178.213.61    | AS-26496-GO-DADDY-COM-LLC - GoDaddy.com, LLC, US
"""
tmp = AS_resolver_cymru().parse(cymru_bulk_data)
assert len(tmp) == 3
assert [l[1] for l in tmp] == ['AS24776', 'AS36459', 'AS26496']

= AS resolver - IPv6
~ netaccess IP
* This test retries on failure because it often fails

def _test():
    as_resolver6 = AS_resolver6()
    ret = as_resolver6.resolve("2001:4860:4860::8888", "2001:4860:4860::4444")
    assert (len(ret) == 2)
    assert all(x[1] == 15169 for x in ret)

retry_test(_test)

= AS resolver - socket error
~ IP
* This test checks that a failing resolver will not crash a script

class MockAS_resolver(object):
  def resolve(self, *ips):
    raise socket.error

asrm = AS_resolver_multi(MockAS_resolver())
assert len(asrm.resolve(["8.8.8.8", "8.8.4.4"])) == 0

= sendpfast
~ tcpreplay

old_interactive = conf.interactive
conf.interactive = False
try:
    sendpfast([])
    assert False
except Exception:
    assert True

conf.interactive = old_interactive
assert True

############
############
+ Generator tests

= Implicit logic 1
~ IP TCP
a = Ether() / IP(ttl=(5, 10)) / TCP(dport=[80, 443])
ls(a)
ls(a, verbose=True)
l = [p for p in a]
len(l) == 12

= Implicit logic 2
~ IP
a = IP(ttl=[1,2,(5,9)])
ls(a)
ls(a, verbose=True)
l = [p for p in a]
len(l) == 7

= Implicit logic 3
# In case there's a single option: __iter__ should not return self
a = Ether()/IP(src="127.0.0.1", dst="127.0.0.1")/ICMP()
for i in a:
    i.sent_time = 1

assert a.sent_time is None

# In case they are several, self should never be returned
a = Ether()/IP(src="127.0.0.1", dst="127.0.0.1")/ICMP(seq=(0, 5))
for i in a:
    i.sent_time = 1

assert a.sent_time is None


############
############
+ Real usages

= Port scan
~ netaccess needs_root IP TCP
def _test():
    with no_debug_dissector():
        ans,unans=sr(IP(dst="www.google.com/30")/TCP(dport=[80,443]), timeout=2)
    
    # Backward compatibility: Python 2 only
    if six.PY2:
        exec("""ans.make_table(lambda (s, r): (s.dst, s.dport, r.sprintf("{TCP:%TCP.flags%}{ICMP:%ICMP.code%}")))""")
    
    # New format: all Python versions
    ans.make_table(lambda s, r: (s.dst, s.dport, r.sprintf("{TCP:%TCP.flags%}{ICMP:%ICMP.code%}")))

retry_test(_test)

= Send & receive with debug_match
~ netaccess needs_root IP ICMP
def _test():
    old_debug_match = conf.debug_match
    conf.debug_match = True
    with no_debug_dissector():
        ans, unans = sr(IP(dst="www.google.fr") / TCP(sport=RandShort(), dport=80, flags="S"), timeout=2)
        assert ans[0].query == ans[0][0]
        assert ans[0].answer == ans[0][1]
    conf.debug_match = old_debug_match
    assert ans and not unans

retry_test(_test)

= Send & receive with retry
~ netaccess needs_root IP ICMP
def _test():
    with no_debug_dissector():
        ans, unans = sr(IP(dst=["8.8.8.8", "1.2.3.4"]) / TCP(sport=RandShort(), dport=53, flags="S"), timeout=2, retry=1)
    assert len(ans) == 1 and len(unans) == 1

retry_test(_test)

= Send & receive with multi
~ netaccess needs_root IP ICMP
def _test():
    with no_debug_dissector():
        ans, unans = sr(IP(dst=["8.8.8.8", "1.2.3.4"]) / TCP(sport=RandShort(), dport=53, flags="S"), timeout=2, multi=1)
    assert len(ans) >= 1 and len(unans) == 1

retry_test(_test)

= Traceroute function
~ netaccess needs_root tcpdump
* Let's test traceroute
def _test():
    ans, unans = traceroute("www.slashdot.org")
    ans.nsummary()
    s,r=ans[0]
    s.show()
    s.show(2)

retry_test(_test)

= send() and sniff()
~ netaccess needs_root

def _test():
    sendp(Ether()/IP(src="9.0.0.0")/UDP(), count=3, iface=conf.iface)

r = sniff(timeout=3, count=1,
          lfilter=lambda x: IP in x and x[IP].src == "9.0.0.0",
          iface=conf.iface,
          started_callback=_test)

assert r

= sniff() with socket failure
* GH issue 3631

REFPACKET = Ether()/IP()/UDP()

# A socket that fails after 10 packets
class OOPipe(ObjectPipe):
     def recv(self, x=MTU):
         self.i = getattr(self, "i", 0) + 1
         if self.i == 11:
             self.close()
             raise OSError("Giant failure")
         pkt = super(OOPipe, self).recv(x)
         self.send(REFPACKET)
         return pkt

o = OOPipe()
o.send(REFPACKET)

pkts = sniff(opened_socket=[o], timeout=3)
assert len(pkts) == 10

= GH issue 3306
~ netaccess needs_root

send(fuzz(ARP()))

= Test SuperSocket.select
~ select

import mock

@mock.patch("scapy.supersocket.select")
def _test_select(select):
    def f(a, b, c, d):
        raise IOError(0)
    select.side_effect = f
    try:
        SuperSocket.select([])
        return False
    except:
        return True

assert _test_select()

= Test L2ListenTcpdump socket
~ netaccess

# Needs to be fixed. Fails randomly
#import time
#for i in range(10):
#    read_s = L2ListenTcpdump(iface=conf.iface)
#    out_s = conf.L2socket(iface=conf.iface)
#    time.sleep(5)  # wait for read_s to be ready
#    icmp_r = Ether()/IP(dst="secdev.org")/ICMP()
#    res = sndrcv(out_s, icmp_r, timeout=5, rcv_pks=read_s)[0]
#    read_s.close()
#    out_s.close()
#    time.sleep(5)
#    if res:
#        break
#
#response = res[0][1]
#assert response[ICMP].type == 0

True

= Test set of sent_time by sr
~ netaccess needs_root IP ICMP
def _test():
    packet = IP(dst="8.8.8.8")/ICMP()
    r = sr(packet, timeout=2)
    assert packet.sent_time is not None

retry_test(_test)

= Test set of sent_time by sr (multiple packets)
~ netaccess needs_root IP ICMP
def _test():
    packet1 = IP(dst="8.8.8.8")/ICMP()
    packet2 = IP(dst="8.8.4.4")/ICMP()
    r = sr([packet1, packet2], timeout=2)
    assert packet1.sent_time is not None
    assert packet2.sent_time is not None

retry_test(_test)

= Test set of sent_time by srflood
~ netaccess needs_root IP ICMP
def _test():
    packet = IP(dst="8.8.8.8")/ICMP()
    r = srflood(packet, timeout=2)
    assert packet.sent_time is not None

retry_test(_test)

= Test set of sent_time by srflood (multiple packets)
~ netaccess needs_root IP ICMP
def _test():
    packet1 = IP(dst="8.8.8.8")/ICMP()
    packet2 = IP(dst="8.8.4.4")/ICMP()
    r = srflood([packet1, packet2], timeout=2)
    assert packet1.sent_time is not None
    assert packet2.sent_time is not None

retry_test(_test)

############
############
+ ManuFDB tests

= __repr__

if conf.manufdb:
    len(conf.manufdb)
else:
    True

= check _resolve_MAC

if conf.manufdb:
    assert conf.manufdb._resolve_MAC("00:00:17") == "Oracle"
else:
    True


############
############
+ Ether tests with IPv6

= Ether IPv6 checking for dst
~ netaccess ipv6

p = Ether()/IPv6(dst="www.google.com")/TCP()
assert p.dst != p[IPv6].dst
p.show()


############
############
+ pcap / pcapng format support

= Variable creations
from io import BytesIO
pcapfile = BytesIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacVo*\n\x00(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV_-\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\xf90\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00')
pcapngfile = BytesIO(b'\n\r\r\n\\\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\x01\x00,\x00File created by merging: \nFile1: test.pcap \n\x04\x00\x08\x00mergecap\x00\x00\x00\x00\\\x00\x00\x00\x01\x00\x00\x00\\\x00\x00\x00e\x00\x00\x00\xff\xff\x00\x00\x02\x006\x00Unknown/not available in original file format(libpcap)\x00\x00\t\x00\x01\x00\x06\x00\x00\x00\x00\x00\x00\x00\\\x00\x00\x00\x06\x00\x00\x00H\x00\x00\x00\x00\x00\x00\x00\x8d*\x05\x00/\xfc[\xcd(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00H\x00\x00\x00\x06\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x8d*\x05\x00\x1f\xff[\xcd\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r<\x00\x00\x00\x06\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x8d*\x05\x00\xb9\x02\\\xcd\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00<\x00\x00\x00')
pcapnanofile = BytesIO(b"M<\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacV\xc9\xc1\xb5'(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV-;\xc1'\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\x9aL\xcf'\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00")
pcapwirelenfile = BytesIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00\x01\x00\x00\x00}\x87pZ.\xa2\x08\x00\x0f\x00\x00\x00\x10\x00\x00\x00\xff\xff\xff\xff\xff\xff GG\xee\xdd\xa8\x90\x00a')
pcapngdefaults = BytesIO(base64_bytes(b'Cg0NChwAAABNPCsaAQAAAP//////////HAAAAAEAAAAgAAAAEgEAAP//AAAJAAEACUeZiQAAAAAgAAAAAQAAACAAAAASAQAA//8AAAkAAQAJAAAAAAAAACAAAAABAAAAIAAAABIBAAD//wAACQABAAkAAAAAAAAAIAAAAAEAAAAgAAAAEgEAAP//AAAJAAEACQAAAAAAAAAgAAAABgAAAIQBAAADAAAApO/bFdgJaeBiAQAAYgEAAFVVVVVVVVXV////////IMbr4D7PCABFAAFIlQkAAEAR5JwAAAAA/////wBEAEMBNJDsAQEGAFSpVwIACoAAAAAAAAAAAAAAAAAAAAAAACDG6+A+zwAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABjglNjNQEB/wAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAsOs+bAAAhAEAAAYAAACAAQAAAwAAAKTv2xXIDYznYAEAAGABAABVVVVVVVVV1QEAXn//+iDG6+A+zwgARQABRgGPAAAEEal3qf5wqO////rhbgdsATJi0U5PVElGWSAqIEhUVFAvMS4xDQpIT1NUOiAyMzkuMjU1LjI1NS4yNTA6MTkwMA0KQ0FDSEUtQ09OVFJPTDogbWF4LWFnZT0xODAwDQpMT0NBVElPTjogaHR0cDovLzE2OS4yNTQuMTEyLjE2ODo1NTAwMC9ucmMvZGRkLnhtbA0KTlQ6IHV1aWQ6NEQ0NTQ5MzAtMDIwMC0xMDAwLTgwMDEtMjBDNkVCRTAzRUNGDQpOVFM6IHNzZHA6YWxpdmUNClNFUlZFUjogRnJlZUJTRC84LjAgVVBuUC8xLjAgUGFuYXNvbmljLU1JTC1ETE5BLVNWLzEuMA0KVVNOOiB1dWlkOjRENDU0OTMwLTAyMDAtMTAwMC04MDAxLTIwQzZFQkUwM0VDRg0KDQpcQcvWgAEAAAYAAAC4AQAAAwAAAKTv2xV4Ao3nlQEAAJUBAABVVVVVVVVV1QEAXn//+iDG6+A+zwgARQABewGQAAAEEalBqf5wqO////rhbgdsAWfu+k5PVElGWSAqIEhUVFAvMS4xDQpIT1NUOiAyMzkuMjU1LjI1NS4yNTA6MTkwMA0KQ0FDSEUtQ09OVFJPTDogbWF4LWFnZT0xODAwDQpMT0NBVElPTjogaHR0cDovLzE2OS4yNTQuMTEyLjE2ODo1NTAwMC9ucmMvZGRkLnhtbA0KTlQ6IHVybjpwYW5hc29uaWMtY29tOmRldmljZTpwMDBSZW1vdGVDb250cm9sbGVyOjENCk5UUzogc3NkcDphbGl2ZQ0KU0VSVkVSOiBGcmVlQlNELzguMCBVUG5QLzEuMCBQYW5hc29uaWMtTUlMLURMTkEtU1YvMS4wDQpVU046IHV1aWQ6NEQ0NTQ5MzAtMDIwMC0xMDAwLTgwMDEtMjBDNkVCRTAzRUNGOjp1cm46cGFuYXNvbmljLWNvbTpkZXZpY2U6cDAwUmVtb3RlQ29udHJvbGxlcjoxDQoNCrLVKmoAAAC4AQAABgAAAHgBAAADAAAApO/bFVjbjedXAQAAVwEAAFVVVVVVVVXVAQBef//6IMbr4D7PCABFAAE9AZEAAAQRqX6p/nCo7///+uFuB2wBKaZATk9USUZZICogSFRUUC8xLjENCkhPU1Q6IDIzOS4yNTUuMjU1LjI1MDoxOTAwDQpDQUNIRS1DT05UUk9MOiBtYXgtYWdlPTE4MDANCkxPQ0FUSU9OOiBodHRwOi8vMTY5LjI1NC4xMTIuMTY4OjU1MDAwL25yYy9kZGQueG1sDQpOVDogdXBucDpyb290ZGV2aWNlDQpOVFM6IHNzZHA6YWxpdmUNClNFUlZFUjogRnJlZUJTRC84LjAgVVBuUC8xLjAgUGFuYXNvbmljLU1JTC1ETE5BLVNWLzEuMA0KVVNOOiB1dWlkOjRENDU0OTMwLTAyMDAtMTAwMC04MDAxLTIwQzZFQkUwM0VDRjo6dXBucDpyb290ZGV2aWNlDQoNCjagXoUAeAEAAAYAAAC0AQAAAwAAAKTv2xXYw47nkwEAAJMBAABVVVVVVVVV1QEAXn//+iDG6+A+zwgARQABeQGSAAAEEalBqf5wqO////rhbgdsAWWV4E5PVElGWSAqIEhUVFAvMS4xDQpIT1NUOiAyMzkuMjU1LjI1NS4yNTA6MTkwMA0KQ0FDSEUtQ09OVFJPTDogbWF4LWFnZT0xODAwDQpMT0NBVElPTjogaHR0cDovLzE2OS4yNTQuMTEyLjE2ODo1NTAwMC9ucmMvZGRkLnhtbA0KTlQ6IHVybjpwYW5hc29uaWMtY29tOnNlcnZpY2U6cDAwTmV0d29ya0NvbnRyb2w6MQ0KTlRTOiBzc2RwOmFsaXZlDQpTRVJWRVI6IEZyZWVCU0QvOC4wIFVQblAvMS4wIFBhbmFzb25pYy1NSUwtRExOQS1TVi8xLjANClVTTjogdXVpZDo0RDQ1NDkzMC0wMjAwLTEwMDAtODAwMS0yMEM2RUJFMDNFQ0Y6OnVybjpwYW5hc29uaWMtY29tOnNlcnZpY2U6cDAwTmV0d29ya0NvbnRyb2w6MQ0KDQovXKFrALQBAAAGAAAAqAEAAAMAAACk79sVuJKP54cBAACHAQAAVVVVVVVVVdUBAF5///ogxuvgPs8IAEUAAW0BkwAABBGpTKn+cKjv///64W4HbAFZRNJOT1RJRlkgKiBIVFRQLzEuMQ0KSE9TVDogMjM5LjI1NS4yNTUuMjUwOjE5MDANCkNBQ0hFLUNPTlRST0w6IG1heC1hZ2U9MTgwMA0KTE9DQVRJT046IGh0dHA6Ly8xNjkuMjU0LjExMi4xNjg6NTUwMDAvbnJjL2RkZC54bWwNCk5UOiB1cm46ZGlhbC1tdWx0aXNjcmVlbi1vcmc6c2VydmljZTpkaWFsOjENCk5UUzogc3NkcDphbGl2ZQ0KU0VSVkVSOiBGcmVlQlNELzguMCBVUG5QLzEuMCBQYW5hc29uaWMtTUlMLURMTkEtU1YvMS4wDQpVU046IHV1aWQ6NEQ0NTQ5MzAtMDIwMC0xMDAwLTgwMDEtMjBDNkVCRTAzRUNGOjp1cm46ZGlhbC1tdWx0aXNjcmVlbi1vcmc6c2VydmljZTpkaWFsOjENCg0KLn5A6QCoAQAA'))

= Read a pcap file
pktpcap = rdpcap(pcapfile)

= Read a pcapng file
pktpcapng = rdpcap(pcapngfile)
assert pktpcapng[0].time == 1454163407.666223

= Read a pcap file with nanosecond precision
pktpcapnano = rdpcap(pcapnanofile)
assert pktpcapnano[0].time == 1454163407.666223049

= Read a pcapng file with nanosecond precision and default tsresol
pktpcapngdefaults = rdpcap(pcapngdefaults)
assert pktpcapngdefaults[0].time == 1575115986.114775512
assert Ether in pktpcapngdefaults[0]

= Read a pcapng with little-endian SHB
pktcapng = sniff(offline=scapy_path("/test/pcaps/macos.pcapng.gz"))
assert len(pktcapng) != 0

= Write a pcapng

tmpfile = get_temp_file(autoext=".pcapng")
r = RawPcapNgWriter(tmpfile)
r._write_block_shb()
r._write_block_idb()
ts = 1632568366.384185
r._write_block_epb(raw(Ether()/"Hello Scapy!!!"), ts)
r.f.close()

assert os.stat(tmpfile).st_size == 108

l = rdpcap(tmpfile)
assert b"Scapy" in l[0][Raw].load
assert l[0].time == ts

= Check wrpcapng()

tmpfile = get_temp_file(autoext=".pcapng")
p = Ether()/"Hello Scapy!!!"
p.time = 1632568366.384185
wrpcapng(tmpfile, p)

assert os.stat(tmpfile).st_size == 108

l = rdpcap(tmpfile)
assert b"Scapy" in l[0][Raw].load
assert l[0].time == ts

p = Ether() / IPv6() / TCP()
p.comment = b"Hello Scapy!"
wrpcapng(tmpfile, p)
l = rdpcap(tmpfile)
assert l[0].comment.strip() == p.comment

= Read a pcap file with wirelen != captured len
pktpcapwirelen = rdpcap(pcapwirelenfile)

= Check all packet lists are the same
assert list(pktpcap) == list(pktpcapng) == list(pktpcapnano)
assert [float(p.time) for p in pktpcap] == [float(p.time) for p in pktpcapng] == [float(p.time) for p in pktpcapnano]

= Check packets from pcap file
assert all(IP in pkt for pkt in pktpcap)
assert all(any(proto in pkt for pkt in pktpcap) for proto in [ICMP, UDP, TCP])

= Check wirelen value from pcap file
assert len(pktpcapwirelen) == 1
assert pktpcapwirelen[0].wirelen is not None
assert len(pktpcapwirelen[0]) < pktpcapwirelen[0].wirelen

= Check wrpcap() then rdpcap() with wirelen
import os, tempfile
fdesc, filename = tempfile.mkstemp()
fdesc = os.fdopen(fdesc, "wb")
wrpcap(fdesc, pktpcapwirelen)
fdesc.close()
newpktpcapwirelen = rdpcap(filename)
assert len(newpktpcapwirelen) == 1
assert newpktpcapwirelen[0].wirelen is not None
assert len(newpktpcapwirelen[0]) < newpktpcapwirelen[0].wirelen
assert newpktpcapwirelen[0].wirelen == pktpcapwirelen[0].wirelen

= Check wrpcap() then rdpcap() with sent_time on SndRcvList
f = get_temp_file()
s = Ether()/IP()
r = Ether()/IP()
s.sent_time = 1
r.time = 2
wrpcap(f, SndRcvList([(s, r)]))
pcap = rdpcap(f)
assert pcap[0].time == 1
assert pcap[1].time == 2

= Check wrpcap()
fdesc, filename = tempfile.mkstemp()
fdesc = os.fdopen(fdesc, "wb")
wrpcap(fdesc, pktpcap)
fdesc.close()

= Check offline sniff() (by PacketList)
l=sniff(offline=PacketList([IP()/TCP(),IP()/TCP()]))
assert len(l) == 2
assert all(TCP in p for p in l)

= Check offline sniff() (by filename)
assert list(pktpcap) == list(sniff(offline=filename))

= Check offline sniff() (by file object)
fdesc = open(filename, "rb")
assert list(pktpcap) == list(sniff(offline=fdesc))
fdesc.close()

= Check offline sniff() with a filter (by filename)
~ tcpdump libpcap
pktpcap_flt = [(proto, sniff(offline=filename, filter=proto.__name__.lower()))
               for proto in [ICMP, UDP, TCP]]
assert all(list(pktpcap[proto]) == list(packets) for proto, packets in pktpcap_flt)

= Check offline sniff() with a filter (by file object)
~ tcpdump libpcap
fdesc = open(filename, "rb")
pktpcap_tcp = sniff(offline=fdesc, filter="tcp")
fdesc.close()
assert list(pktpcap[TCP]) == list(pktpcap_tcp)
os.unlink(filename)

= Check offline sniff() with a PcapNg file and a filter (by file object)
~ tcpdump libpcap

pcapng_data = b'\n\r\r\n`\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xff\xff\xff\xff\xff\xff\xff\xff\x04\x009\x00TShark (Wireshark) 3.2.3 (Git v3.2.3 packaged as 3.2.3-1)\x00\x00\x00\x00\x00\x00\x00`\x00\x00\x00\x01\x00\x00\x00\x14\x00\x00\x00\xe4\x00\x00\x00\xff\xff\x00\x00\x14\x00\x00\x00\x06\x00\x00\x00<\x00\x00\x00\x00\x00\x00\x00\x98\xcd\x05\x00\x19\x83\xf7\x9e\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r<\x00\x00\x00'

if OPENBSD:
    # Note: OpenBSD tcpdump does not support PcapNg
    assert True
else:
    fdesc, filename = tempfile.mkstemp()
    os.close(fdesc)
    fd = open(filename, "wb")
    fd.write(pcapng_data)
    fd.close()
    packets = sniff(offline=filename, filter="udp")
    os.unlink(filename)
    assert UDP in packets[0]

= Check offline sniff() with Packets and tcpdump with a filter
~ tcpdump libpcap

l = sniff(offline=IP()/UDP(sport=(10000, 10001)), filter="udp")
assert len(l) == 2
assert all(UDP in p for p in l)

l = sniff(offline=[p for p in IP()/UDP(sport=(10000, 10001))], filter="udp")
assert len(l) == 2
assert all(UDP in p for p in l)

l = sniff(offline=IP()/UDP(sport=(10000, 10001)), filter="tcp")
assert len(l) == 0

= Check offline sniff() with Packets, tcpdump and a bad filter
~ tcpdump libpcap

try:
    sniff(offline=IP()/UDP(), filter="bad filter")
except Scapy_Exception:
    pass
else:
    assert False

= Check offline sniff with lfilter
assert len(sniff(offline=[IP()/UDP(), IP()/TCP()], lfilter=lambda x: TCP in x)) == 1

= Check offline sniff() without a tcpdump binary
~ tcpdump
import mock

conf_prog_tcpdump = conf.prog.tcpdump
conf.prog.tcpdump = "tcpdump_fake"

def _test_sniff_notcpdump():
    try:
        sniff(offline="fake.pcap", filter="tcp")
        assert False
    except:
        assert True

_test_sniff_notcpdump()
conf.prog.tcpdump = conf_prog_tcpdump

= Check wrpcap(nano=True)
fdesc, filename = tempfile.mkstemp()
fdesc = os.fdopen(fdesc, "wb")
pktpcapnano[0].time += Decimal('1E-9')
wrpcap(fdesc, pktpcapnano, nano=True)
fdesc.close()
pktpcapnanoread = rdpcap(filename)
assert pktpcapnanoread[0].time == pktpcapnano[0].time
os.unlink(filename)

= Check PcapNg with nanosecond precision using obsolete packet block
* first packet from capture file icmp2.ntar -- https://wiki.wireshark.org/Development/PcapNg?action=AttachFile&do=view&target=icmp2.ntar
pcapngfile = BytesIO(b'\n\r\r\n\x1c\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xa8\x03\x00\x00\x00\x00\x00\x00\x1c\x00\x00\x00\x01\x00\x00\x00(\x00\x00\x00\x01\x00\x00\x00\xff\xff\x00\x00\r\x00\x01\x00\x04\x04K\x00\t\x00\x01\x00\tK=N\x00\x00\x00\x00(\x00\x00\x00\x02\x00\x00\x00n\x00\x00\x00\x00\x00\x00\x00e\x14\x00\x00)4\'ON\x00\x00\x00N\x00\x00\x00\x00\x12\xf0\x11h\xd6\x00\x13r\t{\xea\x08\x00E\x00\x00<\x90\xa1\x00\x00\x80\x01\x8e\xad\xc0\xa8M\x07\xc0\xa8M\x1a\x08\x00r[\x03\x00\xd8\x00abcdefghijklmnopqrstuvwabcdefghi\xeay$\xf6\x00\x00n\x00\x00\x00')
pktpcapng = rdpcap(pcapngfile)
assert len(pktpcapng) == 1
pkt = pktpcapng[0]
# weird, but wireshark agrees
assert pkt.time == 22425.352221737
assert isinstance(pkt, Ether)
pkt = pkt.payload
assert isinstance(pkt, IP)
pkt = pkt.payload
assert isinstance(pkt, ICMP)
pkt = pkt.payload
assert isinstance(pkt, Raw) and pkt.load == b'abcdefghijklmnopqrstuvwabcdefghi'
pkt = pkt.payload
assert isinstance(pkt, Padding) and pkt.load == b'\xeay$\xf6'
pkt = pkt.payload
assert isinstance(pkt, NoPayload)

= Check PcapNg using Simple Packet Block
* previous file with the (obsolete) packet block replaced by a Simple Packet Block
pcapngfile = BytesIO(b'\n\r\r\n\x1c\x00\x00\x00M<+\x1a\x01\x00\x00\x00\xa8\x03\x00\x00\x00\x00\x00\x00\x1c\x00\x00\x00\x01\x00\x00\x00(\x00\x00\x00\x01\x00\x00\x00\xff\xff\x00\x00\r\x00\x01\x00\x04\x04K\x00\t\x00\x01\x00\tK=N\x00\x00\x00\x00(\x00\x00\x00\x03\x00\x00\x00`\x00\x00\x00N\x00\x00\x00\x00\x12\xf0\x11h\xd6\x00\x13r\t{\xea\x08\x00E\x00\x00<\x90\xa1\x00\x00\x80\x01\x8e\xad\xc0\xa8M\x07\xc0\xa8M\x1a\x08\x00r[\x03\x00\xd8\x00abcdefghijklmnopqrstuvwabcdefghi\xeay$\xf6\x00\x00`\x00\x00\x00')
pktpcapng = rdpcap(pcapngfile)
assert len(pktpcapng) == 1
pkt = pktpcapng[0]
assert isinstance(pkt, Ether)
pkt = pkt.payload
assert isinstance(pkt, IP)
pkt = pkt.payload
assert isinstance(pkt, ICMP)
pkt = pkt.payload
assert isinstance(pkt, Raw) and pkt.load == b'abcdefghijklmnopqrstuvwabcdefghi'
pkt = pkt.payload
assert isinstance(pkt, Padding) and pkt.load == b'\xeay$\xf6'
pkt = pkt.payload
assert isinstance(pkt, NoPayload)

= Invalid pcapng files

from io import BytesIO

# Invalid PCAPNG format -> Raise
try:
    invalid_pcapngfile_1 = BytesIO(b'\n\r\r\n\r\x00\x00\x00M<+\x1a\xb2<\xb2\xa1\x01\x00\x00\x00\r\x00\x00\x00M<+\x1a\x80\xaa\xb2\x02')
    rdpcap(invalid_pcapngfile_1)
    assert False
except Scapy_Exception:
    pass

# Invalid Packet in PCAPNG -> return
invalid_pcapngfile_2 = BytesIO(b'\n\r\r\n\x00\x00\x00\x10\x1a+<M\x00\x00\x00\x10\x00\x00\x00\x01\x00\x00\x00\x10    \x00\x00\x00\x10')
assert len(rdpcap(invalid_pcapngfile_2)) == 0

# Invalid interface ID in PCAPNG -> raise EOFError
try:
    invalid_pcapngfile_3 = BytesIO(b'\n\n\n\x14\x00\x00\x00M<+\x1a    \x14\x00\x00\x00\x03\x00\x00\x00\x14\x00\x00\x00        \x14\x00\x00\x00')
    rdpcap(invalid_pcapngfile_3)
    assert False
except Scapy_Exception:
    pass

# Invalid SPB in PCAPNG -> raise EOFError
try:
    invalid_pcapngfile_4 = BytesIO(b'\n\n\n\x14\x00\x00\x00M<+\x1a    \x14\x00\x00\x00\x01\x00\x00\x00\x14\x00\x00\x00        \x14\x00\x00\x00\x03\x00\x00\x00\x0c\x00\x00\x00\x0c\x00\x00\x00')
    rdpcap(invalid_pcapngfile_4)
    assert False
except Scapy_Exception:
    pass

= Check PcapWriter on null write

f = BytesIO()
w = PcapWriter(f)
w.write([])
assert len(f.getvalue()) == 0

# Stop being closed for reals, but we still want to have the header written
with mock.patch.object(f, 'close') as cf:
    w.close()

cf.assert_called_once_with()
assert len(f.getvalue()) != 0

= Check PcapWriter sets correct linktype after null write

f = BytesIO()
w = PcapWriter(f)
w.write([])
assert len(f.getvalue()) == 0
w.write(Ether()/IP()/ICMP())
assert len(f.getvalue()) != 0

# Stop being closed for reals, but we still want to have the header written
with mock.patch.object(f, 'close') as cf:
    w.close()

cf.assert_called_once_with()
f.seek(0) or None
assert len(f.getvalue()) != 0

r = PcapReader(f)
f.seek(0) or None
assert r.LLcls is Ether
assert r.linktype == DLT_EN10MB

l = [ p for p in RawPcapReader(f) ]
assert len(l) == 1

= Check RawPcapReader on pcap
~ pcap

fd = get_temp_file()
wrpcap(fd, [Ether()/IP()/ICMP()])
assert len([p for p in RawPcapReader(fd)]) == 1

for (x, y) in RawPcapReader(fd):
    pass

= Check RawPcapWriter
~ pcap

# GH3256
fd = get_temp_file()
with RawPcapWriter(fd, linktype=1) as w:
    w.write(b"test")

fd = get_temp_file()
with RawPcapWriter(fd) as w:
    w.write(b"test")
    assert w.linktype == 1

= Check tcpdump()
~ tcpdump
from io import BytesIO
* No very specific tests because we do not want to depend on tcpdump output
pcapfile = BytesIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00\x01\x00\x00\x000}$]\xff\\\t\x006\x00\x00\x006\x00\x00\x00\xff\xff\xff\xff\xff\xff\x00\x00\x00\x00\x00\x00\x08\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x000}$]\x87i\t\x00*\x00\x00\x00*\x00\x00\x00\xff\xff\xff\xff\xff\xff\x00\x00\x00\x00\x00\x00\x08\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r0}$]\xfbp\t\x00*\x00\x00\x00*\x00\x00\x00\xff\xff\xff\xff\xff\xff\x00\x00\x00\x00\x00\x00\x08\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00')

data = tcpdump(pcapfile, dump=True, args=['-nn']).split(b'\n')
print(data)
assert b'127.0.0.1.20 > 127.0.0.1.80:' in data[0]
assert b'127.0.0.1.53 > 127.0.0.1.53:' in data[1]
assert b'127.0.0.1 > 127.0.0.1:' in data[2]

* Non existing tcpdump binary

import mock

conf_prog_tcpdump = conf.prog.tcpdump
conf.prog.tcpdump = "tcpdump_fake"

def _test_tcpdump_notcpdump():
    try:
        tcpdump(IP()/TCP())
        assert False
    except:
        assert True

_test_tcpdump_notcpdump()
conf.prog.tcpdump = conf_prog_tcpdump

# Also check with use_tempfile=True (for non-OSX platforms)
pcapfile.seek(0) or None
tempfile_count = len(conf.temp_files)
data = tcpdump(pcapfile, dump=True, args=['-nn'], use_tempfile=True).split(b'\n')
print(data)
assert b'127.0.0.1.20 > 127.0.0.1.80:' in data[0]
assert b'127.0.0.1.53 > 127.0.0.1.53:' in data[1]
assert b'127.0.0.1 > 127.0.0.1:' in data[2]
# We should have another tempfile tracked.
assert len(conf.temp_files) > tempfile_count

# Check with a simple packet
data = tcpdump([Ether()/IP()/ICMP()], dump=True, args=['-nn']).split(b'\n')
print(data)
assert b'127.0.0.1 > 127.0.0.1: ICMP' in data[0].upper()

= Check tcpdump() command with linktype 
~ tcpdump libpcap

f = BytesIO()
pkt = Ether()/IP()/ICMP()

with mock.patch('subprocess.Popen', return_value=Bunch(
        stdin=f, wait=lambda: None)) as popen:
    # Prevent closing the BytesIO
    with mock.patch.object(f, 'close'):
        tcpdump([pkt], linktype="DLT_EN10MB", use_tempfile=False)

expected_command = [conf.prog.tcpdump, '-y', 'EN10MB', '-U', '-r', '-']
if OPENBSD:
    expected_command = [conf.prog.tcpdump, '-y', 'EN10MB', '-r', '-']

popen.assert_called_once_with(
    expected_command,
    stdin=subprocess.PIPE, stdout=None, stderr=None)

print(bytes_hex(f.getvalue()))
assert raw(pkt) in f.getvalue()
f.close()
del f, pkt

= Check tcpdump() command with linktype and args
~ tcpdump libpcap

f = BytesIO()
pkt = Ether()/IP()/ICMP()

with mock.patch('subprocess.Popen', return_value=Bunch(
        stdin=f, wait=lambda: None)) as popen:
    # Prevent closing the BytesIO
    with mock.patch.object(f, 'close'):
        tcpdump([pkt], linktype=scapy.data.DLT_EN10MB, use_tempfile=False)

expected_command = [conf.prog.tcpdump, '-y', 'EN10MB', '-U', '-r', '-']
if OPENBSD:
    expected_command = [conf.prog.tcpdump, '-y', 'EN10MB', '-r', '-']

popen.assert_called_once_with(
    expected_command,
    stdin=subprocess.PIPE, stdout=None, stderr=None)

print(bytes_hex(f.getvalue()))
assert raw(pkt) in f.getvalue()
f.close()
del f, pkt

= Check sniff() offline with linktype & 802.11 filter
~ tcpdump linux

fd = get_temp_file()
wrpcap(fd, [RadioTap()/Dot11()/Dot11ProbeReq(), RadioTap()/Dot11()])
lst = sniff(offline=fd, filter="subtype probe-req")
assert len(lst) == 1

= Check tcpdump() command rejects non-string input for prog

pkt = Ether()/IP()/ICMP()

try:
    tcpdump([pkt], prog=+17607067425, args=['-nn'])
except ValueError as e:
    if hasattr(e, 'args'):
        assert 'prog' in e.args[0]
    else:
        assert 'prog' in e.message
else:
    assert False, 'expected exception'

= Check tcpdump() command with tshark
~ tshark
pcapfile = BytesIO(b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00e\x00\x00\x00\xcf\xc5\xacVo*\n\x00(\x00\x00\x00(\x00\x00\x00E\x00\x00(\x00\x01\x00\x00@\x06|\xcd\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x14\x00P\x00\x00\x00\x00\x00\x00\x00\x00P\x02 \x00\x91|\x00\x00\xcf\xc5\xacV_-\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x11|\xce\x7f\x00\x00\x01\x7f\x00\x00\x01\x005\x005\x00\x08\x01r\xcf\xc5\xacV\xf90\n\x00\x1c\x00\x00\x00\x1c\x00\x00\x00E\x00\x00\x1c\x00\x01\x00\x00@\x01|\xde\x7f\x00\x00\x01\x7f\x00\x00\x01\x08\x00\xf7\xff\x00\x00\x00\x00')
# tshark doesn't need workarounds on OSX
tempfile_count = len(conf.temp_files)
values = [tuple(int(val) for val in line[:-1].split(b'\t')) for line in tcpdump(pcapfile, prog=conf.prog.tshark, getfd=True, args=['-T', 'fields', '-e', 'ip.ttl', '-e', 'ip.proto'])]
assert values == [(64, 6), (64, 17), (64, 1)]
assert len(conf.temp_files) == tempfile_count

= Check tdecode command directly for tshark
~ tshark

pkts = [
    Ether()/IP(src='192.0.2.1', dst='192.0.2.2')/ICMP(type='echo-request')/Raw(b'X'*100),
    Ether()/IP(src='192.0.2.2', dst='192.0.2.1')/ICMP(type='echo-reply')/Raw(b'X'*100),
]

# tshark doesn't need workarounds on OSX
tempfile_count = len(conf.temp_files)

r = tdecode(pkts, dump=True)
r
assert b'Src: 192.0.2.1' in r
assert b'Src: 192.0.2.2' in r
assert b'Dst: 192.0.2.2' in r
assert b'Dst: 192.0.2.1' in r
assert b'Echo (ping) request' in r
assert b'Echo (ping) reply' in r
assert b'ICMP' in r
assert len(conf.temp_files) == tempfile_count

= Check tdecode with linktype
~ tshark

# These are the same as the ping packets above
pkts = [
  b'\xff\xff\xff\xff\xff\xff\xac"\x0b\xc5j\xdb\x08\x00E\x00\x00\x80\x00\x01\x00\x00@\x01\xf6x\xc0\x00\x02\x01\xc0\x00\x02\x02\x08\x00\xb6\xbe\x00\x00\x00\x00XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX',
  b'\xff\xff\xff\xff\xff\xff\xac"\x0b\xc5j\xdb\x08\x00E\x00\x00\x80\x00\x01\x00\x00@\x01\xf6x\xc0\x00\x02\x02\xc0\x00\x02\x01\x00\x00\xbe\xbe\x00\x00\x00\x00XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX',
]

# tshark doesn't need workarounds on OSX
tempfile_count = len(conf.temp_files)

r = tdecode(pkts, dump=True, linktype=DLT_EN10MB)
assert b'Src: 192.0.2.1' in r
assert b'Src: 192.0.2.2' in r
assert b'Dst: 192.0.2.2' in r
assert b'Dst: 192.0.2.1' in r
assert b'Echo (ping) request' in r
assert b'Echo (ping) reply' in r
assert b'ICMP' in r
assert len(conf.temp_files) == tempfile_count


= Run scapy's tshark command
~ needs_root
tshark(count=1, timeout=3)

= Check wireshark()
~ wireshark

f = BytesIO()
pkt = Ether()/IP()/ICMP()

with mock.patch('subprocess.Popen', return_value=Bunch(stdin=f)) as popen:
    # Prevent closing the BytesIO
    with mock.patch.object(f, 'close'):
        wireshark([pkt])

popen.assert_called_once_with(
    [conf.prog.wireshark, '-ki', '-'],
    stdin=subprocess.PIPE, stdout=None, stderr=None)

print(bytes_hex(f.getvalue()))
assert raw(pkt) in f.getvalue()
f.close()
del f, pkt

= Check Raw IP pcap files

import tempfile
filename = tempfile.mktemp(suffix=".pcap")
wrpcap(filename, [IP()/UDP(), IPv6()/UDP()], linktype=DLT_RAW)
packets = rdpcap(filename)
assert isinstance(packets[0], IP) and isinstance(packets[1], IPv6)

= Check wrpcap() with no packet

import tempfile
filename = tempfile.mktemp(suffix=".pcap")
wrpcap(filename, [])
fstat = os.stat(filename)
assert fstat.st_size != 0
os.remove(filename)

= Check wrpcap() with SndRcvList

import tempfile
filename = tempfile.mktemp(suffix=".pcap")
wrpcap(filename, SndRcvList(res=[(Ether()/IP(), Ether()/IP())]))
assert len(rdpcap(filename)) == 2
os.remove(filename)

= Check wrpcap() with different packets types

import mock
import os
import tempfile

with mock.patch("scapy.utils.warning") as warning:
    filename = tempfile.mktemp()
    wrpcap(filename, [IP(), Ether(), IP(), IP()])
    os.remove(filename)
    assert any("Inconsistent" in arg for arg in warning.call_args[0])

############
############
+ ERF Ethernet format support 

= Variable creations
erffile = BytesIO(b'3;!E_9\x92_\x02\x04\x00p\x00\x00\x00P\x00\x00\x00\x0fS?\xca\xc0\x1cjz\x18\x90\xed\x81\x00\x01:\x08\x00E\x00\x00(\xdf\xab@\x00;\x06\xb3s\n\x01]\xdb\n\xfb9\xda\xc3v\x84\xecD\x16\xb9\xab\xda\xa1b\xf9P\x10f\x98\x18\xcb\x00\x00\x00\x00\x90\x9e\xd7\xd2_\x929_\x0f\x9e\xcd\x1f\x01\x88\xb9\x15[/s<\x01\x88\xb9\x15[/\xcd\x1f\x01\x88\xb9\x15[/0\xcd"E_9\x92_\x02\x04\x00p\x00\x00\x00P\x00\x00\x1cjz\x18\x90\xed\x00\x0fS?\xca\xc0\x08\x00E\x00\x00(\xa2\xdd@\x00@\x06\xebA\n\xfb9\xda\n\x01]\xdb\x84\xec\xc3v\xda\xa1b\xf9D\x16\xb9\xacP\x10\x9a\xf0\xe4q\x00\x00\x00\x00\x00\x00\x00\x00o\xbc\xe2{_\x929_\x0f\x9f+3\x01\x88\xb9\x15u\x1e(^\x01\x88\xb9\x15u\x1e+3\x01\x88\xb9\x15u\x1e')
erffilewithheader = BytesIO(b'4;!E_9\x92_\x82\x00\x00x\x00\x00\x00P\x00\x00\x1a+<M^o\x00\x00\x00\x0fS?\xca\xc0\x1cjz\x18\x90\xed\x81\x00\x01:\x08\x00E\x00\x00(\xdf\xab@\x00;\x06\xb3s\n\x01]\xdb\n\xfb9\xda\xc3v\x84\xecD\x16\xb9\xab\xda\xa1b\xf9P\x10f\x98\x18\xcb\x00\x00\x00\x00\x90\x9e\xd7\xd2_\x929_\x0f\x9e\xcd\x1f\x01\x88\xb9\x15[/s<\x01\x88\xb9\x15[/\xcd\x1f\x01\x88\xb9\x15[/0\xcd"E_9\x92_\x82\x00\x00x\x00\x00\x00P\x00\x00\x1a+<M^o\x00\x00\x1cjz\x18\x90\xed\x00\x0fS?\xca\xc0\x08\x00E\x00\x00(\xa2\xdd@\x00@\x06\xebA\n\xfb9\xda\n\x01]\xdb\x84\xec\xc3v\xda\xa1b\xf9D\x16\xb9\xacP\x10\x9a\xf0\xe4q\x00\x00\x00\x00\x00\x00\x00\x00o\xbc\xe2{_\x929_\x0f\x9f+3\x01\x88\xb9\x15u\x1e(^\x01\x88\xb9\x15u\x1e+3\x01\x88\xb9\x15u\x1e')

= Check reading of ERF Ethernet file
pkterf = rderf(erffile)
assert pkterf[0].time == 1603418463.270038318
assert pkterf[0][IP].src == "10.1.93.219"
assert pkterf[0][IP].dst == "10.251.57.218"
assert pkterf[0][Ether].src == "1c:6a:7a:18:90:ed"
assert pkterf[0][Ether].dst == "00:0f:53:3f:ca:c0"

= Check writing of ERF Ethernet file
import os, tempfile
fdesc, filename = tempfile.mkstemp()
fdesc = os.fdopen(fdesc, "wb")
wrerf(fdesc, pkterf)
fdesc.close()
newpkterf = rderf(filename)

assert pkterf[1][Ether].src == newpkterf[1][Ether].src

assert len(pkterf) == len(newpkterf)
assert newpkterf[0].time is not None
assert newpkterf[0].wirelen is not None
assert newpkterf[0].time == pkterf[0].time
assert newpkterf[0].wirelen == pkterf[0].wirelen
assert newpkterf[1].time is not None
assert newpkterf[1].wirelen is not None
assert newpkterf[1].time == pkterf[1].time
assert newpkterf[1].wirelen == pkterf[1].wirelen

_, filename = tempfile.mkstemp()
wrerf(filename, pkterf, append=True)
wrerf(filename, pkterf, append=True)
newdoublepkterf = rderf(filename)

assert len(newpkterf) * 2 == len(newdoublepkterf)

= Check rderf
pkterf = rderf(erffilewithheader)
assert pkterf[1].time == 1603418463.270062279
assert pkterf[1][Ether].src == "00:0f:53:3f:ca:c0"

############
############
+ Mocked read_routes() calls

= Truncated netstat -rn output on OS X
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@mock.patch("scapy.arch.unix.os")
def test_osx_netstat_truncated(mock_os, mock_get_if_addr):
    """Test read_routes() on OS X 10.? with a long interface name"""
    # netstat & ifconfig outputs from https://github.com/secdev/scapy/pull/119
    netstat_output = u"""
Routing tables

Internet:
Destination        Gateway            Flags        Refs      Use   Netif Expire
default            192.168.1.1        UGSc          460        0     en1
default            link#11            UCSI            1        0 bridge1
127                127.0.0.1          UCS             1        0     lo0
127.0.0.1          127.0.0.1          UH             10  2012351     lo0
"""
    ifconfig_output = u"lo0 en1 bridge10\n"
    # Mocked file descriptors
    def se_popen(command):
        """Perform specific side effects"""
        if command.startswith("netstat -rn"):
            return StringIO(netstat_output)
        elif command == "ifconfig -l":
            ret = StringIO(ifconfig_output)
            def unit():
                return ret
            ret.__call__ = unit
            ret.__enter__ = unit
            ret.__exit__ = lambda x,y,z: None
            return ret
        raise Exception("Command not mocked: %s" % command)
    mock_os.popen.side_effect = se_popen
    # Mocked get_if_addr() behavior
    def se_get_if_addr(iface):
        """Perform specific side effects"""
        if iface == "bridge1":
            return "0.0.0.0"
        return "1.2.3.4"
    mock_get_if_addr.side_effect = se_get_if_addr
    # Test the function
    from scapy.arch.unix import read_routes
    scapy.arch.unix.DARWIN = True
    scapy.arch.unix.FREEBSD = False
    scapy.arch.unix.NETBSD = False
    scapy.arch.unix.OPENBSD = False
    routes = read_routes()
    assert len(routes) == 4
    assert [r for r in routes if r[3] == "bridge10"]


test_osx_netstat_truncated()


= macOS 10.13
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_13_ipv4(mock_os, mock_get_if_addr):
    """Test read_routes() on OS X 10.13"""
    # 'netstat -rn -f inet' output
    netstat_output = u"""
Routing tables

Internet:
Destination        Gateway            Flags        Refs      Use   Netif Expire
default            192.168.28.1       UGSc           82        0     en0
127                127.0.0.1          UCS             0        0     lo0
127.0.0.1          127.0.0.1          UH              1      878     lo0
169.254            link#5             UCS             0        0     en0
192.168.28         link#5             UCS             4        0     en0
192.168.28.1/32    link#5             UCS             2        0     en0
192.168.28.1       88:32:9c:f5:4e:ea  UHLWIir        40       37     en0   1177
192.168.28.2       62:aa:56:4b:51:54  UHLWI           0        0     en0    619
192.168.28.4       38:17:ed:9a:58:28  UHLWIi          1        6     en0    428
192.168.28.18/32   link#5             UCS             1        0     en0
192.168.28.18      88:32:9c:f5:4e:eb  UHLWI           0        1     lo0
192.168.28.28      04:0e:eb:11:74:a7  UHLWI           0        0     en0    576
224.0.0/4          link#5             UmCS            1        0     en0
224.0.0.251        1:0:5e:0:0:fb      UHmLWI          0        0     en0
255.255.255.255/32 link#5             UCS             0        0     en0
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked get_if_addr() output
    def se_get_if_addr(iface):
        """Perform specific side effects"""
        import socket
        if iface == "en0":
            return "192.168.28.18"
        return "127.0.0.1"
    mock_get_if_addr.side_effect = se_get_if_addr
    # Test the function
    from scapy.arch.unix import read_routes
    scapy.arch.unix.DARWIN = False
    scapy.arch.unix.FREEBSD = True
    scapy.arch.unix.NETBSD = False
    scapy.arch.unix.OPENBSD = False
    routes = read_routes()
    for r in routes:
        print(r)
    assert len(routes) == 15
    default_route = [r for r in routes if r[0] == 0][0]
    assert default_route[3] == "en0" and default_route[4] == "192.168.28.18"

test_osx_10_13_ipv4()


= macOS 10.15
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_15_ipv4(mock_os, mock_get_if_addr):
    """Test read_routes() on OS X 10.15"""
    # 'netstat -rn -f inet' output
    netstat_output = u"""
Routing tables

Internet:
Destination        Gateway            Flags        Netif Expire
default            192.168.122.1      UGSc           en0       
127                127.0.0.1          UCS            lo0       
127.0.0.1          127.0.0.1          UH             lo0       
169.254            link#8             UCS            en0      !
192.168.122        link#8             UCS            en0      !
192.168.122.1/32   link#8             UCS            en0      !
192.168.122.1      52:54:0:c0:b7:af   UHLWIir        en0   1169
192.168.122.63/32  link#8             UCS            en0      !
224.0.0/4          link#8             UmCS           en0      !
224.0.0.251        1:0:5e:0:0:fb      UHmLWI         en0       
255.255.255.255/32 link#8             UCS            en0      !
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked get_if_addr() output
    def se_get_if_addr(iface):
        """Perform specific side effects"""
        import socket
        if iface == "en0":
            return "192.168.122.42"
        return "127.0.0.1"
    mock_get_if_addr.side_effect = se_get_if_addr
    # Test the function
    from scapy.arch.unix import read_routes
    scapy.arch.unix.DARWIN = False
    scapy.arch.unix.FREEBSD = True
    scapy.arch.unix.NETBSD = False
    scapy.arch.unix.OPENBSD = False
    routes = read_routes()
    for r in routes:
        print(r)
    assert len(routes) == 11
    default_route = [r for r in routes if r[0] == 0][0]
    assert default_route[3] == "en0" and default_route[4] == "192.168.122.42"

test_osx_10_15_ipv4()


= OpenBSD 6.3
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.get_if_addr")
@mock.patch("scapy.arch.unix.OPENBSD")
@mock.patch("scapy.arch.unix.os")
def test_openbsd_6_3(mock_os, mock_openbsd, mock_get_if_addr):
    """Test read_routes() on OpenBSD 6.3"""
    # 'netstat -rn -f inet' output
    netstat_output = u"""
Routing tables

Internet:
Destination        Gateway            Flags   Refs      Use   Mtu  Prio Iface
default            10.0.1.254         UGS        0        0     -     8 bge0 
224/4              127.0.0.1          URS        0       23 32768     8 lo0  
10.0.1/24          10.0.1.26          UCn        4      192     -     4 bge0 
10.0.1.1           00:30:48:57:ed:0b  UHLc       2      338     -     3 bge0 
10.0.1.2           00:03:ba:0c:0b:52  UHLc       1      186     -     3 bge0 
10.0.1.26          00:30:48:62:b3:f4  UHLl       0    47877     -     1 bge0 
10.0.1.135         link#1             UHLch      1      194     -     3 bge0 
10.0.1.254         link#1             UHLch      1      190     -     3 bge0 
10.0.1.255         10.0.1.26          UHb        0        0     -     1 bge0 
10.188.6/24        10.188.6.17        Cn         0        0     -     4 tap3 
10.188.6.17        fe:e1:ba:d7:ff:32  UHLl       0       25     -     1 tap3 
10.188.6.255       10.188.6.17        Hb         0        0     -     1 tap3 
10.188.135/24      10.0.1.135         UGS        0        0  1350 L   8 bge0 
127/8              127.0.0.1          UGRS       0        0 32768     8 lo0  
127.0.0.1          127.0.0.1          UHhl       1  3835230 32768     1 lo0  
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    
    # Mocked OpenBSD parsing behavior
    mock_openbsd = True
    
    # Mocked get_if_addr() output
    def se_get_if_addr(iface):
        """Perform specific side effects"""
        import socket
        if iface == "bge0":
            return "192.168.122.42"
        return "10.0.1.26"
    mock_get_if_addr.side_effect = se_get_if_addr
    
    # Test the function
    from scapy.arch.unix import read_routes
    return read_routes()

routes = test_openbsd_6_3()

for r in routes:
    print(ltoa(r[0]), ltoa(r[1]), r)
    # check that default route exists in parsed data structure
    if ltoa(r[0]) == "0.0.0.0":
        default = r
    # check that route with locked mtu exists in parsed data structure
    if ltoa(r[0]) == "10.188.135.0":
        locked = r

assert len(routes) == 11
assert default[2] == "10.0.1.254"
assert default[3] == "bge0"
assert locked[2] == "10.0.1.135"
assert locked[3] == "bge0"

= Solaris 11.1
~ mock_read_routes_bsd

import mock
from io import StringIO

# Mocked Solaris 11.1 parsing behavior

@mock.patch("scapy.arch.get_if_addr")
@mock.patch("scapy.arch.unix.SOLARIS", True)
@mock.patch("scapy.arch.unix.os")
def test_solaris_111(mock_os, mock_get_if_addr):
    """Test read_routes() on Solaris 11.1"""
    # 'netstat -rvn -f inet' output
    netstat_output = u"""
IRE Table: IPv4
  Destination             Mask           Gateway          Device  MTU  Ref Flg  Out  In/Fwd 
-------------------- --------------- -------------------- ------ ----- --- --- ----- ------ 
default              0.0.0.0         10.0.2.2             net0    1500   2 UG       5      0 
10.0.2.0             255.255.255.0   10.0.2.15            net0    1500   3 U        0      0 
127.0.0.1            255.255.255.255 127.0.0.1            lo0     8232   2 UH    1517   1517
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    print(scapy.arch.unix.SOLARIS)
    
    # Mocked get_if_addr() output
    def se_get_if_addr(iface):
        """Perform specific side effects"""
        import socket
        if iface == "net0":
            return "10.0.2.15"
        return "127.0.0.1"
    mock_get_if_addr.side_effect = se_get_if_addr
    
    # Test the function
    from scapy.arch.unix import read_routes
    return read_routes()

routes = test_solaris_111()
print(routes)
assert len(routes) == 3
assert routes[0][:4] == (0, 0, '10.0.2.2', 'net0')
assert routes[1][:4] == (167772672, 4294967040, '0.0.0.0', 'net0')
assert routes[2][:4] == (2130706433, 4294967295, '0.0.0.0', 'lo0')


############
############
+ Mocked _parse_tcpreplay_result(stdout, stderr, argv, results_dict)
~ mock_parse_tcpreplay_result

= Test mocked _parse_tcpreplay_result

from scapy.sendrecv import _parse_tcpreplay_result

stdout = """Actual: 1024 packets (198929 bytes) sent in 67.88 seconds.
Rated: 2930.6 bps, 0.02 Mbps, 15.09 pps
Statistics for network device: mon0
        Attempted packets:         1024
        Successful packets:        1024
        Failed packets:            0
        Retried packets (ENOBUFS): 0
        Retried packets (EAGAIN):  0"""

stderr = """Warning in sendpacket.c:sendpacket_open_pf() line 669:
Unsupported physical layer type 0x0323 on mon0.  Maybe it works, maybe it won't.  See tickets #123/318
sending out mon0
processing file: replay-example.pcap"""

argv = ['tcpreplay', '--intf1=mon0', '--multiplier=1.00', '--timer=nano', 'replay-example.pcap']
results_dict = _parse_tcpreplay_result(stdout, stderr, argv)

results_dict

assert results_dict["packets"] == 1024
assert results_dict["bytes"] == 198929
assert results_dict["time"] == 67.88
assert results_dict["bps"] == 2930.6
assert results_dict["mbps"] == 0.02
assert results_dict["pps"] == 15.09
assert results_dict["attempted"] == 1024
assert results_dict["successful"] == 1024
assert results_dict["failed"] == 0
assert results_dict["retried_enobufs"] == 0
assert results_dict["retried_eagain"] == 0
assert results_dict["command"] == " ".join(argv)
assert len(results_dict["warnings"]) == 3

= Test more recent version with flows

data = """Actual: 1 packets (42 bytes) sent in 0.000278 seconds
Rated: 151079.1 Bps, 1.20 Mbps, 3597.12 pps
Flows: 1 flows, 3597.12 fps, 1 flow packets, 0 non-flow
Statistics for network device: enp0s3
        Successful packets:        1
        Failed packets:            0
        Truncated packets:         0
        Retried packets (ENOBUFS): 0
        Retried packets (EAGAIN):  0
"""

results_dict = _parse_tcpreplay_result(data, "", [])
results_dict

expected = {
    'bps': 151079.1,
    'bytes': 42,
    'command': '',
    'failed': 0,
    'flow_packets': 1,
    'flows': 1,
    'fps': 3597.12,
    'mbps': 1.2,
    'non_flow': 0,
    'packets': 1,
    'pps': 3597.12,
    'retried_eagain': 0,
    'retried_enobufs': 0,
    'successful': 1,
    'time': 0.000278,
    'truncated': 0,
    'warnings': []
}

assert results_dict == expected

############
############
+ Mocked read_routes6() calls

= Preliminary definitions
~ mock_read_routes_bsd

import mock
from io import StringIO

def valid_output_read_routes6(routes):
    """"Return True if 'routes' contains correctly formatted entries, False otherwise"""
    for destination, plen, next_hop, dev, cset, me  in routes:
        if not in6_isvalid(destination) or not type(plen) == int:
            return False
        if not in6_isvalid(next_hop) or not isinstance(dev, six.string_types):
            return False
        for address in cset:
            if not in6_isvalid(address):
                return False
    return True

def check_mandatory_ipv6_routes(routes6):
    """Ensure that mandatory IPv6 routes are present"""
    if sum(1 for r in routes6 if r[0] == "::1" and r[4] == ["::1"]) < 1:
        return False
    if sum(1 for r in routes6 if r[0] == "fe80::" and r[1] == 64) < 1:
        return False
    if sum(1 for r in routes6 if in6_islladdr(r[0]) and r[1] == 128 and \
           r[4] == ["::1"]) < 1:
        return False
    return True


= Mac OS X 10.9.5
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_9_5(mock_os, mock_in6_getifaddr):
    """Test read_routes6() on OS X 10.9.5"""
    # 'netstat -rn -f inet6' output
    netstat_output = u"""
Routing tables

Internet6:
Destination                             Gateway                         Flags         Netif Expire
::1                                     ::1                             UHL             lo0
fe80::%lo0/64                           fe80::1%lo0                     UcI             lo0
fe80::1%lo0                             link#1                          UHLI            lo0
fe80::%en0/64                           link#4                          UCI             en0
fe80::ba26:6cff:fe5f:4eee%en0           b8:26:6c:5f:4e:ee               UHLWIi          en0
fe80::bae8:56ff:fe45:8ce6%en0           b8:e8:56:45:8c:e6               UHLI            lo0
ff01::%lo0/32                           ::1                             UmCI            lo0
ff01::%en0/32                           link#4                          UmCI            en0
ff02::%lo0/32                           ::1                             UmCI            lo0
ff02::%en0/32                           link#4                          UmCI            en0
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked in6_getifaddr() output
    mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
                                       ("fe80::ba26:6cff:fe5f:4eee", IPV6_ADDR_LINKLOCAL, "en0")]
    # Test the function
    from scapy.arch.unix import read_routes6
    scapy.arch.unix.DARWIN = False
    scapy.arch.unix.FREEBSD = True
    scapy.arch.unix.NETBSD = False
    scapy.arch.unix.OPENBSD = False
    routes = read_routes6()
    for r in routes:
        print(r)
    assert len(routes) == 6
    assert check_mandatory_ipv6_routes(routes)

test_osx_10_9_5()


= Mac OS X 10.9.5 with global IPv6 connectivity
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_9_5_global(mock_os, mock_in6_getifaddr):
    """Test read_routes6() on OS X 10.9.5 with an IPv6 connectivity"""
    # 'netstat -rn -f inet6' output
    netstat_output = u"""
Routing tables

Internet6:
Destination                             Gateway                         Flags         Netif Expire
default                                 fe80::ba26:8aff:fe5f:4eef%en0   UGc             en0
::1                                     ::1                             UHL             lo0
2a01:ab09:7d:1f01::/64                  link#4                          UC              en0
2a01:ab09:7d:1f01:420:205c:9fab:5be7    b8:e9:55:44:7c:e5               UHL             lo0
2a01:ab09:7d:1f01:ba26:8aff:fe5f:4eef   b8:26:8a:5f:4e:ef               UHLWI           en0
2a01:ab09:7d:1f01:bae9:55ff:fe44:7ce5   b8:e9:55:44:7c:e5               UHL             lo0
fe80::%lo0/64                           fe80::1%lo0                     UcI             lo0
fe80::1%lo0                             link#1                          UHLI            lo0
fe80::%en0/64                           link#4                          UCI             en0
fe80::5664:d9ff:fe79:4e00%en0           54:64:d9:79:4e:0                UHLWI           en0
fe80::6ead:f8ff:fe74:945a%en0           6c:ad:f8:74:94:5a               UHLWI           en0
fe80::a2f3:c1ff:fec4:5b50%en0           a0:f3:c1:c4:5b:50               UHLWI           en0
fe80::ba26:8aff:fe5f:4eef%en0           b8:26:8a:5f:4e:ef               UHLWIir         en0
fe80::bae9:55ff:fe44:7ce5%en0           b8:e9:55:44:7c:e5               UHLI            lo0
ff01::%lo0/32                           ::1                             UmCI            lo0
ff01::%en0/32                           link#4                          UmCI            en0
ff02::%lo0/32                           ::1                             UmCI            lo
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked in6_getifaddr() output
    mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
                                       ("fe80::ba26:6cff:fe5f:4eee", IPV6_ADDR_LINKLOCAL, "en0")]
    # Test the function
    from scapy.arch.unix import read_routes6
    routes = read_routes6()
    print(routes)
    assert valid_output_read_routes6(routes)
    for r in routes:
        print(r)
    assert len(routes) == 11
    assert check_mandatory_ipv6_routes(routes)

test_osx_10_9_5_global()


= Mac OS X 10.10.4
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_osx_10_10_4(mock_os, mock_in6_getifaddr):
    """Test read_routes6() on OS X 10.10.4"""
    # 'netstat -rn -f inet6' output
    netstat_output = u"""
Routing tables

Internet6:
Destination                             Gateway                         Flags         Netif Expire
::1                                     ::1                             UHL             lo0
fe80::%lo0/64                           fe80::1%lo0                     UcI             lo0
fe80::1%lo0                             link#1                          UHLI            lo0
fe80::%en0/64                           link#4                          UCI             en0
fe80::a00:27ff:fe9b:c965%en0            8:0:27:9b:c9:65                 UHLI            lo0
ff01::%lo0/32                           ::1                             UmCI            lo0
ff01::%en0/32                           link#4                          UmCI            en0
ff02::%lo0/32                           ::1                             UmCI            lo0
ff02::%en0/32                           link#4                          UmCI            en0
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked in6_getifaddr() output
    mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
                                       ("fe80::a00:27ff:fe9b:c965", IPV6_ADDR_LINKLOCAL, "en0")]
    # Test the function
    from scapy.arch.unix import read_routes6
    routes = read_routes6()
    for r in routes:
        print(r)
    assert len(routes) == 5
    assert check_mandatory_ipv6_routes(routes)

test_osx_10_10_4()


= FreeBSD 10.2
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_freebsd_10_2(mock_os, mock_in6_getifaddr):
    """Test read_routes6() on FreeBSD 10.2"""
    # 'netstat -rn -f inet6' output
    netstat_output = u"""
Routing tables

Internet6:
Destination                       Gateway                       Flags      Netif Expire
::/96                             ::1                           UGRS        lo0
::1                               link#2                        UH          lo0
::ffff:0.0.0.0/96                 ::1                           UGRS        lo0
fe80::/10                         ::1                           UGRS        lo0
fe80::%lo0/64                     link#2                        U           lo0
fe80::1%lo0                       link#2                        UHS         lo0
ff01::%lo0/32                     ::1                           U           lo0
ff02::/16                         ::1                           UGRS        lo0
ff02::%lo0/32                     ::1                           U           lo0
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked in6_getifaddr() output
    mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0")]
    # Test the function
    from scapy.arch.unix import read_routes6
    routes = read_routes6()
    scapy.arch.unix.DARWIN = False
    scapy.arch.unix.FREEBSD = True
    scapy.arch.unix.NETBSD = False
    scapy.arch.unix.OPENBSD = False
    for r in routes:
        print(r)
    assert len(routes) == 3
    assert check_mandatory_ipv6_routes(routes)

test_freebsd_10_2()


= FreeBSD 13.0
~ mock_read_routes_bsd

import mock
from io import StringIO
 
@mock.patch("scapy.arch.get_if_addr")
@mock.patch("scapy.arch.unix.os")
def test_freebsd_13(mock_os, mock_get_if_addr):
    """Test read_routes() on FreeBSD 13"""
    # 'netstat -rnW -f inet' output
    netstat_output = u"""
Routing tables

Internet:
Destination        Gateway            Flags   Nhop#    Mtu      Netif Expire
default            10.0.0.1           UGS         3   1500     vtnet0
10.0.0.0/24        link#1             U           2   1500     vtnet0
10.0.0.8           link#2             UHS         1  16384        lo0
127.0.0.1          link#2             UH          1  16384        lo0
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked get_if_addr() behavior
    def se_get_if_addr(iface):
        """Perform specific side effects"""
        if iface == "vtnet0":
            return "10.0.0.1"
        return "1.2.3.4"
    mock_get_if_addr.side_effect = se_get_if_addr
    # Test the function
    from scapy.arch.unix import read_routes
    routes = read_routes()
    scapy.arch.unix.DARWIN = False
    scapy.arch.unix.FREEBSD = True
    scapy.arch.unix.NETBSD = False
    scapy.arch.unix.OPENBSD = False
    for r in routes:
        print(r)
        assert r[3] in ["vtnet0", "lo0"]
    assert len(routes) == 4

test_freebsd_13()


= OpenBSD 5.5
~ mock_read_routes_bsd

import mock
from io import StringIO

@mock.patch("scapy.arch.unix.OPENBSD")
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_openbsd_5_5(mock_os, mock_in6_getifaddr, mock_openbsd):
    """Test read_routes6() on OpenBSD 5.5"""
    # 'netstat -rn -f inet6' output
    netstat_output = u"""
Routing tables

Internet6:
Destination                        Gateway                        Flags   Refs      Use   Mtu  Prio Iface
::/104                             ::1                            UGRS       0        0     -     8 lo0  
::/96                              ::1                            UGRS       0        0     -     8 lo0  
::1                                ::1                            UH        14        0 33144     4 lo0  
::127.0.0.0/104                    ::1                            UGRS       0        0     -     8 lo0  
::224.0.0.0/100                    ::1                            UGRS       0        0     -     8 lo0  
::255.0.0.0/104                    ::1                            UGRS       0        0     -     8 lo0  
::ffff:0.0.0.0/96                  ::1                            UGRS       0        0     -     8 lo0  
2002::/24                          ::1                            UGRS       0        0     -     8 lo0  
2002:7f00::/24                     ::1                            UGRS       0        0     -     8 lo0  
2002:e000::/20                     ::1                            UGRS       0        0     -     8 lo0  
2002:ff00::/24                     ::1                            UGRS       0        0     -     8 lo0  
fe80::/10                          ::1                            UGRS       0        0     -     8 lo0  
fe80::%em0/64                      link#1                         UC         0        0     -     4 em0  
fe80::a00:27ff:fe04:59bf%em0       08:00:27:04:59:bf              UHL        0        0     -     4 lo0  
fe80::%lo0/64                      fe80::1%lo0                    U          0        0     -     4 lo0  
fe80::1%lo0                        link#3                         UHL        0        0     -     4 lo0  
fec0::/10                          ::1                            UGRS       0        0     -     8 lo0  
ff01::/16                          ::1                            UGRS       0        0     -     8 lo0  
ff01::%em0/32                      link#1                         UC         0        0     -     4 em0  
ff01::%lo0/32                      fe80::1%lo0                    UC         0        0     -     4 lo0  
ff02::/16                          ::1                            UGRS       0        0     -     8 lo0  
ff02::%em0/32                      link#1                         UC         0        0     -     4 em0  
ff02::%lo0/32                      fe80::1%lo0                    UC         0        0     -     4 lo0 
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    
    # Mocked in6_getifaddr() output
    mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
                                       ("fe80::a00:27ff:fe04:59bf", IPV6_ADDR_LINKLOCAL, "em0")]
    # Mocked OpenBSD parsing behavior
    mock_openbsd = True
    # Test the function
    from scapy.arch.unix import read_routes6
    routes = read_routes6()
    for r in routes:
        print(r)
    assert len(routes) == 5
    assert check_mandatory_ipv6_routes(routes)

test_openbsd_5_5()


= NetBSD 7.0
~ mock_read_routes_bsd

@mock.patch("scapy.arch.unix.NETBSD")
@mock.patch("scapy.arch.unix.in6_getifaddr")
@mock.patch("scapy.arch.unix.os")
def test_netbsd_7_0(mock_os, mock_in6_getifaddr, mock_netbsd):
    """Test read_routes6() on NetBSD 7.0"""
    # 'netstat -rn -f inet6' output
    netstat_output = u"""
Routing tables

Internet6:
Destination                        Gateway                        Flags    Refs      Use    Mtu Interface
::/104                             ::1                            UGRS        -        -      -  lo0
::/96                              ::1                            UGRS        -        -      -  lo0
::1                                ::1                            UH          -        -  33648  lo0
::127.0.0.0/104                    ::1                            UGRS        -        -      -  lo0
::224.0.0.0/100                    ::1                            UGRS        -        -      -  lo0
::255.0.0.0/104                    ::1                            UGRS        -        -      -  lo0
::ffff:0.0.0.0/96                  ::1                            UGRS        -        -      -  lo0
2001:db8::/32                      ::1                            UGRS        -        -      -  lo0
2002::/24                          ::1                            UGRS        -        -      -  lo0
2002:7f00::/24                     ::1                            UGRS        -        -      -  lo0
2002:e000::/20                     ::1                            UGRS        -        -      -  lo0
2002:ff00::/24                     ::1                            UGRS        -        -      -  lo0
fe80::/10                          ::1                            UGRS        -        -      -  lo0
fe80::%wm0/64                      link#1                         UC          -        -      -  wm0
fe80::acd1:3989:180e:fde0          08:00:27:a1:64:d8              UHL         -        -      -  lo0
fe80::%lo0/64                      fe80::1                        U           -        -      -  lo0
fe80::1                            link#2                         UHL         -        -      -  lo0
ff01:1::/32                        link#1                         UC          -        -      -  wm0
ff01:2::/32                        ::1                            UC          -        -      -  lo0
ff02::%wm0/32                      link#1                         UC          -        -      -  wm0
ff02::%lo0/32                      ::1                            UC          -        -      -  lo0
"""
    # Mocked file descriptor
    strio = StringIO(netstat_output)
    mock_os.popen = mock.MagicMock(return_value=strio)
    # Mocked in6_getifaddr() output
    mock_in6_getifaddr.return_value = [("::1", IPV6_ADDR_LOOPBACK, "lo0"),
                                       ("fe80::acd1:3989:180e:fde0", IPV6_ADDR_LINKLOCAL, "wm0")]
    # Test the function
    from scapy.arch.unix import read_routes6
    routes = read_routes6()
    for r in routes:
        print(r)
    assert len(routes) == 5
    assert check_mandatory_ipv6_routes(routes)

test_netbsd_7_0()


############
############
+ Mocked route() calls

= Mocked IPv4 routes calls

import scapy

IFACES._add_fake_iface("enp3s0")
IFACES._add_fake_iface("lo")

old_routes = conf.route.routes
old_iface = conf.iface
old_loopback = conf.loopback_name
try:
    conf.iface = 'enp3s0'
    conf.loopback_name = 'lo'
    conf.route.invalidate_cache()
    conf.route.routes = [
        (4294967295, 4294967295, '0.0.0.0', 'wlan0', '', 281),
        (4294967295, 4294967295, '0.0.0.0', 'lo', '', 291),
        (4294967295, 4294967295, '0.0.0.0', 'enp3s0', '192.168.0.119', 281),
        (3758096384, 4026531840, '0.0.0.0', 'lo', '', 291),
        (3758096384, 4026531840, '0.0.0.0', 'wlan0', '', 281),
        (3758096384, 4026531840, '0.0.0.0', 'enp3s0', '1.1.1.1', 281),
        (3232235775, 4294967295, '0.0.0.0', 'enp3s0', '2.2.2.2', 281),
        (3232235639, 4294967295, '0.0.0.0', 'enp3s0', '3.3.3.3', 281),
        (3232235520, 4294967040, '0.0.0.0', 'enp3s0', '4.4.4.4', 281),
        (0, 0, '192.168.0.254', 'enp3s0', '192.168.0.119', 25)
    ]
    assert conf.route.route("192.168.0.0-10") == ('enp3s0', '4.4.4.4', '0.0.0.0')
    assert conf.route.route("192.168.0.119") == ('lo', '192.168.0.119', '0.0.0.0')
    assert conf.route.route("224.0.0.0") == ('enp3s0', '1.1.1.1', '0.0.0.0')
    assert conf.route.route("255.255.255.255") == ('enp3s0', '192.168.0.119', '0.0.0.0')
    assert conf.route.route("*") == ('enp3s0', '192.168.0.119', '192.168.0.254')
finally:
    conf.loopback_name = old_loopback
    conf.iface = old_iface
    conf.route.routes = old_routes
    conf.route.invalidate_cache()
    IFACES.reload()


= Mocked IPv6 routes calls

IFACES._add_fake_iface("enp3s0")
IFACES._add_fake_iface("lo")

old_routes = conf.route6.routes
old_iface = conf.iface
old_loopback = conf.loopback_name
try:
    conf.route6.ipv6_ifaces = set(['enp3s0', 'wlan0', 'lo'])
    conf.iface = 'enp3s0'
    conf.loopback_name = 'lo'
    conf.route6.invalidate_cache()
    conf.route6.routes = [
        ('fe80::dd17:1fa6:a123:ab4', 128, '::', 'lo', ['fe80::dd17:1fa6:a123:ab4'], 291),
        ('fe80::7101:5678:1234:da65', 128, '::', 'enp3s0', ['fe80::7101:5678:1234:da65'], 281),
        ('fe80::1f:ae12:4d2c:abff', 128, '::', 'wlan0', ['fe80::1f:ae12:4d2c:abff'], 281),
        ('fe80::', 64, '::', 'wlan0', ['fe80::1f:ae12:4d2c:abff'], 281),
        ('fe80::', 64, '::', 'lo', ['fe80::dd17:1fa6:a123:ab4'], 291),
        ('fe80::', 64, '::', 'enp3s0', ['fe80::7101:5678:1234:da65'], 281),
        ('2a01:e35:1e06:ab56:7010:6548:9646:fa77', 128, '::', 'enp3s0', ['2a01:e35:1e06:ab56:7010:6548:9646:fa77', '2a01:e35:1e06:ab56:512:8bb7:8ab8:14a8'], 281),
        ('2a01:e35:1e06:ab56:512:8bb7:8ab8:14a8', 128, '::', 'enp3s0', ['2a01:e35:1e06:ab56:7010:6548:9646:fa77', '2a01:e35:1e06:ab56:512:8bb7:8ab8:14a8'], 281),
        ('2a01:e35:1e06:ab56::', 64, '::', 'enp3s0', ['2a01:e35:1e06:ab56:7010:6548:9646:fa77', '2a01:e35:1e06:ab56:512:8bb7:8ab8:14a8'], 281),
        ('::', 0, 'fe80::160c:64aa:ef6f:fe14', 'enp3s0', ['2a01:e35:1e06:ab56:7010:6548:9646:fa77', '2a01:e35:1e06:ab56:512:8bb7:8ab8:14a8'], 281)
    ]
    assert conf.route6.route("2a01:e35:1e06:ab56:512:8bb7:8ab8:14a8") == ('enp3s0', '2a01:e35:1e06:ab56:7010:6548:9646:fa77', '::')
    assert conf.route6.route("::1") == ('enp3s0', '2a01:e35:1e06:ab56:7010:6548:9646:fa77', 'fe80::160c:64aa:ef6f:fe14')
    assert conf.route6.route("ff02::1") == ('enp3s0', 'fe80::7101:5678:1234:da65', '::')
    assert conf.route6.route("fe80::1") == ('enp3s0', 'fe80::7101:5678:1234:da65', '::')
    assert conf.route6.route("fe80::1", dev='lo') == ('lo', 'fe80::dd17:1fa6:a123:ab4', '::')
finally:
    conf.loopback_name = old_loopback
    conf.iface = old_iface
    conf.route6.resync()

= Find a link-local address when conf.iface does not support IPv6

old_iface = conf.iface
conf.route6.ipv6_ifaces = set(['eth1', 'lo'])
conf.iface = "eth0"
conf.route6.routes = [("fe80::", 64, "::", "eth1", ["fe80::a00:28ff:fe07:1980"], 256), ("::1", 128, "::", "lo", ["::1"], 0), ("fe80::a00:28ff:fe07:1980", 128, "::", "lo", ["::1"], 0)]
assert conf.route6.route("fe80::2807") == ("eth1", "fe80::a00:28ff:fe07:1980", "::")
conf.iface = old_iface
conf.route6.resync()

= Windows: reset routes properly

if WINDOWS:
    from scapy.arch.windows import _route_add_loopback
    _route_add_loopback()


############
############
############
+ Tests of StreamSocket

= Test with DNS over TCP
~ netaccess

import socket
sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sck.connect(("8.8.8.8", 53))

class DNSTCP(Packet):
    name = "DNS over TCP"
    fields_desc = [ FieldLenField("len", None, fmt="!H", length_of="dns"),
                    PacketLenField("dns", 0, DNS, length_from=lambda p: p.len)]

ssck = StreamSocket(sck)
ssck.basecls = DNSTCP

r = ssck.sr1(DNSTCP(dns=DNS(rd=1, qd=DNSQR(qname="www.example.com"))), timeout=3)
sck.close()
assert DNSTCP in r and len(r.dns.an)

############
+ Tests of SSLStreamContext

= Test with recv() calls that return exact packet-length rawings
~ sslraweamsocket

import socket
class MockSocket(object):
    def __init__(self):
        self.l = [ b'\x00\x00\x00\x01', b'\x00\x00\x00\x02', b'\x00\x00\x00\x03' ]
    def recv(self, x):
        if len(self.l) == 0:
            raise socket.error(100, 'EOF')
        return self.l.pop(0)
    def fileno(self):
        return -1
    def close(self):
        return


class TestPacket(Packet):
    name = 'TestPacket'
    fields_desc = [
        IntField('data', 0)
    ]
    def guess_payload_class(self, p):
        return conf.padding_layer

s = MockSocket()
ss = SSLStreamSocket(s, basecls=TestPacket)

p = ss.recv()
assert p.data == 1
p = ss.recv()
assert p.data == 2
p = ss.recv()
assert p.data == 3
try:
    ss.recv()
    ret = False
except socket.error:
    ret = True

assert ret

= Test with recv() calls that return twice as much data as the exact packet-length
~ sslraweamsocket

import socket
class MockSocket(object):
    def __init__(self):
        self.l = [ b'\x00\x00\x00\x01\x00\x00\x00\x02', b'\x00\x00\x00\x03\x00\x00\x00\x04' ]
    def recv(self, x):
        if len(self.l) == 0:
            raise socket.error(100, 'EOF')
        return self.l.pop(0)
    def fileno(self):
        return -1
    def close(self):
        return


class TestPacket(Packet):
    name = 'TestPacket'
    fields_desc = [
        IntField('data', 0)
    ]
    def guess_payload_class(self, p):
        return conf.padding_layer

s = MockSocket()
ss = SSLStreamSocket(s, basecls=TestPacket)

p = ss.recv()
assert p.data == 1
p = ss.recv()
assert p.data == 2
p = ss.recv()
assert p.data == 3
p = ss.recv()
assert p.data == 4
try:
    ss.recv()
    ret = False
except socket.error:
    ret = True

assert ret

= Test with recv() calls that return not enough data
~ sslraweamsocket

import socket
class MockSocket(object):
    def __init__(self):
        self.l = [ b'\x00\x00', b'\x00\x01', b'\x00\x00\x00', b'\x02', b'\x00\x00', b'\x00', b'\x03' ]
    def recv(self, x):
        if len(self.l) == 0:
            raise socket.error(100, 'EOF')
        return self.l.pop(0)
    def fileno(self):
        return -1
    def close(self):
        return


class TestPacket(Packet):
    name = 'TestPacket'
    fields_desc = [
        IntField('data', 0)
    ]
    def guess_payload_class(self, p):
        return conf.padding_layer

s = MockSocket()
ss = SSLStreamSocket(s, basecls=TestPacket)

try:
    p = ss.recv()
    ret = False
except:
    ret = True

assert ret
p = ss.recv()
assert p.data == 1
try:
    p = ss.recv()
    ret = False
except:
    ret = True

assert ret
p = ss.recv()
assert p.data == 2
try:
    p = ss.recv()
    ret = False
except:
    ret = True

assert ret
try:
    p = ss.recv()
    ret = False
except:
    ret = True

assert ret
p = ss.recv()
assert p.data == 3


############
############
+ Test correct conversion from binary to rawing of IPv6 addresses

= IPv6 bin to rawing conversion
from scapy.pton_ntop import _inet6_ntop, inet_ntop
import socket
for binfrm, address in [
        (b'\x00' * 16, '::'),
        (b'\x11\x11\x22\x22\x33\x33\x44\x44\x55\x55\x66\x66\x77\x77\x88\x88',
         '1111:2222:3333:4444:5555:6666:7777:8888'),
        (b'\x11\x11\x22\x22\x33\x33\x44\x44\x55\x55\x00\x00\x00\x00\x00\x00',
         '1111:2222:3333:4444:5555::'),
        (b'\x00\x00\x00\x00\x00\x00\x44\x44\x55\x55\x66\x66\x77\x77\x88\x88',
         '::4444:5555:6666:7777:8888'),
        (b'\x00\x00\x00\x00\x33\x33\x44\x44\x00\x00\x00\x00\x00\x00\x88\x88',
         '0:0:3333:4444::8888'),
        (b'\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
         '1::'),
        (b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01',
         '::1'),
        (b'\x11\x11\x00\x00\x00\x00\x44\x44\x00\x00\x00\x00\x77\x77\x88\x88',
         '1111::4444:0:0:7777:8888'),
        (b'\x10\x00\x02\x00\x00\x30\x00\x04\x00\x05\x00\x60\x07\x00\x80\x00',
         '1000:200:30:4:5:60:700:8000'),
]:
    addr1 = inet_ntop(socket.AF_INET6, binfrm)
    addr2 = _inet6_ntop(binfrm)
    assert address == addr1 == addr2

= IPv6 bin to rawing conversion - Zero-block of length 1
binfrm = b'\x11\x11\x22\x22\x33\x33\x44\x44\x55\x55\x66\x66\x00\x00\x88\x88'
addr1, addr2 = inet_ntop(socket.AF_INET6, binfrm), _inet6_ntop(binfrm)
# On Mac OS socket.inet_ntop is not fully compliant with RFC 5952 and
# shortens the single zero block to '::'. This is a valid IPv6 address
# representation anyway.
assert(addr1 in ['1111:2222:3333:4444:5555:6666:0:8888',
                 '1111:2222:3333:4444:5555:6666::8888'])
assert addr2 == '1111:2222:3333:4444:5555:6666:0:8888'

= IPv6 bin to rawing conversion - Illegal sizes
for binfrm in ["\x00" * 15, b"\x00" * 17]:
    rc = False
    try:
        inet_ntop(socket.AF_INET6, binfrm)
    except Exception as exc1:
        _exc1 = exc1
        rc = True
    assert rc
    try:
        _inet6_ntop(binfrm)
    except Exception as exc2:
        rc = isinstance(exc2, type(_exc1))
    assert rc


############
############
+ Addresses generators

= Net

assert list(Net("192.168.0.0/31")) == ["192.168.0.0", "192.168.0.1"]

assert "1.2.3.4" in Net("0.0.0.0/0")

assert "192.168.0.0/25" in Net("192.168.0.0/24")

assert "192.168.0.0/23" not in Net("192.168.0.0/24")

assert "0.0.0.0/1" in Net("0.0.0.0/0")

assert "0.0.0.0/0" not in Net("0.0.0.0/1")

assert Net("1.2.3.0/24") == Net("1.2.3.0", "1.2.3.255")

assert hash(Net("1.2.3.0/24")) == hash(Net("1.2.3.0", "1.2.3.255"))

= Net using name
~ netaccess

ip = IP(dst="www.google.com")
n1 = ip.dst
assert isinstance(n1, Net)
ip.show()

= Multiple IP addresses test
~ netaccess

ip = IP(dst=['192.168.0.1', 'www.google.fr'],ihl=(1,5))
assert ip.dst[0] == '192.168.0.1'
assert isinstance(ip.dst[1], Net)
src = ip.src
assert src
assert isinstance(src, str)

= OID

oid = OID("1.2.3.4.5.6-8")
sum(1 for o in oid) == 3
assert oid.__iterlen__() == 3

= Net6

n1 = Net6("2001:db8::/127")
assert len(list(n1)) == 2
assert len(n1) == 2

n2 = Net6("fec0::/110")
assert len(n2) == 262144

assert "ffff::ffff" in Net6("::/0")

assert "::/1" in Net6("::/0")

assert "::/0" not in Net6("::/1")

assert Net6("::/120") == Net6("::", "::ff")

assert hash(Net6("::/120")) == hash(Net6("::", "::ff"))

assert Net6("::1.2.3.0/120") == Net6("::1.2.3.0", "::1.2.3.255")

assert hash(Net6("::1.2.3.0/120")) == hash(Net6("::1.2.3.0", "::1.2.3.255"))

assert Net6("::1.2.3.0/120") != Net("1.2.3.0/24")

assert hash(Net6("::1.2.3.0/120")) != hash(Net("1.2.3.0/24"))

= Net6 using web address
~ netaccess ipv6

ip = IPv6(dst="www.google.com")
n1 = ip.dst
assert isinstance(n1, Net6)
assert "www.google.com" in repr(n1)
ip.show()

ip = IPv6(dst="www.yahoo.com")
assert IPv6(raw(ip)).dst == [p.dst for p in ip][0]

= Multiple IPv6 addresses test
~ netaccess ipv6

ip = IPv6(dst=['2001:db8::1', 'www.google.fr'],hlim=(1,5))
assert ip.dst[0] == '2001:db8::1'
assert isinstance(ip.dst[1], Net6)
src = ip.src
assert src
assert isinstance(src, str)

= Test repr on Net
~ netaccess

conf.color_theme = BlackAndWhite()
output = repr(IP(src="www.google.com"))
assert 'Net("www.google.com/32")' in output

= Test repr on Net
~ netaccess ipv6

conf.color_theme = BlackAndWhite()
assert 'Net6("www.google.com/128")' in repr(IPv6(src="www.google.com"))

############
############
+ IPv6 helpers

= in6_getLocalUniquePrefix()

p = in6_getLocalUniquePrefix()
len(inet_pton(socket.AF_INET6, p)) == 16 and p.startswith("fd")

= Misc addresses manipulation functions

teredoAddrExtractInfo("2001:0:0a0b:0c0d:0028:f508:f508:08f5") == ("10.11.12.13", 40, "10.247.247.10", 2807)

ip6 = IP6Field("test", None)
ip6.i2repr("", "2001:0:0a0b:0c0d:0028:f508:f508:08f5") == "2001:0:0a0b:0c0d:0028:f508:f508:08f5 [Teredo srv: 10.11.12.13 cli: 10.247.247.10:2807]"
ip6.i2repr("", "2002:0102:0304::1") == "2002:0102:0304::1 [6to4 GW: 1.2.3.4]"

in6_iseui64("fe80::bae8:58ff:fed4:e5f6") == True

in6_isanycast("2001:db8::fdff:ffff:ffff:ff80") == True

a = inet_pton(socket.AF_INET6, "2001:db8::2807")
in6_xor(a, a) == b"\x00" * 16

a = inet_pton(socket.AF_INET6, "fe80::bae8:58ff:fed4:e5f6")
r = inet_ntop(socket.AF_INET6, in6_getnsma(a)) 
r == "ff02::1:ffd4:e5f6"

in6_isllsnmaddr(r) == True

in6_isdocaddr("2001:db8::2807") == True

in6_isaddrllallnodes("ff02::1") == True

in6_isaddrllallservers("ff02::2") == True

= in6_getscope()

assert in6_getscope("2001:db8::2807") == IPV6_ADDR_GLOBAL
assert in6_getscope("fec0::2807") == IPV6_ADDR_SITELOCAL
assert in6_getscope("fe80::2807") == IPV6_ADDR_LINKLOCAL
assert in6_getscope("ff02::2807") == IPV6_ADDR_LINKLOCAL
assert in6_getscope("ff0e::2807") == IPV6_ADDR_GLOBAL
assert in6_getscope("ff05::2807") == IPV6_ADDR_SITELOCAL
assert in6_getscope("ff01::2807") == IPV6_ADDR_LOOPBACK
assert in6_getscope("::1") == IPV6_ADDR_LOOPBACK

= construct_source_candidate_set()

dev_addresses = [('fe80::', IPV6_ADDR_LINKLOCAL, "linklocal"),('fec0::', IPV6_ADDR_SITELOCAL, "sitelocal"),('ff0e::', IPV6_ADDR_GLOBAL, "global")]

assert construct_source_candidate_set("2001:db8::2807", 0, dev_addresses) == ["ff0e::"]
assert construct_source_candidate_set("fec0::2807", 0, dev_addresses) == ["fec0::"]
assert construct_source_candidate_set("fe80::2807", 0, dev_addresses) == ["fe80::"]
assert construct_source_candidate_set("ff02::2807", 0, dev_addresses) == ["fe80::"]
assert construct_source_candidate_set("ff0e::2807", 0, dev_addresses) == ["ff0e::"]
assert construct_source_candidate_set("ff05::2807", 0, dev_addresses) == ["fec0::"]
assert construct_source_candidate_set("ff01::2807", 0, dev_addresses) == ["::1"]
assert construct_source_candidate_set("::", 0, dev_addresses) == ["ff0e::"]

= inet_pton()

from scapy.pton_ntop import _inet6_pton, inet_pton
import socket

ip6_bad_addrs = ["fe80::2e67:ef2d:7eca::ed8a",
                 "fe80:1234:abcd::192.168.40.12:abcd",
                 "fe80:1234:abcd::192.168.40",
                 "fe80:1234:abcd::192.168.400.12",
                 "1234:5678:9abc:def0:1234:5678:9abc:def0:",
                 "1234:5678:9abc:def0:1234:5678:9abc:def0:1234"]
for ip6 in ip6_bad_addrs:
    rc = False
    exc1 = None
    try:
        res1 = inet_pton(socket.AF_INET6, ip6)
    except Exception as e:
        rc = True
        exc1 = e
    assert rc
    rc = False
    try:
        res2 = _inet6_pton(ip6)
    except Exception as exc2:
        rc = isinstance(exc2, type(exc1))
    assert rc

ip6_good_addrs = [("fe80:1234:abcd::192.168.40.12",
                   b'\xfe\x80\x124\xab\xcd\x00\x00\x00\x00\x00\x00\xc0\xa8(\x0c'),
                  ("fe80:1234:abcd::fe06",
                   b'\xfe\x80\x124\xab\xcd\x00\x00\x00\x00\x00\x00\x00\x00\xfe\x06'),
                  ("fe80::2e67:ef2d:7ece:ed8a",
                   b'\xfe\x80\x00\x00\x00\x00\x00\x00.g\xef-~\xce\xed\x8a'),
                  ("::ffff",
                   b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff'),
                  ("ffff::",
                   b'\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'),
                  ('::', b'\x00' * 16)]
for ip6, res in ip6_good_addrs:
    res1 = inet_pton(socket.AF_INET6, ip6)
    res2 = _inet6_pton(ip6)
    assert res == res1 == res2


############
############
+ Test Route class

= make_route()

r4 = Route()
tmp_route = r4.make_route(host="10.12.13.14")
(tmp_route[0], tmp_route[1], tmp_route[2]) == (168561934, 4294967295, '0.0.0.0')

tmp_route = r4.make_route(net="10.12.13.0/24")
(tmp_route[0], tmp_route[1], tmp_route[2]) == (168561920, 4294967040, '0.0.0.0')

= add() & delt()

r4 = Route()
len_r4 = len(r4.routes)
r4.add(net="192.168.1.0/24", gw="1.2.3.4")
len(r4.routes) == len_r4 + 1
r4.delt(net="192.168.1.0/24", gw="1.2.3.4")
len(r4.routes) == len_r4

= ifchange()

r4.add(net="192.168.1.0/24", gw="1.2.3.4", dev=get_dummy_interface())
r4.ifchange(get_dummy_interface(), "5.6.7.8")
r4.routes[-1][4] == "5.6.7.8"

= ifdel()

r4.ifdel(get_dummy_interface())
len(r4.routes) == len_r4

= ifadd() & get_if_bcast()

r4 = Route()
len_r4 = len(r4.routes)

r4.ifadd(get_dummy_interface(), "1.2.3.4/24")
len(r4.routes) == len_r4 +1 

r4.get_if_bcast(get_dummy_interface()) == "1.2.3.255"

r4.ifdel(get_dummy_interface())
len(r4.routes) == len_r4

dummy_interface = get_dummy_interface()

bck_conf_route_routes = conf.route.routes
conf.route.routes = [
    (0, 0, '172.21.230.1', dummy_interface, '172.21.230.10', 1),                # 0.0.0.0         / 0.0.0.0         == 255.255.255.255
    (2851995648, 4294901760, '0.0.0.0', dummy_interface, '172.21.230.10', 1),   # 169.254.0.0     / 255.255.0.0     == 169.254.255.255
    (2887116288, 4294967040, '0.0.0.0', dummy_interface, '172.21.230.10', 1),   # 172.21.230.0    / 255.255.255.0   == 172.21.230.255
    (2887116289, 4294967295, '0.0.0.0', dummy_interface, '172.21.230.10', 1),   # 172.21.230.1    / 255.255.255.255 == 172.21.230.1
    (3758096384, 4026531840, '0.0.0.0', dummy_interface, '172.21.230.10', 1),   # 224.0.0.0       / 240.0.0.0       == 239.255.255.255
    (3758096635, 4294967295, '0.0.0.0', dummy_interface, '172.21.230.10', 1),   # 224.0.0.251     / 255.255.255.255 == 224.0.0.251
    (4294967295, 4294967295, '0.0.0.0', dummy_interface, '172.21.230.10', 1),   # 255.255.255.255 / 255.255.255.255 == 255.255.255.255
    ]

assert sorted(conf.route.get_if_bcast(dummy_interface)) == sorted(['169.254.255.255', '172.21.230.255', '239.255.255.255'])
conf.route.routes = bck_conf_route_routes


############
############
+ Flags

= IP flags
~ IP

pkt = IP(flags="MF")
assert pkt.flags.MF
assert not pkt.flags.DF
assert not pkt.flags.evil
assert repr(pkt.flags) == '<Flag 1 (MF)>'
pkt.flags.MF = 0
pkt.flags.DF = 1
assert not pkt.flags.MF
assert pkt.flags.DF
assert not pkt.flags.evil
assert repr(pkt.flags) == '<Flag 2 (DF)>'
pkt.flags |= 'evil+MF'
pkt.flags &= 'DF+MF'
assert pkt.flags.MF
assert pkt.flags.DF
assert not pkt.flags.evil
assert repr(pkt.flags) == '<Flag 3 (MF+DF)>'

pkt = IP(flags=3)
assert pkt.flags.MF
assert pkt.flags.DF
assert not pkt.flags.evil
assert repr(pkt.flags) == '<Flag 3 (MF+DF)>'
pkt.flags = 6
assert not pkt.flags.MF
assert pkt.flags.DF
assert pkt.flags.evil
assert repr(pkt.flags) == '<Flag 6 (DF+evil)>'

assert len({IP().flags, IP().flags}) == 1

pkt = IP()
pkt.flags = ""
assert pkt.flags == 0

= TCP flags
~ TCP

pkt = TCP(flags="SA")
assert pkt.flags == 18
assert pkt.flags.S
assert pkt.flags.A
assert pkt.flags.SA
assert not any(getattr(pkt.flags, f) for f in 'FRPUECN')
assert repr(pkt.flags) == '<Flag 18 (SA)>'
pkt.flags.U = True
pkt.flags.S = False
assert pkt.flags.A
assert pkt.flags.U
assert pkt.flags.AU
assert not any(getattr(pkt.flags, f) for f in 'FSRPECN')
assert repr(pkt.flags) == '<Flag 48 (AU)>'
pkt.flags &= 'SFA'
pkt.flags |= 'P'
assert pkt.flags.P
assert pkt.flags.A
assert pkt.flags.PA
assert not any(getattr(pkt.flags, f) for f in 'FSRUECN')

pkt = TCP(flags=56)
assert all(getattr(pkt.flags, f) for f in 'PAU')
assert pkt.flags.PAU
assert not any(getattr(pkt.flags, f) for f in 'FSRECN')
assert repr(pkt.flags) == '<Flag 56 (PAU)>'
pkt.flags = 50
assert all(getattr(pkt.flags, f) for f in 'SAU')
assert pkt.flags.SAU
assert not any(getattr(pkt.flags, f) for f in 'FRPECN')
assert repr(pkt.flags) == '<Flag 50 (SAU)>'

= Flag values mutation with .raw_packet_cache
~ IP TCP

pkt = IP(raw(IP(flags="MF")/TCP(flags="SA")))
assert pkt.raw_packet_cache is not None
assert pkt[TCP].raw_packet_cache is not None
assert pkt.flags.MF
assert not pkt.flags.DF
assert not pkt.flags.evil
assert repr(pkt.flags) == '<Flag 1 (MF)>'
assert pkt[TCP].flags.S
assert pkt[TCP].flags.A
assert pkt[TCP].flags.SA
assert not any(getattr(pkt[TCP].flags, f) for f in 'FRPUECN')
assert repr(pkt[TCP].flags) == '<Flag 18 (SA)>'
pkt.flags.MF = 0
pkt.flags.DF = 1
pkt[TCP].flags.U = True
pkt[TCP].flags.S = False
pkt = IP(raw(pkt))
assert not pkt.flags.MF
assert pkt.flags.DF
assert not pkt.flags.evil
assert repr(pkt.flags) == '<Flag 2 (DF)>'
assert pkt[TCP].flags.A
assert pkt[TCP].flags.U
assert pkt[TCP].flags.AU
assert not any(getattr(pkt[TCP].flags, f) for f in 'FSRPECN')
assert repr(pkt[TCP].flags) == '<Flag 48 (AU)>'

= Operations on flag values
~ TCP

p1, p2 = TCP(flags="SU"), TCP(flags="AU")
assert (p1.flags & p2.flags).U
assert not any(getattr(p1.flags & p2.flags, f) for f in 'FSRPAECN')
assert all(getattr(p1.flags | p2.flags, f) for f in 'SAU')
assert (p1.flags | p2.flags).SAU
assert not any(getattr(p1.flags | p2.flags, f) for f in 'FRPECN')

assert TCP(flags="SA").flags & TCP(flags="S").flags == TCP(flags="S").flags
assert TCP(flags="SA").flags | TCP(flags="S").flags == TCP(flags="SA").flags


############
############
+ 802.3

= Test detection

assert isinstance(Dot3(raw(Ether())),Ether)
assert isinstance(Ether(raw(Dot3())),Dot3)

a = Ether(b'\xff\xff\xff\xff\xff\xff\x00\x00\x00\x00\x00\x00\x00\x00')
assert isinstance(a,Dot3)
assert a.dst == 'ff:ff:ff:ff:ff:ff'
assert a.src == '00:00:00:00:00:00'

a = Dot3(b'\xff\xff\xff\xff\xff\xff\x00\x00\x00\x00\x00\x00\x90\x00')
assert isinstance(a,Ether)
assert a.dst == 'ff:ff:ff:ff:ff:ff'
assert a.src == '00:00:00:00:00:00'


############
############
+ ASN.1

= MIB
~ mib

import tempfile
fd, fname = tempfile.mkstemp()
os.write(fd, b"-- MIB test\nscapy       OBJECT IDENTIFIER ::= {test 2807}\n")
os.close(fd)

load_mib(fname)
assert sum(1 for k in six.itervalues(conf.mib.d) if "scapy" in k) == 1

assert sum(1 for oid in conf.mib) > 100

= MIB - graph
~ mib

import mock

@mock.patch("scapy.asn1.mib.do_graph")
def get_mib_graph(do_graph):
    def store_graph(graph, **kargs):
        assert graph.startswith("""digraph "mib" {""")
        assert """"test.2807" [ label="scapy"  ];""" in graph
    do_graph.side_effect = store_graph
    conf.mib._make_graph()

get_mib_graph()

= MIB - test aliases
~ mib

# https://github.com/secdev/scapy/issues/2542
assert conf.mib._oidname("2.5.29.19") == "basicConstraints"

= DADict tests

a = DADict("test")
a[0] = "test_value1"
a["scapy"] = "test_value2"

assert a.test_value1 == 0
assert a.test_value2 == "scapy"

with ContextManagerCaptureOutput() as cmco:
    a._show()
    outp = cmco.get_output()

assert "scapy = 'test_value2'" in outp
assert "0 = 'test_value1'" in outp

= Test ETHER_TYPES

assert ETHER_TYPES.IPv4 == 2048
try:
    import warnings
    
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        ETHER_TYPES["BAOBAB"] = 0xffff
        assert ETHER_TYPES.BAOBAB == 0xffff
        assert issubclass(w[-1].category, DeprecationWarning)
except DeprecationWarning:
    # -Werror is used
    pass

= BER tests

BER_id_enc(42) == '*'
BER_id_enc(2807) == b'\xbfw'

b = BERcodec_IPADDRESS()
r1 = b.enc("8.8.8.8")
r1 == b'@\x04\x08\x08\x08\x08'

r2 = b.dec(r1)[0]
r2.val == '8.8.8.8'

a = b'\x1f\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\xfe\x01\x01\x00C\x02\x01U0\x0f0\r\x06\x08+\x06\x01\x02\x01\x02\x01\x00\x02\x01!'
ret = False
try:
    BERcodec_Object.check_type(a)
except BER_BadTag_Decoding_Error:
    ret = True
else:
    ret = False

assert ret

= BER trigger failures

try:
    BERcodec_INTEGER.do_dec(b"\x02\x01")
    assert False
except BER_Decoding_Error:
    pass


############
############
+ Fields

= FieldLenField with BitField
class Test(Packet):
    name = "Test"
    fields_desc = [
        FieldLenField("BitCount", None, fmt="H", count_of="Values"),
        FieldLenField("ByteCount", None, fmt="B", length_of="Values"),
        FieldListField("Values", [], BitField("data", 0x0, size=1),
                       count_from=lambda pkt: pkt.BitCount),
    ]

pkt = Test(raw(Test(Values=[0, 0, 0, 0, 1, 1, 1, 1])))
assert pkt.BitCount == 8
assert pkt.ByteCount == 1

= PacketListField

class TestPacket(Packet):
  name = 'TestPacket'
  fields_desc = [ PacketListField('list', [], 0) ]

a = TestPacket()
a.list.append(1)
assert len(a.list) == 1

b = TestPacket()
assert len(b.list) == 0

= Test PacketListField deepcopy
class SubPacket(Packet):
    name = "SubPacket"
    fields_desc = [
        ByteField("mem", 1),
    ]

class TestPacket(Packet):
    name = "TestPacket"
    fields_desc = [
        PacketListField("packlist", SubPacket(), SubPacket),
    ]

a = TestPacket()
b = a.copy()
fuzz(b)
assert a.packlist[0].mem == 1

= PacketField

class InnerPacket(Packet):
    fields_desc = [ StrField("f_name", "test") ]

class TestPacket(Packet):
    fields_desc = [ PacketField("inner", InnerPacket(), InnerPacket) ]

p = TestPacket()
print(p.inner.f_name)
assert p.inner.f_name == b"test"

p = TestPacket()
p.inner.f_name = b"scapy"
assert p.inner.f_name == b"scapy"

p = TestPacket()
assert p.inner.f_name == b"test"

+ UUIDField

= Parsing a human-readable UUID
f = UUIDField('f', '01234567-89ab-cdef-0123-456789abcdef')
f.addfield(None, b'', f.default) == hex_bytes('0123456789abcdef0123456789abcdef')

= Parsing a machine-encoded UUID
f = UUIDField('f', bytearray.fromhex('0123456789abcdef0123456789abcdef'))
f.addfield(None, b'', f.default) == hex_bytes('0123456789abcdef0123456789abcdef')

= Parsing a tuple of values
f = UUIDField('f', (0x01234567, 0x89ab, 0xcdef, 0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef))
f.addfield(None, b'', f.default) == hex_bytes('0123456789abcdef0123456789abcdef')

= Handle None values
f = UUIDField('f', None)
f.addfield(None, b'', f.default) == hex_bytes('00000000000000000000000000000000')

= Get a UUID for dissection
from uuid import UUID
f = UUIDField('f', None)
f.getfield(None, bytearray.fromhex('0123456789abcdef0123456789abcdef01')) == (b'\x01', UUID('01234567-89ab-cdef-0123-456789abcdef'))

= Verify little endian UUIDField
* The endianness of a UUIDField should be apply by block on each block in parenthesis '(01234567)-(89ab)-(cdef)-(01)(23)-(45)(67)(89)(ab)(cd)(ef)'
f = UUIDField('f', '01234567-89ab-cdef-0123-456789abcdef', uuid_fmt=UUIDField.FORMAT_LE)
f.addfield(None, b'', f.default) == hex_bytes('67452301ab89efcd0123456789abcdef')

= Verify reversed UUIDField
* This should reverse the entire value as 128-bits
f = UUIDField('f', '01234567-89ab-cdef-0123-456789abcdef', uuid_fmt=UUIDField.FORMAT_REV)
f.addfield(None, b'', f.default) == hex_bytes('efcdab8967452301efcdab8967452301')

+ RandUUID

= RandUUID setup

RANDUUID_TEMPLATE = '01234567-89ab-*-01*-*****ef'
RANDUUID_FIXED = uuid.uuid4()

= RandUUID default behaviour

ru = RandUUID()
assert ru._fix().version == 4
assert ru.command() == "RandUUID()"

= RandUUID incorrect implicit args

assert expect_exception(ValueError, lambda: RandUUID(node=0x1234, name="scapy"))
assert expect_exception(ValueError, lambda: RandUUID(node=0x1234, namespace=uuid.uuid4()))
assert expect_exception(ValueError, lambda: RandUUID(clock_seq=0x1234, name="scapy"))
assert expect_exception(ValueError, lambda: RandUUID(clock_seq=0x1234, namespace=uuid.uuid4()))
assert expect_exception(ValueError, lambda: RandUUID(name="scapy"))
assert expect_exception(ValueError, lambda: RandUUID(namespace=uuid.uuid4()))

= RandUUID v4 UUID (correct args)

u = RandUUID(version=4)._fix()
assert u.version == 4

u2 = RandUUID(version=4)._fix()
assert u2.version == 4

assert str(u) != str(u2)

= RandUUID v4 UUID (incorrect args)

assert expect_exception(ValueError, lambda: RandUUID(version=4, template=RANDUUID_TEMPLATE))
assert expect_exception(ValueError, lambda: RandUUID(version=4, node=0x1234))
assert expect_exception(ValueError, lambda: RandUUID(version=4, clock_seq=0x1234))
assert expect_exception(ValueError, lambda: RandUUID(version=4, namespace=uuid.uuid4()))
assert expect_exception(ValueError, lambda: RandUUID(version=4, name="scapy"))

= RandUUID v1 UUID

u = RandUUID(version=1)._fix()
assert u.version in [1, 4]

u = RandUUID(version=1, node=0x1234)._fix()
assert u.version == 1
assert u.node == 0x1234

u = RandUUID(version=1, clock_seq=0x1234)._fix()
assert u.version == 1
assert u.clock_seq == 0x1234

ru = RandUUID(version=1, node=0x1234, clock_seq=0x1bcd)
assert ru.command() == "RandUUID(node=4660, clock_seq=7117, version=1)"
u = ru._fix()
assert u.version == 1
assert u.node == 0x1234
assert u.clock_seq == 0x1bcd

= RandUUID v1 UUID (implicit version)

u = RandUUID(node=0x1234)._fix()
assert u.version == 1
assert u.node == 0x1234

u = RandUUID(clock_seq=0x1234)._fix()
assert u.version == 1
assert u.clock_seq == 0x1234

u = RandUUID(node=0x1234, clock_seq=0x1bcd)._fix()
assert u.version == 1
assert u.node == 0x1234
assert u.clock_seq == 0x1bcd

= RandUUID v1 UUID (incorrect args)

assert expect_exception(ValueError, lambda: RandUUID(version=1, template=RANDUUID_TEMPLATE))
assert expect_exception(ValueError, lambda: RandUUID(version=1, namespace=uuid.uuid4()))
assert expect_exception(ValueError, lambda: RandUUID(version=1, name="scapy"))

= RandUUID v5 UUID

ru = RandUUID(version=5, namespace=RANDUUID_FIXED, name="scapy")
u = ru._fix()
assert u.version == 5
assert ru.command() == "RandUUID(namespace=%r, name='scapy', version=5)" % RANDUUID_FIXED

u2 = RandUUID(version=5, namespace=RANDUUID_FIXED, name="scapy")._fix()
assert u2.version == 5
assert u.bytes == u2.bytes

# implicit v5
u2 = RandUUID(namespace=RANDUUID_FIXED, name="scapy")._fix()
assert u.bytes == u2.bytes

= RandUUID v5 UUID (incorrect args)

assert expect_exception(ValueError, lambda: RandUUID(version=5, template=RANDUUID_TEMPLATE))
assert expect_exception(ValueError, lambda: RandUUID(version=5, node=0x1234))
assert expect_exception(ValueError, lambda: RandUUID(version=5, clock_seq=0x1234))

= RandUUID v3 UUID

u = RandUUID(version=3, namespace=RANDUUID_FIXED, name="scapy")._fix()
assert u.version == 3

u2 = RandUUID(version=3, namespace=RANDUUID_FIXED, name="scapy")._fix()
assert u2.version == 3
assert u.bytes == u2.bytes

# implicit v5
u2 = RandUUID(namespace=RANDUUID_FIXED, name="scapy")._fix()
assert u.bytes != u2.bytes

= RandUUID v3 UUID (incorrect args)

assert expect_exception(ValueError, lambda: RandUUID(version=5, template=RANDUUID_TEMPLATE))
assert expect_exception(ValueError, lambda: RandUUID(version=5, node=0x1234))
assert expect_exception(ValueError, lambda: RandUUID(version=5, clock_seq=0x1234))

= RandUUID looks like a UUID with str
assert re.match(r'[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}', str(RandUUID()), re.I) is not None

= RandUUID with a static part
* RandUUID template can contain static part such a 01234567-89ab-*-01*-*****ef
ru = RandUUID('01234567-89ab-*-01*-*****ef')
assert re.match(r'01234567-89ab-[0-9a-f]{4}-01[0-9a-f]{2}-[0-9a-f]{10}ef', str(ru), re.I) is not None
assert ru.command() == "RandUUID(template='01234567-89ab-*-01*-*****ef')"

= RandUUID with a range part
* RandUUID template can contain a part with a range of values such a 01234567-89ab-*-01*-****c0:c9ef
assert re.match(r'01234567-89ab-[0-9a-f]{4}-01[0-9a-f]{2}-[0-9a-f]{8}c[0-9]ef', str(RandUUID('01234567-89ab-*-01*-****c0:c9ef')), re.I) is not None

############
############
+ MPLS tests

= MPLS - build/dissection
from scapy.contrib.mpls import EoMCW, MPLS
p1 = MPLS()/IP()/UDP()
assert p1[MPLS].s == 1
p2 = MPLS()/MPLS()/IP()/UDP()
assert p2[MPLS].s == 0

p1[MPLS]
p1[IP]
p2[MPLS]
p2[MPLS:1]
p2[IP]

= MPLS encapsulated Ethernet with CW - build/dissection
p = Ether(dst="11:11:11:11:11:11", src="22:22:22:22:22:22")
p /= MPLS(label=1)/EoMCW(seq=1234)
p /= Ether(dst="33:33:33:33:33:33", src="44:44:44:44:44:44")/IP()
p = Ether(raw(p))
assert p[EoMCW].zero == 0
assert p[EoMCW].reserved == 0
assert p[EoMCW].seq == 1234

=  MPLS encapsulated Ethernet without CW - build/dissection
p = Ether(dst="11:11:11:11:11:11", src="22:22:22:22:22:22")
p /= MPLS(label=2)/MPLS(label=1)
p /= Ether(dst="33:33:33:33:33:33", src="44:44:44:44:44:44")/IP()
p = Ether(raw(p))
assert p[Ether:2].type == 0x0800

try:
    p[EoMCW]
except IndexError:
    ret = True
else:
    ret = False

assert ret
assert p[Ether:2].type == 0x0800

= MPLS encapsulated IP - build/dissection
p = Ether(dst="11:11:11:11:11:11", src="22:22:22:22:22:22")
p /= MPLS(label=1)/IP()
p = Ether(raw(p))

try:
    p[EoMCW]
except IndexError:
    ret = True
else:
    ret = False

assert ret

try:
    p[Ether:2]
except IndexError:
    ret = True
else:
    ret = False

assert ret

p[IP]


############
############
+ PacketList methods

= sr()

class Req(Packet):
    fields_desc = [
        ByteField("raw", 0)
    ]
    def answers(self, other):
        return False

class Res(Packet):
    fields_desc = [
        ByteField("raw", 0)
    ]
    def answers(self, other):
        return other.__class__ == Req and other.raw == self.raw

pl = PacketList([Req(b"1"), Res(b"1"), Req(b"2"), Req(b"3"), Req(b"4"), Res(b"3"), Res(b"1"), Res(b"1"), Res(b"4")])

srl, rl = pl.sr()
assert len(srl) == 3
assert len(rl) == 3

srl, rl = pl.sr(lookahead=1)
assert len(srl) == 1
assert len(rl) == 7

srl, rl = pl.sr(lookahead=2)
assert len(srl) == 2
assert len(rl) == 5

srl, rl = pl.sr(lookahead=3)
assert len(srl) == 3
assert len(rl) == 3

pl = PacketList([Req(b"\x05"), Res(b"1"), Res(b"2"), Res(b"3"), Res(b"4"), Res(b"3"), Res(b"1"), Res(b"1"), Res(b"\x05")])

srl, rl = pl.sr(lookahead=3)
assert len(srl) == 0
assert len(rl) == 9

srl, rl = pl.sr(lookahead=7)
assert len(srl) == 0
assert len(rl) == 9

srl, rl = pl.sr(lookahead=8)
assert len(srl) == 1
assert len(rl) == 7

srl, rl = pl.sr(lookahead=0)
assert len(srl) == 1
assert len(rl) == 7

srl, rl = pl.sr(lookahead=None)
assert len(srl) == 1
assert len(rl) == 7

= pickle test
import pickle
import io

srl, rl = PacketList([Raw(b"1"), Raw(b"1"), Raw(b"2"), Raw(b"3"), Raw(b"4"), Raw(b"3"), Raw(b"1"), Raw(b"1"), Raw(b"4")]).sr()
assert len(srl) == 4

f = io.BytesIO()

pickle.dump(srl, f)

unp = pickle.loads(f.getvalue())

assert len(unp) == len(srl)
assert all(bytes(a[0]) == bytes(b[0]) for a, b in zip(unp, srl))

= plot()

import mock
import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
def test_plot(mock_plt):
    def fake_plot(data, **kwargs):
        return data
    mock_plt.plot = fake_plot
    plist = PacketList([IP(id=i)/TCP() for i in range(10)])
    lines = plist.plot(lambda p: (p.time, p.id))
    assert len(lines) == 10

test_plot()

= diffplot()

import mock
import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
def test_diffplot(mock_plt):
    def fake_plot(data, **kwargs):
        return data
    mock_plt.plot = fake_plot
    plist = PacketList([IP(id=i)/TCP() for i in range(10)])
    lines = plist.diffplot(lambda x,y: (x.time, y.id-x.id))
    assert len(lines) == 9

test_diffplot()

= multiplot()

import mock
import scapy.libs.matplot

@mock.patch("scapy.libs.matplot.plt")
def test_multiplot(mock_plt):
    def fake_plot(data, **kwargs):
        return data
    mock_plt.plot = fake_plot
    tmp = [IP(id=i)/TCP() for i in range(10)]
    plist = PacketList([tuple(tmp[i-2:i]) for i in range(2, 10, 2)])
    lines = plist.multiplot(lambda x: (x[1][IP].src, (x[1].time, x[1][IP].id)))
    assert len(lines) == 1
    assert len(lines[0]) == 4

test_multiplot()

= rawhexdump()

def test_rawhexdump():
    with ContextManagerCaptureOutput() as cmco:
        p = PacketList([IP()/TCP() for i in range(2)])
        p.rawhexdump()
        result_pl_rawhexdump = cmco.get_output()
    assert len(result_pl_rawhexdump.split('\n')) == 7
    assert result_pl_rawhexdump.startswith("0000  45 00 00 28")

test_rawhexdump()

= hexraw()

def test_hexraw():
    with ContextManagerCaptureOutput() as cmco:
        p = PacketList([IP()/Raw(str(i)) for i in range(2)])
        p.hexraw()
        result_pl_hexraw = cmco.get_output()
    assert len(result_pl_hexraw.split('\n')) == 5
    assert "0000  30" in result_pl_hexraw

test_hexraw()

= hexdump()

def test_hexdump():
    with ContextManagerCaptureOutput() as cmco:
        p = PacketList([IP()/Raw(str(i)) for i in range(2)])
        p.hexdump()
        result_pl_hexdump = cmco.get_output()
    assert len(result_pl_hexdump.split('\n')) == 7
    assert "0010  7F 00 00 01 31" in result_pl_hexdump

test_hexdump()

= import_hexcap()

@mock.patch("scapy.utils.input")
def test_import_hexcap(mock_input):
    data = """
0000  FF FF FF FF FF FF AA AA AA AA AA AA 08 00 45 00  ..............E.
0010  00 1C 00 01 00 00 40 01 7C DE 7F 00 00 01 7F 00  ......@.|.......
0020  00 01 08 00 F7 FF 00 00 00 00                    ..........
"""[1:].split("\n")
    lines = iter(data)
    mock_input.side_effect = lambda: next(lines)
    return import_hexcap()

pkt = test_import_hexcap()
pkt = Ether(pkt)
assert pkt[Ether].dst == "ff:ff:ff:ff:ff:ff"
assert pkt[IP].dst == "127.0.0.1"
assert ICMP in pkt

= import_hexcap(input_string)
data = """
0000  FF FF FF FF FF FF AA AA AA AA AA AA 08 00 45 00  ..............E.
0010  00 1C 00 01 00 00 40 01 7C DE 7F 00 00 01 7F 00  ......@.|.......
0020  00 01 08 00 F7 FF 00 00 00 00                    ..........
"""[1:]
pkt = import_hexcap(data)
pkt = Ether(pkt)
assert pkt[Ether].dst == "ff:ff:ff:ff:ff:ff"
assert pkt[IP].dst == "127.0.0.1"
assert ICMP in pkt

= padding()

def test_padding():
    with ContextManagerCaptureOutput() as cmco:
        p = PacketList([IP()/conf.padding_layer(str(i)) for i in range(2)])
        p.padding()
        result_pl_padding = cmco.get_output()
    assert len(result_pl_padding.split('\n')) == 5
    assert "0000  30" in result_pl_padding

test_padding()

= nzpadding()

def test_nzpadding():
    with ContextManagerCaptureOutput() as cmco:
        p = PacketList([IP()/conf.padding_layer("AB"), IP()/conf.padding_layer("\x00\x00")])
        p.nzpadding()
        result_pl_nzpadding = cmco.get_output()
    assert len(result_pl_nzpadding.split('\n')) == 3
    assert "0000  41 42" in result_pl_nzpadding

test_nzpadding()

= conversations()

import mock
@mock.patch("scapy.plist.do_graph")
def test_conversations(mock_do_graph):
    def fake_do_graph(graph, **kwargs):
        return graph
    mock_do_graph.side_effect = fake_do_graph
    plist = PacketList([IP(dst="127.0.0.2")/TCP(dport=i) for i in range(2)])
    plist.extend([IP(src="127.0.0.2")/TCP(sport=i) for i in range(2)])
    plist.extend([IPv6(dst="::2", src="::1")/TCP(sport=i) for i in range(2)])
    plist.extend([IPv6(src="::2", dst="::1")/TCP(sport=i) for i in range(2)])
    plist.extend([Ether()/ARP(pdst="127.0.0.1")])
    result_conversations = plist.conversations()
    assert len(result_conversations.split('\n')) == 8
    assert result_conversations.startswith('digraph "conv" {')
    assert "127.0.0.1" in result_conversations
    assert "::1" in result_conversations

test_conversations()

= sessions()

pl = PacketList([Ether()/IPv6()/ICMPv6EchoRequest(), Ether()/IPv6()/IPv6()])
pl.extend([Ether()/IP()/IP(), Ether()/ARP()])
pl.extend([Ether()/Ether()/IP()])
assert len(pl.sessions().keys()) == 5

= afterglow()

import mock
@mock.patch("scapy.plist.do_graph")
def test_afterglow(mock_do_graph):
    def fake_do_graph(graph, **kwargs):
        return graph
    mock_do_graph.side_effect = fake_do_graph
    plist = PacketList([IP(dst="127.0.0.2")/TCP(dport=i) for i in range(2)])
    plist.extend([IP(src="127.0.0.2")/TCP(sport=i) for i in range(2)])
    result_afterglow = plist.afterglow()
    assert len(result_afterglow.split('\n')) == 19
    assert result_afterglow.startswith('digraph "afterglow" {')

test_afterglow()

= psdump()

print("PYX: %d" % PYX)
if PYX:
    import tempfile
    import os
    filename = tempfile.mktemp(suffix=".eps")
    plist = PacketList([IP()/TCP()])
    plist.psdump(filename)
    assert os.path.exists(filename)
    os.unlink(filename)

= pdfdump()

print("PYX: %d" % PYX)
if PYX:
    import tempfile
    import os
    filename = tempfile.mktemp(suffix=".pdf")
    plist = PacketList([IP()/TCP()])
    plist.pdfdump(filename)
    assert os.path.exists(filename)
    os.unlink(filename)

= svgdump()

print("PYX: %d" % PYX)
if PYX and not six.PY2:
    import tempfile
    import os
    filename = tempfile.mktemp(suffix=".svg")
    plist = PacketList([IP()/TCP()])
    plist.svgdump(filename)
    assert os.path.exists(filename)
    os.unlink(filename)

= __getstate__ / __setstate__ (used by pickle)

import pickle

frm = Ether(src='00:11:22:33:44:55', dst='00:22:33:44:55:66')/Raw()
frm.time = EDecimal(123.45)
frm.sniffed_on = "iface"
frm.wirelen = 1
pl = PacketList(res=[frm, frm], name='WhatAGreatName')
pickled = pickle.dumps(pl)
pl = pickle.loads(pickled)
assert pl.listname == "WhatAGreatName"
assert len(pl) == 2
assert pl[0].time == 123.45
assert pl[0].sniffed_on == "iface"
assert pl[0].wirelen == 1
assert pl[0][Ether].src == '00:11:22:33:44:55'
assert pl[1][Ether].dst == '00:22:33:44:55:66'


############
############
+ Scapy version

= UTscapy HTML output

import tempfile, os
from scapy.tools.UTscapy import TestCampaign, pack_html_campaigns
test_campaign = TestCampaign("test")
test_campaign.output_file = tempfile.mktemp()
html = pack_html_campaigns([test_campaign], None, local=True)
dirname = os.path.dirname(test_campaign.output_file)
filename_js = "%s/UTscapy.js" % dirname
filename_css = "%s/UTscapy.css" % dirname
assert os.path.isfile(filename_js)
assert os.path.isfile(filename_css)
os.remove(filename_js)
os.remove(filename_css)

= test get_temp_dir

dname = get_temp_dir()
assert os.path.isdir(dname)

= test fragleak functions
~ netaccess linux fragleak

import mock

@mock.patch("scapy.layers.inet.conf.L3socket")
@mock.patch("scapy.layers.inet.select.select")
@mock.patch("scapy.layers.inet.sr1")
def _test_fragleak(func, sr1, select, L3socket):
    packets = [IP(src="4.4.4.4")/ICMP()/IPerror(dst="8.8.8.8")/conf.padding_layer(load=b"greatdata")]
    iterator = iter(packets)
    ne = lambda *args, **kwargs: next(iterator)
    L3socket.side_effect = lambda: Bunch(recv=ne, send=lambda x: None)
    sr1.side_effect = ne
    select.side_effect = lambda a, b, c, d: a+b+c
    with ContextManagerCaptureOutput() as cmco:
        func("8.8.8.8", count=1)
        out = cmco.get_output()
        return "greatdata" in out

assert _test_fragleak(fragleak)
assert _test_fragleak(fragleak2)