test_ip.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
from __future__ import annotations

from zope import interface

from twisted.pair import ip, raw
from twisted.python import components
from twisted.trial import unittest

@interface.implementer(raw.IRawDatagramProtocol)
class MyProtocol:
    def __init__(self, expecting: list[tuple[bytes, dict[str, str | int]]]) -> None:
        self.expecting = list(expecting)

def datagramReceived(
        self,
        data: bytes,
        partial: int,
        source: str,
        dest: str,
        protocol: int,
        version: int,
        ihl: int,
        tos: int,
        tot_len: int,
        fragment_id: int,
        fragment_offset: int,
        dont_fragment: int,
        more_fragments: int,
        ttl: int,
    ) -> None:
        assert self.expecting, "Got a packet when not expecting anymore."
        expectData, expectKw = self.expecting.pop(0)

expectKwKeys = list(sorted(expectKw.keys()))
        localVariables = locals()

for k in expectKwKeys:
            assert (
                expectKw[k] == localVariables[k]
            ), f"Expected {k}={expectKw[k]!r}, got {localVariables[k]!r}"
        assert expectData == data, f"Expected {expectData!r}, got {data!r}"

def addProto(self, num: object, proto: object) -> None:
        # IRawDatagramProtocol.addProto
        pass

class IPTests(unittest.TestCase):
    def testPacketParsing(self) -> None:
        proto = ip.IPProtocol()
        p1 = MyProtocol(
            [
                (
                    b"foobar",
                    {
                        "partial": 0,
                        "dest": "1.2.3.4",
                        "source": "5.6.7.8",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
            ]
        )
        proto.addProto(0x0F, p1)

proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0F"  # protocol
            + b"FE"  # checksum
            + b"\x05\x06\x07\x08"
            + b"\x01\x02\x03\x04"
            + b"foobar",
            partial=0,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )

assert not p1.expecting, (
            "Should not expect any more packets, but still want %r" % p1.expecting
        )

def testMultiplePackets(self) -> None:
        proto = ip.IPProtocol()
        p1 = MyProtocol(
            [
                (
                    b"foobar",
                    {
                        "partial": 0,
                        "dest": "1.2.3.4",
                        "source": "5.6.7.8",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
                (
                    b"quux",
                    {
                        "partial": 1,
                        "dest": "5.4.3.2",
                        "source": "6.7.8.9",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
            ]
        )
        proto.addProto(0x0F, p1)
        proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0F"  # protocol
            + b"FE"  # checksum
            + b"\x05\x06\x07\x08"
            + b"\x01\x02\x03\x04"
            + b"foobar",
            partial=0,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )
        proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0F"  # protocol
            + b"FE"  # checksum
            + b"\x06\x07\x08\x09"
            + b"\x05\x04\x03\x02"
            + b"quux",
            partial=1,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )

assert not p1.expecting, (
            "Should not expect any more packets, but still want %r" % p1.expecting
        )

def testMultipleSameProtos(self) -> None:
        proto = ip.IPProtocol()
        p1 = MyProtocol(
            [
                (
                    b"foobar",
                    {
                        "partial": 0,
                        "dest": "1.2.3.4",
                        "source": "5.6.7.8",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
            ]
        )

p2 = MyProtocol(
            [
                (
                    b"foobar",
                    {
                        "partial": 0,
                        "dest": "1.2.3.4",
                        "source": "5.6.7.8",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
            ]
        )

proto.addProto(0x0F, p1)
        proto.addProto(0x0F, p2)

assert not p1.expecting, (
            "Should not expect any more packets, but still want %r" % p1.expecting
        )
        assert not p2.expecting, (
            "Should not expect any more packets, but still want %r" % p2.expecting
        )

def testWrongProtoNotSeen(self) -> None:
        proto = ip.IPProtocol()
        p1 = MyProtocol([])
        proto.addProto(1, p1)

def testDemuxing(self) -> None:
        proto = ip.IPProtocol()
        p1 = MyProtocol(
            [
                (
                    b"foobar",
                    {
                        "partial": 0,
                        "dest": "1.2.3.4",
                        "source": "5.6.7.8",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
                (
                    b"quux",
                    {
                        "partial": 1,
                        "dest": "5.4.3.2",
                        "source": "6.7.8.9",
                        "protocol": 0x0F,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
            ]
        )
        proto.addProto(0x0F, p1)

p2 = MyProtocol(
            [
                (
                    b"quux",
                    {
                        "partial": 1,
                        "dest": "5.4.3.2",
                        "source": "6.7.8.9",
                        "protocol": 0x0A,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
                (
                    b"foobar",
                    {
                        "partial": 0,
                        "dest": "1.2.3.4",
                        "source": "5.6.7.8",
                        "protocol": 0x0A,
                        "version": 4,
                        "ihl": 20,
                        "tos": 7,
                        "tot_len": 20 + 6,
                        "fragment_id": 0xDEAD,
                        "fragment_offset": 0x1EEF,
                        "dont_fragment": 0,
                        "more_fragments": 1,
                        "ttl": 0xC0,
                    },
                ),
            ]
        )
        proto.addProto(0x0A, p2)

proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0A"  # protocol
            + b"FE"  # checksum
            + b"\x06\x07\x08\x09"
            + b"\x05\x04\x03\x02"
            + b"quux",
            partial=1,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )
        proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0F"  # protocol
            + b"FE"  # checksum
            + b"\x05\x06\x07\x08"
            + b"\x01\x02\x03\x04"
            + b"foobar",
            partial=0,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )
        proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0F"  # protocol
            + b"FE"  # checksum
            + b"\x06\x07\x08\x09"
            + b"\x05\x04\x03\x02"
            + b"quux",
            partial=1,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )
        proto.datagramReceived(
            b"\x54"  # ihl version
            + b"\x07"  # tos
            + b"\x00\x1a"  # tot_len
            + b"\xDE\xAD"  # id
            + b"\xBE\xEF"  # frag_off
            + b"\xC0"  # ttl
            + b"\x0A"  # protocol
            + b"FE"  # checksum
            + b"\x05\x06\x07\x08"
            + b"\x01\x02\x03\x04"
            + b"foobar",
            partial=0,
            dest="dummy",
            source="dummy",
            protocol="dummy",
        )

def testAddingBadProtos_WrongLevel(self) -> None:
        """Adding a wrong level protocol raises an exception."""
        e = ip.IPProtocol()
        try:
            e.addProto(42, "silliness")
        except components.CannotAdapt:
            pass
        else:
            raise AssertionError("addProto must raise an exception for bad protocols")

def testAddingBadProtos_TooSmall(self) -> None:
        """Adding a protocol with a negative number raises an exception."""
        e = ip.IPProtocol()
        try:
            e.addProto(-1, MyProtocol([]))
        except TypeError as e:
            if e.args == ("Added protocol must be positive or zero",):
                pass
            else:
                raise
        else:
            raise AssertionError("addProto must raise an exception for bad protocols")

def testAddingBadProtos_TooBig(self) -> None:
        """Adding a protocol with a number >=2**32 raises an exception."""
        e = ip.IPProtocol()
        try:
            e.addProto(2**32, MyProtocol([]))
        except TypeError as e:
            if e.args == ("Added protocol must fit in 32 bits",):
                pass
            else:
                raise
        else:
            raise AssertionError("addProto must raise an exception for bad protocols")

def testAddingBadProtos_TooBig2(self) -> None:
        """Adding a protocol with a number >=2**32 raises an exception."""
        e = ip.IPProtocol()
        try:
            e.addProto(2**32 + 1, MyProtocol([]))
        except TypeError as e:
            if e.args == ("Added protocol must fit in 32 bits",):
                pass
            else:
                raise
        else:
            raise AssertionError("addProto must raise an exception for bad protocols")

docroot: /var/www/469ed2f6-4d4b-4a8a-9f5e-ee4ef04d3bc8/public_html

curdir: /usr/lib/python3/dist-packages/twisted/pair/test

	.	4096
	..	4096
X	__init__.py	13
	__pycache__	4096
X	test_ethernet.py	7929
X	test_ip.py	15454
X	test_rawudp.py	10648
X	test_tuntap.py	46759