twisted/pair/ethernet.py

   1 # -*- test-case-name: twisted.pair.test.test_ethernet -*-
   2 # Copyright (c) Twisted Matrix Laboratories.
   3 # See LICENSE for details.
   4
   5 #
   6
   7
   8 """Support for working directly with ethernet frames"""
   9
  10 import struct
  11
  12
  13 from twisted.internet import protocol
  14 from twisted.pair import raw
  15 from zope.interface import implements, Interface
  16
  17
  18 class IEthernetProtocol(Interface):
  19     """An interface for protocols that handle Ethernet frames"""
  20     def addProto():
  21         """Add an IRawPacketProtocol protocol"""
  22
  23     def datagramReceived():
  24         """An Ethernet frame has been received"""
  25
  26 class EthernetHeader:
  27     def __init__(self, data):
  28
  29         (self.dest, self.source, self.proto) \
  30                     = struct.unpack("!6s6sH", data[:6+6+2])
  31
  32 class EthernetProtocol(protocol.AbstractDatagramProtocol):
  33
  34     implements(IEthernetProtocol)
  35
  36     def __init__(self):
  37         self.etherProtos = {}
  38
  39     def addProto(self, num, proto):
  40         proto = raw.IRawPacketProtocol(proto)
  41         if num < 0:
  42             raise TypeError, 'Added protocol must be positive or zero'
  43         if num >= 2**16:
  44             raise TypeError, 'Added protocol must fit in 16 bits'
  45         if num not in self.etherProtos:
  46             self.etherProtos[num] = []
  47         self.etherProtos[num].append(proto)
  48
  49     def datagramReceived(self, data, partial=0):
  50         header = EthernetHeader(data[:14])
  51         for proto in self.etherProtos.get(header.proto, ()):
  52             proto.datagramReceived(data=data[14:],
  53                                    partial=partial,
  54                                    dest=header.dest,
  55                                    source=header.source,
  56                                    protocol=header.proto)