Changeset 183

Timestamp:

07/29/12 04:11:13 (12 years ago)

Author:

atzm

Message:

• global change: enables remote control

Location:

etherws/trunk

Files:

• etherws.py (modified) (17 diffs)
• setup.py (modified) (1 diff)

etherws/trunk/etherws.py

@@ -8 +8 @@
 #   - python-pytun-0.2
 #   - websocket-client-0.7.0
-#   - tornado-2.2.1
+#   - tornado-2.3
 #
 # todo:
@@ -45 +45 @@
 import ssl
 import time
+import json
 import fcntl
 import base64
@@ -53 +54 @@
 
 import websocket
-import tornado.web
-import tornado.ioloop
-import tornado.httpserver
-
+
+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
 
@@ -96 +98 @@
         if self.tagged:
             return ((ord(self.data[14]) << 8) | ord(self.data[15])) & 0x0fff
-        return -1
+        return 0
 
 
@@ -118 +120 @@
 
         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: [%d] %s\n',
-                         lambda: (vid, mac.encode('hex')))
+            self.dprintf('aged out: port:%d; vid:%d; mac:%s\n',
+                         lambda: (port.number, vid, mac.encode('hex')))
             return None
 
@@ -135 +138 @@
 
         self._dict[vid][mac] = {'time': time.time(), 'port': port}
-        self.dprintf('learned: [%d] %s\n',
-                     lambda: (vid, mac.encode('hex')))
+        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'] is port:
+                if self._dict[vid][mac]['port'].number == port.number:
                     del self._dict[vid][mac]
-                    self.dprintf('deleted: [%d] %s\n',
-                                 lambda: (vid, mac.encode('hex')))
+                    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)
 
 
@@ -153 +169 @@
         self._fdb = fdb
         self._debug = debug
