FM / pair

/usr/lib/python3/dist-packages/twisted/pair/

/usr/lib/python3/dist-packages/twisted/pair

Name	Size	Mode	Actions
test/	-	0755	rm
__pycache__/	-	0755	rm
ethernet.py	1698	0644	edit dl rm
ip.py	2595	0644	edit dl rm
raw.py	811	0644	edit dl rm
rawudp.py	1813	0644	edit dl rm
testing.py	17307	0644	edit dl rm
tuntap.py	12538	0644	edit dl rm
__init__.py	631	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/twisted/pair/ethernet.py (1698B)

# -*- test-case-name: twisted.pair.test.test_ethernet -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""Support for working directly with ethernet frames"""

import struct

from twisted.internet import protocol
from twisted.pair import raw
from zope.interface import implementer, Interface

class IEthernetProtocol(Interface):
    """An interface for protocols that handle Ethernet frames"""
    def addProto():
        """Add an IRawPacketProtocol protocol"""

def datagramReceived():
        """An Ethernet frame has been received"""

class EthernetHeader:
    def __init__(self, data):

(self.dest, self.source, self.proto) \
                    = struct.unpack("!6s6sH", data[:6+6+2])

@implementer(IEthernetProtocol)
class EthernetProtocol(protocol.AbstractDatagramProtocol):
    def __init__(self):
        self.etherProtos = {}

def addProto(self, num, proto):
        proto = raw.IRawPacketProtocol(proto)
        if num < 0:
            raise TypeError('Added protocol must be positive or zero')
        if num >= 2**16:
            raise TypeError('Added protocol must fit in 16 bits')
        if num not in self.etherProtos:
            self.etherProtos[num] = []
        self.etherProtos[num].append(proto)

def datagramReceived(self, data, partial=0):
        header = EthernetHeader(data[:14])
        for proto in self.etherProtos.get(header.proto, ()):
            proto.datagramReceived(data=data[14:],
                                   partial=partial,
                                   dest=header.dest,
                                   source=header.source,
                                   protocol=header.proto)

# -*- test-case-name: twisted.pair.test.test_ethernet -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. # """Support for working directly with ethernet frames""" import struct from twisted.internet import protocol from twisted.pair import raw from zope.interface import implementer, Interface class IEthernetProtocol(Interface): """An interface for protocols that handle Ethernet frames""" def addProto(): """Add an IRawPacketProtocol protocol""" def datagramReceived(): """An Ethernet frame has been received""" class EthernetHeader: def __init__(self, data): (self.dest, self.source, self.proto) \ = struct.unpack("!6s6sH", data[:6+6+2]) @implementer(IEthernetProtocol) class EthernetProtocol(protocol.AbstractDatagramProtocol): def __init__(self): self.etherProtos = {} def addProto(self, num, proto): proto = raw.IRawPacketProtocol(proto) if num < 0: raise TypeError('Added protocol must be positive or zero') if num >= 2**16: raise TypeError('Added protocol must fit in 16 bits') if num not in self.etherProtos: self.etherProtos[num] = [] self.etherProtos[num].append(proto) def datagramReceived(self, data, partial=0): header = EthernetHeader(data[:14]) for proto in self.etherProtos.get(header.proto, ()): proto.datagramReceived(data=data[14:], partial=partial, dest=header.dest, source=header.source, protocol=header.proto)