#!/usr/bin/env python # -*- coding: utf-8 -*- # # Ethernet over WebSocket # # depends on: # - python-2.7.2 # - python-pytun-0.2 # - websocket-client-0.7.0 # - tornado-2.3 # # =========================================================================== # Copyright (c) 2012, Atzm WATANABE # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # # 1. Redistributions of source code must retain the above copyright notice, # this list of conditions and the following disclaimer. # 2. Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. # =========================================================================== # # $Id$ import os import sys import ssl import time import json import fcntl import base64 import urllib2 import hashlib import getpass import argparse import traceback import tornado import websocket from tornado.web import Application, RequestHandler from tornado.websocket import WebSocketHandler from tornado.httpserver import HTTPServer from tornado.ioloop import IOLoop from pytun import TunTapDevice, IFF_TAP, IFF_NO_PI class DebugMixIn(object): def dprintf(self, msg, func=lambda: ()): if self._debug: prefix = '[%s] %s - ' % (time.asctime(), self.__class__.__name__) sys.stderr.write(prefix + (msg % func())) class EthernetFrame(object): def __init__(self, data): self.data = data @property def dst_multicast(self): return ord(self.data[0]) & 1 @property def src_multicast(self): return ord(self.data[6]) & 1 @property def dst_mac(self): return self.data[:6] @property def src_mac(self): return self.data[6:12] @property def tagged(self): return ord(self.data[12]) == 0x81 and ord(self.data[13]) == 0 @property def vid(self): if self.tagged: return ((ord(self.data[14]) << 8) | ord(self.data[15])) & 0x0fff return 0 class FDB(DebugMixIn): def __init__(self, ageout, debug=False): self._ageout = ageout self._debug = debug self._dict = {} def lookup(self, frame): mac = frame.dst_mac vid = frame.vid group = self._dict.get(vid) if not group: return None entry = group.get(mac) if not entry: return None if time.time() - entry['time'] > self._ageout: port = self._dict[vid][mac]['port'] del self._dict[vid][mac] if not self._dict[vid]: del self._dict[vid] self.dprintf('aged out: port:%d; vid:%d; mac:%s\n', lambda: (port.number, vid, mac.encode('hex'))) return None return entry['port'] def learn(self, port, frame): mac = frame.src_mac vid = frame.vid if vid not in self._dict: self._dict[vid] = {} self._dict[vid][mac] = {'time': time.time(), 'port': port} self.dprintf('learned: port:%d; vid:%d; mac:%s\n', lambda: (port.number, vid, mac.encode('hex'))) def delete(self, port): for vid in self._dict.keys(): for mac in self._dict[vid].keys(): if self._dict[vid][mac]['port'].number == port.number: del self._dict[vid][mac] self.dprintf('deleted: port:%d; vid:%d; mac:%s\n', lambda: (port.number, vid, mac.encode('hex'))) if not self._dict[vid]: del self._dict[vid] class SwitchPort(object): def __init__(self, number, interface): self.number = number self.interface = interface self.tx = 0 self.rx = 0 self.shut = False @staticmethod def cmp_by_number(x, y): return cmp(x.number, y.number) class SwitchingHub(DebugMixIn): def __init__(self, fdb, debug=False): self._fdb = fdb self._debug = debug self._table = {} self._next = 1 @property def portlist(self): return sorted(self._table.itervalues(), cmp=SwitchPort.cmp_by_number) def get_port(self, portnum): return self._table[portnum] def register_port(self, interface): try: self._set_privattr('portnum', interface, self._next) # XXX self._table[self._next] = SwitchPort(self._next, interface) return self._next finally: self._next += 1 def unregister_port(self, interface): portnum = self._get_privattr('portnum', interface) self._del_privattr('portnum', interface) self._fdb.delete(self._table[portnum]) del self._table[portnum] def send(self, dst_interfaces, frame): portnums = (self._get_privattr('portnum', i) for i in dst_interfaces) ports = (self._table[n] for n in portnums) ports = (p for p in ports if not p.shut) ports = sorted(ports, cmp=SwitchPort.cmp_by_number) for p in ports: p.interface.write_message(frame.data, True) p.tx += 1 if ports: self.dprintf('sent: port:%s; vid:%d; %s -> %s\n', lambda: (','.join(str(p.number) for p in ports), frame.vid, frame.src_mac.encode('hex'), frame.dst_mac.encode('hex'))) def receive(self, src_interface, frame): port = self._table[self._get_privattr('portnum', src_interface)] if not port.shut: port.rx += 1 self._forward(port, frame) def _forward(self, src_port, frame): try: if not frame.src_multicast: self._fdb.learn(src_port, frame) if not frame.dst_multicast: dst_port = self._fdb.lookup(frame) if dst_port: self.send([dst_port.interface], frame) return ports = set(self.portlist) - set([src_port]) self.send((p.interface for p in ports), frame) except: # ex. received invalid frame traceback.print_exc() def _privattr(self, name): return '_%s_%s_%s' % (self.__class__.__name__, id(self), name) def _set_privattr(self, name, obj, value): return setattr(obj, self._privattr(name), value) def _get_privattr(self, name, obj, defaults=None): return getattr(obj, self._privattr(name), defaults) def _del_privattr(self, name, obj): return delattr(obj, self._privattr(name)) class Htpasswd(object): def __init__(self, path): self._path = path self._stat = None self._data = {} def auth(self, name, passwd): passwd = base64.b64encode(hashlib.sha1(passwd).digest()) return self._data.get(name) == passwd def load(self): old_stat = self._stat with open(self._path) as fp: fileno = fp.fileno() fcntl.flock(fileno, fcntl.LOCK_SH | fcntl.LOCK_NB) self._stat = os.fstat(fileno) unchanged = old_stat and \ old_stat.st_ino == self._stat.st_ino and \ old_stat.st_dev == self._stat.st_dev and \ old_stat.st_mtime == self._stat.st_mtime if not unchanged: self._data = self._parse(fp) return self def _parse(self, fp): data = {} for line in fp: line = line.strip() if 0 <= line.find(':'): name, passwd = line.split(':', 1) if passwd.startswith('{SHA}'): data[name] = passwd[5:] return data class BasicAuthMixIn(object): def _execute(self, transforms, *args, **kwargs): def do_execute(): sp = super(BasicAuthMixIn, self) return sp._execute(transforms, *args, **kwargs) def auth_required(): stream = getattr(self, 'stream', self.request.connection.stream) stream.write(tornado.escape.utf8( 'HTTP/1.1 401 Authorization Required\r\n' 'WWW-Authenticate: Basic realm=etherws\r\n\r\n' )) stream.close() try: if not self._htpasswd: return do_execute() creds = self.request.headers.get('Authorization') if not creds or not creds.startswith('Basic '): return auth_required() name, passwd = base64.b64decode(creds[6:]).split(':', 1) if self._htpasswd.load().auth(name, passwd): return do_execute() except: traceback.print_exc() return auth_required() class EtherWebSocketHandler(DebugMixIn, BasicAuthMixIn, WebSocketHandler): IFTYPE = 'server' def __init__(self, app, req, switch, htpasswd=None, debug=False): super(EtherWebSocketHandler, self).__init__(app, req) self._switch = switch self._htpasswd = htpasswd self._debug = debug def get_target(self): return self.request.remote_ip def open(self): try: return self._switch.register_port(self) finally: self.dprintf('connected: %s\n', lambda: self.request.remote_ip) def on_message(self, message): self._switch.receive(self, EthernetFrame(message)) def on_close(self): self._switch.unregister_port(self) self.dprintf('disconnected: %s\n', lambda: self.request.remote_ip) class TapHandler(DebugMixIn): IFTYPE = 'tap' READ_SIZE = 65535 def __init__(self, ioloop, switch, dev, debug=False): self._ioloop = ioloop self._switch = switch self._dev = dev self._debug = debug self._tap = None def get_target(self): if self.closed: return self._dev return self._tap.name @property def closed(self): return not self._tap def open(self): if not self.closed: raise ValueError('already opened') self._tap = TunTapDevice(self._dev, IFF_TAP | IFF_NO_PI) self._tap.up() self._ioloop.add_handler(self.fileno(), self, self._ioloop.READ) return self._switch.register_port(self) def close(self): if self.closed: raise ValueError('I/O operation on closed tap') self._switch.unregister_port(self) self._ioloop.remove_handler(self.fileno()) self._tap.close() self._tap = None def fileno(self): if self.closed: raise ValueError('I/O operation on closed tap') return self._tap.fileno() def write_message(self, message, binary=False): if self.closed: raise ValueError('I/O operation on closed tap') self._tap.write(message) def __call__(self, fd, events): try: self._switch.receive(self, EthernetFrame(self._read())) return except: traceback.print_exc() self.close() def _read(self): if self.closed: raise ValueError('I/O operation on closed tap') buf = [] while True: buf.append(self._tap.read(self.READ_SIZE)) if len(buf[-1]) < self.READ_SIZE: break return ''.join(buf) class EtherWebSocketClient(DebugMixIn): IFTYPE = 'client' def __init__(self, ioloop, switch, url, ssl_=None, cred=None, debug=False): self._ioloop = ioloop self._switch = switch self._url = url self._ssl = ssl_ self._debug = debug self._sock = None self._options = {} if isinstance(cred, dict) and cred['user'] and cred['passwd']: token = base64.b64encode('%s:%s' % (cred['user'], cred['passwd'])) auth = ['Authorization: Basic %s' % token] self._options['header'] = auth def get_target(self): return self._url @property def closed(self): return not self._sock def open(self): sslwrap = websocket._SSLSocketWrapper if not self.closed: raise websocket.WebSocketException('already opened') if self._ssl: websocket._SSLSocketWrapper = self._ssl try: self._sock = websocket.WebSocket() self._sock.connect(self._url, **self._options) self._ioloop.add_handler(self.fileno(), self, self._ioloop.READ) return self._switch.register_port(self) finally: websocket._SSLSocketWrapper = sslwrap self.dprintf('connected: %s\n', lambda: self._url) def close(self): if self.closed: raise websocket.WebSocketException('already closed') self._switch.unregister_port(self) self._ioloop.remove_handler(self.fileno()) self._sock.close() self._sock = None self.dprintf('disconnected: %s\n', lambda: self._url) def fileno(self): if self.closed: raise websocket.WebSocketException('closed socket') return self._sock.io_sock.fileno() def write_message(self, message, binary=False): if self.closed: raise websocket.WebSocketException('closed socket') if binary: flag = websocket.ABNF.OPCODE_BINARY else: flag = websocket.ABNF.OPCODE_TEXT self._sock.send(message, flag) def __call__(self, fd, events): try: data = self._sock.recv() if data is not None: self._switch.receive(self, EthernetFrame(data)) return except: traceback.print_exc() self.close() class EtherWebSocketControlHandler(DebugMixIn, BasicAuthMixIn, RequestHandler): NAMESPACE = 'etherws.control' IFTYPES = { TapHandler.IFTYPE: TapHandler, EtherWebSocketClient.IFTYPE: EtherWebSocketClient, } def __init__(self, app, req, ioloop, switch, htpasswd=None, debug=False): super(EtherWebSocketControlHandler, self).__init__(app, req) self._ioloop = ioloop self._switch = switch self._htpasswd = htpasswd self._debug = debug def post(self): id_ = None try: req = json.loads(self.request.body) method = req['method'] params = req['params'] id_ = req.get('id') if not method.startswith(self.NAMESPACE + '.'): raise ValueError('invalid method: %s' % method) if not isinstance(params, list): raise ValueError('invalid params: %s' % params) handler = 'handle_' + method[len(self.NAMESPACE) + 1:] result = getattr(self, handler)(params) self.finish({'result': result, 'error': None, 'id': id_}) except Exception as e: traceback.print_exc() msg = '%s: %s' % (e.__class__.__name__, str(e)) self.finish({'result': None, 'error': {'message': msg}, 'id': id_}) def handle_listPort(self, params): list_ = [self._portstat(p) for p in self._switch.portlist] return {'portlist': list_} def handle_addPort(self, params): list_ = [] for p in params: type_ = p['type'] target = p['target'] options = getattr(self, '_optparse_' + type_)(p.get('options', {})) klass = self.IFTYPES[type_] interface = klass(self._ioloop, self._switch, target, **options) portnum = interface.open() list_.append(self._portstat(self._switch.get_port(portnum))) return {'portlist': list_} def handle_delPort(self, params): list_ = [] for p in params: port = self._switch.get_port(int(p['port'])) list_.append(self._portstat(port)) port.interface.close() return {'portlist': list_} def handle_shutPort(self, params): list_ = [] for p in params: port = self._switch.get_port(int(p['port'])) port.shut = bool(p['shut']) list_.append(self._portstat(port)) return {'portlist': list_} def _optparse_tap(self, opt): return {'debug': self._debug} def _optparse_client(self, opt): args = {'cert_reqs': ssl.CERT_REQUIRED, 'ca_certs': opt.get('cacerts')} if opt.get('insecure'): args = {} ssl_ = lambda sock: ssl.wrap_socket(sock, **args) cred = {'user': opt.get('user'), 'passwd': opt.get('passwd')} return {'ssl_': ssl_, 'cred': cred, 'debug': self._debug} @staticmethod def _portstat(port): return { 'port': port.number, 'type': port.interface.IFTYPE, 'target': port.interface.get_target(), 'tx': port.tx, 'rx': port.rx, 'shut': port.shut, } def start_sw(args): def daemonize(nochdir=False, noclose=False): if os.fork() > 0: sys.exit(0) os.setsid() if os.fork() > 0: sys.exit(0) if not nochdir: os.chdir('/') if not noclose: os.umask(0) sys.stdin.close() sys.stdout.close() sys.stderr.close() os.close(0) os.close(1) os.close(2) sys.stdin = open(os.devnull) sys.stdout = open(os.devnull, 'a') sys.stderr = open(os.devnull, 'a') def checkabspath(ns, path): val = getattr(ns, path, '') if not val.startswith('/'): raise ValueError('invalid %: %s' % (path, val)) def getsslopt(ns, key, cert): kval = getattr(ns, key, None) cval = getattr(ns, cert, None) if kval and cval: return {'keyfile': kval, 'certfile': cval} elif kval or cval: raise ValueError('both %s and %s are required' % (key, cert)) return None def setrealpath(ns, *keys): for k in keys: v = getattr(ns, k, None) if v is not None: v = os.path.realpath(v) open(v).close() # check readable setattr(ns, k, v) def setport(ns, port, isssl): val = getattr(ns, port, None) if val is None: if isssl: return setattr(ns, port, 443) return setattr(ns, port, 80) if not (0 <= val <= 65535): raise ValueError('invalid %s: %s' % (port, val)) def sethtpasswd(ns, htpasswd): val = getattr(ns, htpasswd, None) if val: return setattr(ns, htpasswd, Htpasswd(val)) #if args.debug: # websocket.enableTrace(True) if args.ageout <= 0: raise ValueError('invalid ageout: %s' % args.ageout) setrealpath(args, 'htpasswd', 'sslkey', 'sslcert') setrealpath(args, 'ctlhtpasswd', 'ctlsslkey', 'ctlsslcert') checkabspath(args, 'path') checkabspath(args, 'ctlpath') sslopt = getsslopt(args, 'sslkey', 'sslcert') ctlsslopt = getsslopt(args, 'ctlsslkey', 'ctlsslcert') setport(args, 'port', sslopt) setport(args, 'ctlport', ctlsslopt) sethtpasswd(args, 'htpasswd') sethtpasswd(args, 'ctlhtpasswd') ioloop = IOLoop.instance() fdb = FDB(ageout=args.ageout, debug=args.debug) switch = SwitchingHub(fdb, debug=args.debug) if args.port == args.ctlport and args.host == args.ctlhost: if args.path == args.ctlpath: raise ValueError('same path/ctlpath on same host') if args.sslkey != args.ctlsslkey: raise ValueError('different sslkey/ctlsslkey on same host') if args.sslcert != args.ctlsslcert: raise ValueError('different sslcert/ctlsslcert on same host') app = Application([ (args.path, EtherWebSocketHandler, { 'switch': switch, 'htpasswd': args.htpasswd, 'debug': args.debug, }), (args.ctlpath, EtherWebSocketControlHandler, { 'ioloop': ioloop, 'switch': switch, 'htpasswd': args.ctlhtpasswd, 'debug': args.debug, }), ]) server = HTTPServer(app, ssl_options=sslopt) server.listen(args.port, address=args.host) else: app = Application([(args.path, EtherWebSocketHandler, { 'switch': switch, 'htpasswd': args.htpasswd, 'debug': args.debug, })]) server = HTTPServer(app, ssl_options=sslopt) server.listen(args.port, address=args.host) ctl = Application([(args.ctlpath, EtherWebSocketControlHandler, { 'ioloop': ioloop, 'switch': switch, 'htpasswd': args.ctlhtpasswd, 'debug': args.debug, })]) ctlserver = HTTPServer(ctl, ssl_options=ctlsslopt) ctlserver.listen(args.ctlport, address=args.ctlhost) if not args.foreground: daemonize() ioloop.start() def start_ctl(args): import yaml def request(args, method, params): method = '.'.join([EtherWebSocketControlHandler.NAMESPACE, method]) data = json.dumps({'method': method, 'params': params}) req = urllib2.Request(args.ctlurl) req.add_header('Content-type', 'application/json') if args.ctluser: if not args.ctlpasswd: args.ctlpasswd = getpass.getpass() token = base64.b64encode('%s:%s' % (args.ctluser, args.ctlpasswd)) req.add_header('Authorization', 'Basic %s' % token) return json.loads(urllib2.urlopen(req, data).read()) def handle_ctl_addport(args): params = [{ 'type': args.type, 'target': args.target, 'options': { 'insecure': args.insecure, 'cacerts': args.cacerts, 'user': args.user, 'passwd': args.passwd, } }] return request(args, 'addPort', params) def handle_ctl_shutport(args): if args.port <= 0: raise ValueError('invalid port: %d' % args.port) params = [{'port': args.port, 'shut': args.no}] return request(args, 'shutPort', params) def handle_ctl_delport(args): if args.port <= 0: raise ValueError('invalid port: %d' % args.port) params = [{'port': args.port}] return request(args, 'delPort', params) def handle_ctl_listport(args): return request(args, 'listPort', []) res = locals()['handle_ctl_' + args.control_method](args) if res['error']: print(res['error']['message']) else: print(yaml.safe_dump(res['result']['portlist']).strip()) def main(): parser = argparse.ArgumentParser() subcommand = parser.add_subparsers(dest='subcommand') # -- sw command parser parser_s = subcommand.add_parser('sw') parser_s.add_argument('--debug', action='store_true', default=False) parser_s.add_argument('--foreground', action='store_true', default=False) parser_s.add_argument('--ageout', type=int, default=300) parser_s.add_argument('--path', default='/') parser_s.add_argument('--host', default='') parser_s.add_argument('--port', type=int) parser_s.add_argument('--htpasswd') parser_s.add_argument('--sslkey') parser_s.add_argument('--sslcert') parser_s.add_argument('--ctlpath', default='/ctl') parser_s.add_argument('--ctlhost', default='') parser_s.add_argument('--ctlport', type=int) parser_s.add_argument('--ctlhtpasswd') parser_s.add_argument('--ctlsslkey') parser_s.add_argument('--ctlsslcert') # -- ctl command parser parser_c = subcommand.add_parser('ctl') parser_c.add_argument('--ctlurl', default='http://localhost/ctl') parser_c.add_argument('--ctluser') parser_c.add_argument('--ctlpasswd') control_method = parser_c.add_subparsers(dest='control_method') parser_c_ap = control_method.add_parser('addport') parser_c_ap.add_argument( 'type', choices=EtherWebSocketControlHandler.IFTYPES.keys()) parser_c_ap.add_argument('target') parser_c_ap.add_argument('--insecure', action='store_true', default=False) parser_c_ap.add_argument('--cacerts') parser_c_ap.add_argument('--user') parser_c_ap.add_argument('--passwd') parser_c_sp = control_method.add_parser('shutport') parser_c_sp.add_argument('port', type=int) parser_c_sp.add_argument('--no', action='store_false', default=True) parser_c_dp = control_method.add_parser('delport') parser_c_dp.add_argument('port', type=int) parser_c_lp = control_method.add_parser('listport') # -- go args = parser.parse_args() globals()['start_' + args.subcommand](args) if __name__ == '__main__': main()