-        self._ports = []
-
-    def register_port(self, port):
-        self._ports.append(port)
-
-    def unregister_port(self, port):
-        self._fdb.delete(port)
-        self._ports.remove(port)
-
-    def forward(self, src_port, frame):
+        self._table = {}
+        self._next = 1
+
+    @property
+    def portlist(self):
+        return sorted(self._table.itervalues(), cmp=SwitchPort.cmp_by_number)
+
+    def shut_port(self, portnum, flag=True):
+        self._table[portnum].shut = flag
+
+    def get_port(self, portnum):
+        return self._table[portnum]
+
+    def register_port(self, interface):
+        interface._switch_portnum = self._next  # XXX
+        self._table[self._next] = SwitchPort(self._next, interface)
+        self._next += 1
+
+    def unregister_port(self, interface):
+        self._fdb.delete(self._table[interface._switch_portnum])
+        del self._table[interface._switch_portnum]
+        del interface._switch_portnum
+
+    def send(self, dst_interfaces, frame):
+        ports = sorted((self._table[i._switch_portnum] for i in dst_interfaces
+                        if not self._table[i._switch_portnum].shut),
+                       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[src_interface._switch_portnum]
+
+        if not port.shut:
+            port.rx += 1
+            self._forward(port, frame)
+
+    def _forward(self, src_port, frame):
         try:
             if not frame.src_multicast:
@@ -171 +224 @@
 
                 if dst_port:
-                    self._unicast(frame, dst_port)
+                    self.send([dst_port.interface], frame)
                     return
 
-            self._broadcast(frame, src_port)
+            ports = set(self._table.itervalues()) - set([src_port])
+            self.send((p.interface for p in ports), frame)
 
         except:  # ex. received invalid frame
             traceback.print_exc()
-
-    def _unicast(self, frame, port):
-        port.write_message(frame.data, True)
-        self.dprintf('sent unicast: [%d] %s -> %s\n',
-                     lambda: (frame.vid,
-                              frame.src_mac.encode('hex'),
-                              frame.dst_mac.encode('hex')))
-
-    def _broadcast(self, frame, *except_ports):
-        ports = self._ports[:]
-        for port in except_ports:
-            ports.remove(port)
-        for port in ports:
-            port.write_message(frame.data, True)
-        self.dprintf('sent broadcast: [%d] %s -> %s\n',
-                     lambda: (frame.vid,
-                              frame.src_mac.encode('hex'),
-                              frame.dst_mac.encode('hex')))
 
 
@@ -283 +319 @@
         return not self._tap
 
+    def get_type(self):
+        return 'tap'
+
+    def get_name(self):
+        if self.closed:
+            return self._dev
+        return self._tap.name
+
     def open(self):
         if not self.closed:
@@ -311 +355 @@
     def __call__(self, fd, events):
         try:
-            self._switch.forward(self, EthernetFrame(self._read()))
+            self._switch.receive(self, EthernetFrame(self._read()))
             return
         except:
@@ -338 +382 @@
             self._htpasswd = Htpasswd(self._htpasswd)
 
+    def get_type(self):
+        return 'server'
+
+    def get_name(self):
+        return self.request.remote_ip
+
     def open(self):
         self._switch.register_port(self)
@@ -343 +393 @@
 
     def on_message(self, message):
-        self._switch.forward(self, EthernetFrame(message))
+        self._switch.receive(self, EthernetFrame(message))
 
     def on_close(self):
@@ -369 +419 @@
         return not self._sock
 
+    def get_type(self):
+        return 'client'
+
+    def get_name(self):
+        return self._url
+
     def open(self):
         sslwrap = websocket._SSLSocketWrapper
@@ -414 +470 @@
             data = self._sock.recv()
             if data is not None:
-                self._switch.forward(self, EthernetFrame(data))
+                self._switch.receive(self, EthernetFrame(data))
                 return
         except:
             traceback.print_exc()
         self.close()
+
+
+class EtherWebSocketControlHandler(DebugMixIn, BasicAuthMixIn, RequestHandler):
+    NAMESPACE = 'etherws.control'
+
+    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()
+            self.finish({'result': None, 'error': str(e), 'id': id_})
+
+    def handle_listPort(self, params):
+        list_ = []
+        for port in self._switch.portlist:
+            list_.append({
+                'port': port.number,
+                'type': port.interface.get_type(),
+                'name': port.interface.get_name(),
+                'tx':   port.tx,
+                'rx':   port.rx,
+                'shut': port.shut,
+            })
+        return {'portlist': list_}
+
+    def handle_addPort(self, params):
+        for p in params:
+            getattr(self, '_openport_' + p['type'])(p)
+        return self.handle_listPort(params)
+
+    def handle_delPort(self, params):
+        for p in params:
+            self._switch.get_port(int(p['port'])).interface.close()
+        return self.handle_listPort(params)
+
+    def handle_shutPort(self, params):
+        for p in params:
+            self._switch.shut_port(int(p['port']), bool(p['flag']))
+        return self.handle_listPort(params)
+
+    def _openport_tap(self, p):
+        dev = p['device']
+        tap = TapHandler(self._ioloop, self._switch, dev, debug=self._debug)
+        tap.open()
+
+    def _openport_client(self, p):
+        ssl_ = self._ssl_wrapper(p.get('insecure'), p.get('cacerts'))
+        cred = {'user': p.get('user'), 'passwd': p.get('passwd')}
+        url = p['url']
+        client = EtherWebSocketClient(self._ioloop, self._switch,
+                                      url, ssl_, cred, self._debug)
+        client.open()
+
+    @staticmethod
+    def _ssl_wrapper(insecure, ca_certs):
+        args = {'cert_reqs': ssl.CERT_REQUIRED, 'ca_certs': ca_certs}
+        if insecure:
+            args = {}
+        return lambda sock: ssl.wrap_socket(sock, **args)
 
 
@@ -446 +584 @@
 
 
-def realpath(ns, *keys):
-    for k in keys:
-        v = getattr(ns, k, None)
-        if v is not None:
-            v = os.path.realpath(v)
-            setattr(ns, k, v)
-            open(v).close()  # check readable
-    return ns
-
-
-def ssl_wrapper(insecure, ca_certs):
-    args = {'cert_reqs': ssl.CERT_REQUIRED, 'ca_certs': ca_certs}
-    if insecure:
-        args = {}
-    return lambda sock: ssl.wrap_socket(sock, **args)
-
-
-def server_main(args):
-    realpath(args, 'keyfile', 'certfile', 'htpasswd')
-
-    if args.keyfile and args.certfile:
-        ssl_options = {'keyfile': args.keyfile, 'certfile': args.certfile}
-    elif args.keyfile or args.certfile:
-        raise ValueError('both keyfile and certfile are required')
+def main():
+    def realpath(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)
+        return ns
+
+    parser = argparse.ArgumentParser()
+
+    parser.add_argument('--debug', action='store_true', default=False)
+    parser.add_argument('--foreground', action='store_true', default=False)
+    parser.add_argument('--ageout', action='store', type=int, default=300)
+
+    parser.add_argument('--path', action='store', default='/')
+    parser.add_argument('--address', action='store', default='')
+    parser.add_argument('--port', action='store', type=int)
+    parser.add_argument('--htpasswd', action='store')
+    parser.add_argument('--sslkey', action='store')
+    parser.add_argument('--sslcert', action='store')
+
+    parser.add_argument('--ctlpath', action='store', default='/ctl')
+    parser.add_argument('--ctladdress', action='store', default='127.0.0.1')
+    parser.add_argument('--ctlport', action='store', type=int, default=7867)
+
+    args = realpath(parser.parse_args(), 'htpasswd', 'sslkey', 'sslcert')
+
+    #if args.debug:
+    #    websocket.enableTrace(True)
+
+    if args.ageout <= 0:
+        raise ValueError('invalid ageout: %s' % args.ageout)
+
+    if not args.path.startswith('/'):
+        raise ValueError('invalid path: %s' % args.path)
+
+    if not args.ctlpath.startswith('/'):
+        raise ValueError('invalid ctlpath: %s' % args.ctlpath)
+
+    if args.sslkey and args.sslcert:
+        sslopt = {'keyfile': args.sslkey, 'certfile': args.sslcert}
+    elif args.sslkey or args.sslcert:
+        raise ValueError('both sslkey and sslcert are required')
     else:
-        ssl_options = None
+        sslopt = None
 
     if args.port is None:
-        if ssl_options:
+        if sslopt:
             args.port = 443
         else:
@@ -481 +640 @@
         raise ValueError('invalid port: %s' % args.port)
 
-    if args.ageout <= 0:
-        raise ValueError('invalid ageout: %s' % args.ageout)
-
-    ioloop = tornado.ioloop.IOLoop.instance()
+    if not (0 <= args.ctlport <= 65535):
+        raise ValueError('invalid ctlport: %s' % args.ctlport)
+
+    if args.htpasswd:
+        args.htpasswd = Htpasswd(args.htpasswd)
+
+    ioloop = IOLoop.instance()
     fdb = FDB(ageout=args.ageout, debug=args.debug)
-    sw = SwitchingHub(fdb, debug=args.debug)
-
-    harg = {'switch': sw, 'htpasswd': args.htpasswd, 'debug': args.debug}
-    serv = (args.path, EtherWebSocketHandler, harg)
-    app = tornado.web.Application([serv])
-    server = tornado.httpserver.HTTPServer(app, ssl_options=ssl_options)
+    switch = SwitchingHub(fdb, debug=args.debug)
+
+    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.address)
 
