Commit 83e8726d authored by Joel Sommers's avatar Joel Sommers
Browse files

v1 commit

parent 7d7b3b7d
Switchyard
==========
Switchyard (or "switchy") is a low-level networking library for software switches/routers in Python. Intended for use in layer2-/layer3-oriented labs and projects in computer networking courses.
There is some [documentation](https://github.com/jsommers/switchyard/wiki) in the Github wiki for Switchyard, though note that it is a work in progress.
Switchyard currently makes use of the [POX Openflow controller](https://github.com/noxrepo/pox) platform for packet parsing libraries, and other functions.
Switchyard can run in a standalone test mode, or also nicely within Mininet. The [Mininet project](http://www.mininet.org) pages have documentation for Mininet, as well as lots of other good stuff.
----
I gratefully acknowledge support from the NSF. The materials here are
based upon work supported by the National Science Foundation under
grant CNS-1054985 ("CAREER: Expanding the functionality of Internet
routers").
Any opinions, findings, and conclusions or recommendations expressed
in this material are those of the author and do not necessarily
reflect the views of the National Science Foundation.
This diff is collapsed.
import sys
import logging
# add POX to python path in a relatively generic way;
# assumes that there is a pox subdirectory off of user's home dir
import os.path
sys.path.append(os.path.join(os.environ['HOME'],'pox'))
sys.path.append(os.path.join(os.getcwd(),'pox'))
from pox.lib.addresses import IPAddr,EthAddr
import curses
curses.setupterm()
setaf = curses.tigetstr('setaf')
import atexit
curses.setupterm()
def resetterm():
print reset_term_color()
atexit.register(resetterm)
class SwitchyException(Exception):
pass
class Shutdown(Exception):
pass
class NoPackets(Exception):
pass
class ScenarioFailure(Exception):
pass
class PacketFormatter(object):
__fulldisp = False
@staticmethod
def full_display():
PacketFormatter.__fulldisp = True
@staticmethod
def format_pkt(pkt, cls=None):
'''
Return a string representation of a packet. If display_class is a known
header type, just show the string repr of that header. Otherwise, dump
the whole thing.
'''
if PacketFormatter.__fulldisp:
cls = None
if cls:
if not pkt.parsed:
raw = pkt.pack()
pkt.parse(raw)
header = pkt.find(cls)
if header is not None:
return str(header)
return str(pkt.dump())
class Interface(object):
'''
Class that models a single logical interface on a network
device. An interface has a name, 48-bit Ethernet MAC address,
and a 32-bit IPv4 address and mask.
'''
def __init__(self, name, ethaddr, ipaddr, netmask):
self.__name = name
if isinstance(ethaddr, EthAddr):
self.__ethaddr = ethaddr
else:
self.__ethaddr = EthAddr(ethaddr)
if isinstance(ipaddr, IPAddr):
self.__ipaddr = ipaddr
elif isinstance(ipaddr, str):
self.__ipaddr = IPAddr(ipaddr)
else:
self.__ipaddr = ipaddr
if isinstance(netmask, IPAddr):
self.__netmask = netmask
elif isinstance(netmask, str):
self.__netmask = IPAddr(netmask)
else:
self.__netmask = netmask
@property
def name(self):
return self.__name
@property
def ethaddr(self):
return self.__ethaddr
@property
def ipaddr(self):
return self.__ipaddr
@property
def netmask(self):
return self.__netmask
def __str__(self):
return "{} mac:{} ip {}/{}".format(str(self.name), str(self.ethaddr), str(self.ipaddr), str(self.netmask))
def term_color(s):
'''
Convenience function for setting foreground color.
'''
colordict = {
'green':curses.COLOR_GREEN,
'blue':curses.COLOR_BLUE,
'red':curses.COLOR_RED,
'yellow':curses.COLOR_YELLOW
}
if s in colordict:
return curses.tparm(setaf,colordict[s])
else:
return curses.tparm(curses.tigetstr('op'))
def reset_term_color():
'''
Convenience function for resetting terminal color.
'''
return curses.tparm(curses.tigetstr('op'))
def setup_logging(debug):
'''
Setup logging format and log level.
'''
if debug:
level = logging.DEBUG
else:
level = logging.INFO
logging.basicConfig(format="%(asctime)s %(levelname)8s %(message)s", datefmt="%T %Y/%m/%d", level=level)
def log_failure(s):
'''Convenience function for failure message.'''
logging.fatal("{}{}{}".format(term_color('red'), s, reset_term_color()))
def log_debug(s):
'''Convenience function for debugging message.'''
logging.debug("{}{}".format(s, reset_term_color()))
def log_warn(s):
'''Convenience function for warning message.'''
logging.warn("{}{}{}".format(term_color('red'), s, reset_term_color()))
def log_info(s):
'''Convenience function for info message.'''
logging.info("{}{}".format(s, reset_term_color()))
# decorate the "real" debugger entrypoint by
# disabling any SIGALRM invocations -- just ignore
# them if we're going into the debugger
import pdb
def setup_debugger(real_debugger):
def inner():
from switchy import disable_timer
disable_timer()
return real_debugger
return inner()
debugger = setup_debugger(pdb.set_trace)
# Copyright 2011,2013 James McCauley
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
The POX packet library for packet parsing and creation.
This is based heavily on NOX's packet library, though it has undergone
some signficant change, particularly with regard to making packet
assembly easier.
Could still use more work.
"""
# None of this is probably that big, and almost all of it gets loaded
# under most circumstances anyway. Let's just load all of it.
import arp as ARP
import dhcp as DHCP
import dns as DNS
import eap as EAP
import eapol as EAPOL
import ethernet as ETHERNET
import ipv4 as IPV4
import ipv6 as IPV6
import icmp as ICMP
import icmpv6 as ICMPV6
import lldp as LLDP
import tcp as TCP
import udp as UDP
import vlan as VLAN
import mpls as MPLS
from arp import *
from dhcp import *
from dns import *
from eap import *
from eapol import *
from ethernet import *
from ipv6 import *
from ipv4 import *
from icmpv6 import *
from icmp import *
from lldp import *
from tcp import *
from udp import *
from vlan import *
from mpls import *
__all__ = [
'arp',
'dhcp',
'dns',
'eap',
'eapol',
'ethernet',
'ipv4',
'ipv6',
'icmp',
'icmpv6',
'lldp',
'tcp',
'tcp_opt',
'udp',
'vlan',
'mpls',
'ARP',
'DHCP',
'DNS',
'EAP',
'EAPOL',
'ETHERNET',
'IPV4',
'IPV6',
'ICMP',
'ICMPV6',
'LLDP',
'TCP',
'UDP',
'VLAN',
'MPLS',
]
# Copyright 2011,2012,2013 James McCauley
# Copyright 2008 (C) Nicira, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# This file is derived from the packet library in NOX, which was
# developed by Nicira, Inc.
#======================================================================
# Ethernet header
#
#======================================================================
import struct
from packet_base import packet_base
from packet_utils import ethtype_to_str
from pox.lib.addresses import *
ETHER_ANY = EthAddr(b"\x00\x00\x00\x00\x00\x00")
ETHER_BROADCAST = EthAddr(b"\xff\xff\xff\xff\xff\xff")
BRIDGE_GROUP_ADDRESS = EthAddr(b"\x01\x80\xC2\x00\x00\x00")
LLDP_MULTICAST = EthAddr(b"\x01\x80\xc2\x00\x00\x0e")
PAE_MULTICAST = EthAddr(b'\x01\x80\xc2\x00\x00\x03') # 802.1x Port
# Access Entity
NDP_MULTICAST = EthAddr(b'\x01\x23\x20\x00\x00\x01') # Nicira discovery
# multicast
class ethernet(packet_base):
"Ethernet packet struct"
resolve_names = False
MIN_LEN = 14
IP_TYPE = 0x0800
ARP_TYPE = 0x0806
RARP_TYPE = 0x8035
VLAN_TYPE = 0x8100
LLDP_TYPE = 0x88cc
PAE_TYPE = 0x888e # 802.1x Port Access Entity
#MPLS_UNICAST_TYPE = 0x8847
#MPLS_MULTICAST_TYPE = 0x8848
MPLS_TYPE = 0x8847
MPLS_MC_TYPE = 0x8848 # Multicast
IPV6_TYPE = 0x86dd
PPP_TYPE = 0x880b
LWAPP_TYPE = 0x88bb
GSMP_TYPE = 0x880c
IPX_TYPE = 0x8137
IPX_TYPE = 0x8137
WOL_TYPE = 0x0842
TRILL_TYPE = 0x22f3
JUMBO_TYPE = 0x8870
SCSI_TYPE = 0x889a
ATA_TYPE = 0x88a2
QINQ_TYPE = 0x9100
INVALID_TYPE = 0xffff
type_parsers = {}
def __init__(self, raw=None, prev=None, **kw):
packet_base.__init__(self)
if len(ethernet.type_parsers) == 0:
from vlan import vlan
ethernet.type_parsers[ethernet.VLAN_TYPE] = vlan
from arp import arp
ethernet.type_parsers[ethernet.ARP_TYPE] = arp
ethernet.type_parsers[ethernet.RARP_TYPE] = arp
from ipv4 import ipv4
ethernet.type_parsers[ethernet.IP_TYPE] = ipv4
from ipv6 import ipv6
ethernet.type_parsers[ethernet.IPV6_TYPE] = ipv6
from lldp import lldp
ethernet.type_parsers[ethernet.LLDP_TYPE] = lldp
from eapol import eapol
ethernet.type_parsers[ethernet.PAE_TYPE] = eapol
from mpls import mpls
ethernet.type_parsers[ethernet.MPLS_TYPE] = mpls
ethernet.type_parsers[ethernet.MPLS_MC_TYPE] = mpls
from llc import llc
ethernet._llc = llc
self.prev = prev
self.dst = ETHER_ANY
self.src = ETHER_ANY
self.type = 0
self.next = b''
if raw is not None:
self.parse(raw)
self._init(kw)
def parse (self, raw):
assert isinstance(raw, bytes)
self.next = None # In case of unfinished parsing
self.raw = raw
alen = len(raw)
if alen < ethernet.MIN_LEN:
self.msg('warning eth packet data too short to parse header: data len %u'
% (alen,))
return
self.dst = EthAddr(raw[:6])
self.src = EthAddr(raw[6:12])
self.type = struct.unpack('!H', raw[12:ethernet.MIN_LEN])[0]
self.hdr_len = ethernet.MIN_LEN
self.payload_len = alen - self.hdr_len
self.next = ethernet.parse_next(self, self.type, raw, ethernet.MIN_LEN)
self.parsed = True
@staticmethod
def parse_next (prev, typelen, raw, offset=0, allow_llc=True):
parser = ethernet.type_parsers.get(typelen)
if parser is not None:
return parser(raw[offset:], prev)
elif typelen < 1536 and allow_llc:
return ethernet._llc(raw[offset:], prev)
else:
return raw[offset:]
@staticmethod
def getNameForType (ethertype):
""" Returns a string name for a numeric ethertype """
return ethtype_to_str(ethertype)
@property
def effective_ethertype (self):
return self._get_effective_ethertype(self)
@staticmethod
def _get_effective_ethertype (self):
"""
Get the "effective" ethertype of a packet.
This means that if the payload is something like a VLAN or SNAP header,
we want the type from that deeper header. This is kind of ugly here in
the packet library, but it should make user code somewhat simpler.
"""
if not self.parsed:
return ethernet.INVALID_TYPE
if self.type == ethernet.VLAN_TYPE or type(self.payload) == ethernet._llc:
try:
return self.payload.effective_ethertype
except:
return ethernet.INVALID_TYPE
return self.type
def _to_str(self):
s = ''.join(('[',str(EthAddr(self.src)),'>',str(EthAddr(self.dst)),' ',
ethernet.getNameForType(self.type),']'))
return s
def hdr(self, payload):
dst = self.dst
src = self.src
if type(dst) is EthAddr:
dst = dst.toRaw()
if type(src) is EthAddr:
src = src.toRaw()
return struct.pack('!6s6sH', dst, src, self.type)
import sys
import os
import os.path
import unittest
import copy
import time
sys.path.append(os.path.join(os.environ['HOME'],'pox'))
sys.path.append(os.path.join(os.getcwd(),'pox'))
from switchy import PacketMatcher,setup_logging,ScenarioFailure
import pox.lib.packet as pktlib
from pox.lib.packet import ethernet,ETHER_BROADCAST,ipv4,packet_base
from pox.lib.packet import arp
from pox.lib.addresses import EthAddr,IPAddr
class SrpyMatcherTest(unittest.TestCase):
def testExactMatch1(self):
pkt = ethernet()
matcher = PacketMatcher(pkt, exact=True)
self.assertTrue(matcher.match(pkt))
def testExactMatch2(self):
pkt = ethernet()
matcher = PacketMatcher(pkt, exact=True)
pkt.type = pkt.IP_TYPE
self.assertRaises(ScenarioFailure, matcher.match, pkt)
def testOFPMatch(self):
pkt = ethernet()
matcher = PacketMatcher(pkt, exact=False)
self.assertTrue(matcher.match(pkt))
def testPredicateMatch1(self):
pkt = ethernet()
matcher = PacketMatcher(pkt, '''lambda eth: str(eth.src) == '00:00:00:00:00:00' ''', exact=False)
self.assertTrue(matcher.match(pkt))
def testPredicateMatch2(self):
pkt = ethernet()
matcher = PacketMatcher(pkt, '''lambda eth: str(eth.src) == '00:00:00:00:00:01' ''', exact=False)
self.assertRaises(ScenarioFailure, matcher.match, pkt)
def testPredicateMatch3(self):
pkt = ethernet()
ip = ipv4()
pkt.payload = ip
pkt.type = pkt.IP_TYPE
matcher = PacketMatcher(pkt,
'''lambda eth: str(eth.src) == '00:00:00:00:00:00' ''',
'''lambda eth: isinstance(eth.next, ipv4) and eth.next.ttl > 0 ''',
exact=False)
self.assertTrue(matcher.match(pkt))
def testWildcarding(self):
pkt = ethernet()
ip = ipv4()
ip.srcip = IPAddr("192.168.1.1")
ip.dstip = IPAddr("192.168.1.2")
ip.ttl = 64
pkt.payload = ip
pkt.type = pkt.IP_TYPE
matcher = PacketMatcher(pkt, wildcard=['dl_dst', 'nw_dst'], exact=False)
pkt.dst = EthAddr("11:11:11:11:11:11")
ip.dstip = IPAddr("192.168.1.3")
self.assertTrue(matcher.match(copy.deepcopy(pkt)))
if __name__ == '__main__':
setup_logging(False)
unittest.main(verbosity=2)
import sys
import os
import os.path
import unittest
import copy
import time
sys.path.append(os.path.join(os.environ['HOME'],'pox'))
sys.path.append(os.path.join(os.getcwd(),'pox'))
from switchy_common import ScenarioFailure, setup_logging
from switchy import Scenario,PacketInputEvent,PacketOutputEvent,compile_scenario,uncompile_scenario,get_test_scenario_from_file
import pox.lib.packet as pktlib
from pox.lib.packet import ethernet,ETHER_BROADCAST
from pox.lib.packet import arp
from pox.lib.addresses import EthAddr,IPAddr
class SrpyCompileTest(unittest.TestCase):
CONTENTS1 = '''
from switchy import Scenario,PacketInputEvent,PacketOutputEvent
import pox.lib.packet as pktlib
from pox.lib.packet import ethernet,ETHER_BROADCAST
from pox.lib.packet import arp
from pox.lib.addresses import EthAddr,IPAddr
s = Scenario("ARP request")
s.add_interface('router-eth0', '40:00:00:00:00:00', '192.168.1.1', '255.255.255.0')
s.add_interface('router-eth1', '40:00:00:00:00:01', '192.168.100.1', '255.255.255.0')
s.add_interface('router-eth2', '40:00:00:00:00:02', '10.0.1.2', '255.255.255.0')
s.add_interface('router-eth3', '40:00:00:00:00:03', '10.1.1.2', '255.255.255.0')
# arp coming from client
arp_req = arp()
arp_req.opcode = arp.REQUEST
arp_req.protosrc = IPAddr("10.1.1.1")
arp_req.protodst = IPAddr("10.1.1.2")
arp_req.hwsrc = EthAddr("30:00:00:00:00:01")
arp_req.hwdst = EthAddr("ff:ff:ff:ff:ff:ff")
ether = ethernet()
ether.dst = EthAddr("ff:ff:ff:ff:ff:ff")
ether.src = EthAddr("30:00:00:00:00:01")
ether.type = ethernet.ARP_TYPE
ether.set_payload(arp_req)
s.expect(PacketInputEvent("router-eth3", ether), "Incoming ARP request")
ether = ethernet()
ether.src = EthAddr("40:00:00:00:00:03")
ether.dst = EthAddr("30:00:00:00:00:01")
ether.type = ethernet.ARP_TYPE
arp_reply = arp()
arp_reply.opcode = arp.REPLY
arp_reply.protosrc = IPAddr("10.1.1.2")
arp_reply.protodst = IPAddr("10.1.1.1")
arp_reply.hwsrc = EthAddr("40:00:00:00:00:03")
arp_reply.hwdst = EthAddr("30:00:00:00:00:01")
ether.set_payload(arp_reply)
s.expect(PacketOutputEvent("router-eth3", ether), "Outgoing ARP reply")
scenario = s
'''
def setUp(self):
self.writeScenario1('stest.py', SrpyCompileTest.CONTENTS1)
self.scenario = get_test_scenario_from_file('stest.py')
def tearDown(self):
self.removeScenario('stest')
def writeScenario1(self, name, contents):
outfile = open(name, 'w')
outfile.write(contents)
outfile.close()