-    for dev in args.device:
-        tap = TapHandler(ioloop, sw, dev, debug=args.debug)
-        tap.open()
+    ctl = Application([(args.ctlpath, EtherWebSocketControlHandler, {
+        'ioloop':   ioloop,
+        'switch':   switch,
+        'htpasswd': None,
+        'debug':    args.debug,
+    })])
+    ctlserver = HTTPServer(ctl)
+    ctlserver.listen(args.ctlport, address=args.ctladdress)
 
     if not args.foreground:
@@ -504 +673 @@
 
 
-def client_main(args):
-    realpath(args, 'cacerts')
-
-    if args.debug:
-        websocket.enableTrace(True)
-
-    if args.ageout <= 0:
-        raise ValueError('invalid ageout: %s' % args.ageout)
-
-    if args.user and args.passwd is None:
-        args.passwd = getpass.getpass()
-
-    ssl_ = ssl_wrapper(args.insecure, args.cacerts)
-    cred = {'user': args.user, 'passwd': args.passwd}
-    ioloop = tornado.ioloop.IOLoop.instance()
-    fdb = FDB(ageout=args.ageout, debug=args.debug)
-    sw = SwitchingHub(fdb, debug=args.debug)
-
-    for uri in args.uri:
-        client = EtherWebSocketClient(ioloop, sw, uri, ssl_, cred, args.debug)
-        client.open()
-
-    for dev in args.device:
-        tap = TapHandler(ioloop, sw, dev, debug=args.debug)
-        tap.open()
-
-    if not args.foreground:
-        daemonize()
-
-    ioloop.start()
-
-
-def main():
-    parser = argparse.ArgumentParser()
-    parser.add_argument('--device', action='append', default=[])
-    parser.add_argument('--ageout', action='store', type=int, default=300)
-    parser.add_argument('--foreground', action='store_true', default=False)
-    parser.add_argument('--debug', action='store_true', default=False)
-
-    subparsers = parser.add_subparsers(dest='subcommand')
-
-    parser_s = subparsers.add_parser('server')
-    parser_s.add_argument('--address', action='store', default='')
-    parser_s.add_argument('--port', action='store', type=int)
-    parser_s.add_argument('--path', action='store', default='/')
-    parser_s.add_argument('--htpasswd', action='store')
-    parser_s.add_argument('--keyfile', action='store')
-    parser_s.add_argument('--certfile', action='store')
-
-    parser_c = subparsers.add_parser('client')
-    parser_c.add_argument('--uri', action='append', default=[])
-    parser_c.add_argument('--insecure', action='store_true', default=False)
-    parser_c.add_argument('--cacerts', action='store')
-    parser_c.add_argument('--user', action='store')
-    parser_c.add_argument('--passwd', action='store')
-
-    args = parser.parse_args()
-
-    if args.subcommand == 'server':
-        server_main(args)
-    elif args.subcommand == 'client':
-        client_main(args)
-
-
 if __name__ == '__main__':
     main()

etherws/trunk/setup.py

@@ -50 +50 @@
         'python-pytun>=0.2',
         'websocket-client>=0.7.0',
-        'tornado>=2.2.1',
+        'tornado>=2.3',
     ],
     classifiers=[