Merge branch 'python23' - python 2.3 compatibility

* python23:
  Oops, missed another << operator to replace with _shl().
  socket.SHUT_RD and socket.SHUT_WR don't exist in python 2.3.
  compat/ssubprocess.py: some python versions don't have os.closerange().
  _nb_clean: don't catch EPIPE after all.
  Fix busy-waiting in two situations:
  Factor out common mainloop code between client and server.
  Implement our own left-shift operator to shut up python 2.3 warnings.
  Don't use set() since it's not in python 2.3.
  import and use subprocess.py from python 2.6.
  Remove list comprehensions for python 2.3 compatibility.

client.py

@@ -1,8 +1,10 @@
-import struct, socket, select, subprocess, errno, re
+import struct, socket, select, errno, re
+import compat.ssubprocess as ssubprocess
 import helpers, ssnet, ssh
 from ssnet import SockWrapper, Handler, Proxy, Mux, MuxWrapper
 from helpers import *
 
+
 def original_dst(sock):
     try:
         SO_ORIGINAL_DST = 80
@@ -45,7 +47,7 @@ class FirewallClient:
         e = None
         for argv in argv_tries:
             try:
-                self.p = subprocess.Popen(argv, stdout=s1, preexec_fn=setup)
+                self.p = ssubprocess.Popen(argv, stdout=s1, preexec_fn=setup)
                 e = None
                 break
             except OSError, e:
@@ -100,7 +102,13 @@ def _main(listener, fw, use_server, remotename, python, seed_hosts, auto_nets):
         else:
             helpers.logprefix = 'client: '
         debug1('connecting to server...\n')
-        (serverproc, serversock) = ssh.connect(remotename, python)
+        try:
+            (serverproc, serversock) = ssh.connect(remotename, python)
+        except socket.error, e:
+            if e.errno == errno.EPIPE:
+                raise Fatal("failed to establish ssh session")
+            else:
+                raise
         mux = Mux(serversock, serversock)
         handlers.append(mux)
 
@@ -169,20 +177,7 @@ def _main(listener, fw, use_server, remotename, python, seed_hosts, auto_nets):
             if rv:
                 raise Fatal('server died with error code %d' % rv)
         
-        r = set()
-        w = set()
-        x = set()
-        handlers = filter(lambda s: s.ok, handlers)
-        for s in handlers:
-            s.pre_select(r,w,x)
-        debug2('Waiting: %d[%d,%d,%d]...\n' 
-            % (len(handlers), len(r), len(w), len(x)))
-        (r,w,x) = select.select(r,w,x)
-        #log('r=%r w=%r x=%r\n' % (r,w,x))
-        ready = set(r) | set(w) | set(x)
-        for s in handlers:
-            if s.socks & ready:
-                s.callback()
+        ssnet.runonce(handlers, mux)
         if use_server:
             mux.callback()
             mux.check_fullness()

firewall.py

@@ -1,11 +1,12 @@
-import subprocess, re, errno
+import re, errno
+import compat.ssubprocess as ssubprocess
 import helpers
 from helpers import *
 
 
 def ipt_chain_exists(name):
     argv = ['iptables', '-t', 'nat', '-nL']
-    p = subprocess.Popen(argv, stdout = subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout = ssubprocess.PIPE)
     for line in p.stdout:
         if line.startswith('Chain %s ' % name):
             return True
@@ -17,7 +18,7 @@ def ipt_chain_exists(name):
 def ipt(*args):
     argv = ['iptables', '-t', 'nat'] + list(args)
     debug1('>> %s\n' % ' '.join(argv))
-    rv = subprocess.call(argv)
+    rv = ssubprocess.call(argv)
     if rv:
         raise Fatal('%r returned %d' % (argv, rv))
 
@@ -64,7 +65,7 @@ def do_iptables(port, subnets):
 
 def ipfw_rule_exists(n):
     argv = ['ipfw', 'list']
-    p = subprocess.Popen(argv, stdout = subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout = ssubprocess.PIPE)
     found = False
     for line in p.stdout:
         if line.startswith('%05d ' % n):
@@ -82,7 +83,7 @@ def ipfw_rule_exists(n):
 
 def sysctl_get(name):
     argv = ['sysctl', '-n', name]
-    p = subprocess.Popen(argv, stdout = subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout = ssubprocess.PIPE)
     line = p.stdout.readline()
     rv = p.wait()
     if rv:
@@ -96,7 +97,7 @@ def sysctl_get(name):
 def _sysctl_set(name, val):
     argv = ['sysctl', '-w', '%s=%s' % (name, val)]
     debug1('>> %s\n' % ' '.join(argv))
-    rv = subprocess.call(argv, stdout = open('/dev/null', 'w'))
+    rv = ssubprocess.call(argv, stdout = open('/dev/null', 'w'))
 
 
 _oldctls = []
@@ -110,7 +111,7 @@ def sysctl_set(name, val):
 def ipfw(*args):
     argv = ['ipfw', '-q'] + list(args)
     debug1('>> %s\n' % ' '.join(argv))
-    rv = subprocess.call(argv)
+    rv = ssubprocess.call(argv)
     if rv:
         raise Fatal('%r returned %d' % (argv, rv))
 

helpers.py

@@ -28,3 +28,10 @@ def debug3(s):
 
 class Fatal(Exception):
     pass
+
+
+def list_contains_any(l, sub):
+    for i in sub:
+        if i in l:
+            return True
+    return False

hostwatch.py

@@ -1,5 +1,6 @@
-import subprocess, time, socket, re, select, errno
+import time, socket, re, select, errno
 if not globals().get('skip_imports'):
+    import compat.ssubprocess as ssubprocess
     import helpers
     from helpers import *
 
@@ -108,7 +109,7 @@ def _check_netstat():
     debug2(' > netstat\n')
     argv = ['netstat', '-n']
     try:
-        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=null)
+        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
         content = p.stdout.read()
         p.wait()
     except OSError, e:
@@ -128,7 +129,7 @@ def _check_smb(hostname):
     argv = ['smbclient', '-U', '%', '-L', hostname]
     debug2(' > smb: %s\n' % hostname)
     try:
-        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=null)
+        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
         lines = p.stdout.readlines()
         p.wait()
     except OSError, e:
@@ -185,7 +186,7 @@ def _check_nmb(hostname, is_workgroup, is_master):
     argv = ['nmblookup'] + ['-M']*is_master + ['--', hostname]
     debug2(' > n%d%d: %s\n' % (is_workgroup, is_master, hostname))
     try:
-        p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=null)
+        p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE, stderr=null)
         lines = p.stdout.readlines()
         rv = p.wait()
     except OSError, e:

server.py

@@ -1,6 +1,7 @@
-import re, struct, socket, select, subprocess, traceback
+import re, struct, socket, select, traceback
 if not globals().get('skip_imports'):
     import ssnet, helpers, hostwatch
+    import compat.ssubprocess as ssubprocess
     from ssnet import SockWrapper, Handler, Proxy, Mux, MuxWrapper
     from helpers import *
 
@@ -36,14 +37,18 @@ def _maskbits(netmask):
     if not netmask:
         return 32
     for i in range(32):
-        if netmask[0] & (1<<i):
+        if netmask[0] & _shl(1, i):
             return 32-i
     return 0
     
     
+def _shl(n, bits):
+    return n * int(2**bits)
+
+
 def _list_routes():
     argv = ['netstat', '-rn']
-    p = subprocess.Popen(argv, stdout=subprocess.PIPE)
+    p = ssubprocess.Popen(argv, stdout=ssubprocess.PIPE)
     routes = []
     for line in p.stdout:
         cols = re.split(r'\s+', line)
@@ -53,7 +58,7 @@ def _list_routes():
         maskw = _ipmatch(cols[2])  # linux only
         mask = _maskbits(maskw)   # returns 32 if maskw is null
         width = min(ipw[1], mask)
-        ip = ipw[0] & (((1<<width)-1) << (32-width))
+        ip = ipw[0] & _shl(_shl(1, width) - 1, 32-width)
         routes.append((socket.inet_ntoa(struct.pack('!I', ip)), width))
     rv = p.wait()
     if rv != 0:
@@ -122,8 +127,9 @@ def main():
               socket.fromfd(sys.stdout.fileno(),
                             socket.AF_INET, socket.SOCK_STREAM))
     handlers.append(mux)
-    routepkt = ''.join('%s,%d\n' % r
-                       for r in routes)
+    routepkt = ''
+    for r in routes:
+        routepkt += '%s,%d\n' % r
     mux.send(0, ssnet.CMD_ROUTES, routepkt)
 
     hw = Hostwatch()
@@ -156,21 +162,6 @@ def main():
             if rpid:
                 raise Fatal('hostwatch exited unexpectedly: code 0x%04x\n' % rv)
         
-        r = set()
-        w = set()
-        x = set()
-        handlers = filter(lambda s: s.ok, handlers)
-        for s in handlers:
-            s.pre_select(r,w,x)
-        debug2('Waiting: %d[%d,%d,%d] (fullness=%d/%d)...\n' 
-               % (len(handlers), len(r), len(w), len(x),
-                  mux.fullness, mux.too_full))
-        (r,w,x) = select.select(r,w,x)
-        #log('r=%r w=%r x=%r\n' % (r,w,x))
-        ready = set(r) | set(w) | set(x)
-        for s in handlers:
-            #debug2('check: %r: %r\n' % (s, s.socks & ready))
-            if s.socks & ready:
-                s.callback()
+        ssnet.runonce(handlers, mux)
         mux.check_fullness()
         mux.callback()

ssh.py

@@ -1,4 +1,5 @@
-import sys, os, re, subprocess, socket, zlib
+import sys, os, re, socket, zlib
+import compat.ssubprocess as ssubprocess
 import helpers
 from helpers import *
 
@@ -14,9 +15,10 @@ def readfile(name):
 
 
 def empackage(z, filename):
+    (path,basename) = os.path.split(filename)
     content = z.compress(readfile(filename))
     content += z.flush(zlib.Z_SYNC_FLUSH)
-    return '%s\n%d\n%s' % (filename,len(content), content)
+    return '%s\n%d\n%s' % (basename,len(content), content)
 
 
 def connect(rhostport, python):
@@ -33,6 +35,7 @@ def connect(rhostport, python):
     z = zlib.compressobj(1)
     content = readfile('assembler.py')
     content2 = (empackage(z, 'helpers.py') +
+                empackage(z, 'compat/ssubprocess.py') +
                 empackage(z, 'ssnet.py') +
                 empackage(z, 'hostwatch.py') +
                 empackage(z, 'server.py') +
@@ -55,11 +58,10 @@ def connect(rhostport, python):
     def setup():
         # runs in the child process
         s2.close()
-        os.setsid()
     s1a,s1b = os.dup(s1.fileno()), os.dup(s1.fileno())
     s1.close()
     debug2('executing: %r\n' % argv)
-    p = subprocess.Popen(argv, stdin=s1a, stdout=s1b, preexec_fn=setup,
+    p = ssubprocess.Popen(argv, stdin=s1a, stdout=s1b, preexec_fn=setup,
                          close_fds=True)
     os.close(s1a)
     os.close(s1b)

ssnet.py

@@ -2,6 +2,12 @@ import struct, socket, errno, select
 if not globals().get('skip_imports'):
     from helpers import *
     
+# these don't exist in the socket module in python 2.3!
+SHUT_RD = 0
+SHUT_WR = 1
+SHUT_RDWR = 2
+
+
 HDR_LEN = 8
 
 
@@ -31,13 +37,30 @@ cmd_to_name = {
     
 
 
+def _add(l, elem):
+    if not elem in l:
+        l.append(elem)
+
+
+def _fds(l):
+    out = []
+    for i in l:
+        try:
+            out.append(i.fileno())
+        except AttributeError:
+            out.append(i)
+    out.sort()
+    return out
+
+
 def _nb_clean(func, *args):
     try:
         return func(*args)
     except OSError, e:
-        if e.errno not in (errno.EWOULDBLOCK, errno.EAGAIN, errno.EPIPE):
+        if e.errno not in (errno.EWOULDBLOCK, errno.EAGAIN):
             raise
         else:
+            debug3('%s: err was: %s\n' % (func.__name__, e))
             return None
 
 
@@ -69,21 +92,30 @@ class SockWrapper:
             debug1('%r: error was: %r\n' % (self, self.exc))
 
     def __repr__(self):
-        return 'SW:%s' % (self.peername,)
+        if self.rsock == self.wsock:
+            fds = '#%d' % self.rsock.fileno()
+        else:
+            fds = '#%d,%d' % (self.rsock.fileno(), self.wsock.fileno())
+        return 'SW%s:%s' % (fds, self.peername)
 
     def seterr(self, e):
         if not self.exc:
             self.exc = e
 
     def try_connect(self):
+        if self.connect_to and self.shut_write:
+            self.noread()
+            self.connect_to = None
         if not self.connect_to:
             return  # already connected
         self.rsock.setblocking(False)
+        debug3('%r: trying connect to %r\n' % (self, self.connect_to))
         try:
             self.rsock.connect(self.connect_to)
             # connected successfully (Linux)
             self.connect_to = None
         except socket.error, e:
+            debug3('%r: connect result: %r\n' % (self, e))
             if e.args[0] in [errno.EINPROGRESS, errno.EALREADY]:
                 pass  # not connected yet
             elif e.args[0] == errno.EISCONN:
@@ -102,14 +134,14 @@ class SockWrapper:
         if not self.shut_read:
             debug2('%r: done reading\n' % self)
             self.shut_read = True
-            #self.rsock.shutdown(socket.SHUT_RD)  # doesn't do anything anyway
+            #self.rsock.shutdown(SHUT_RD)  # doesn't do anything anyway
         
     def nowrite(self):
         if not self.shut_write:
             debug2('%r: done writing\n' % self)
             self.shut_write = True
             try:
-                self.wsock.shutdown(socket.SHUT_WR)
+                self.wsock.shutdown(SHUT_WR)
             except socket.error, e:
                 self.seterr(e)
 
@@ -159,7 +191,7 @@ class SockWrapper:
             wrote = outwrap.write(self.buf[0])
             self.buf[0] = self.buf[0][wrote:]
         while self.buf and not self.buf[0]:
-            self.buf[0:1] = []
+            self.buf.pop(0)
         if not self.buf and self.shut_read:
             outwrap.nowrite()
 
@@ -167,12 +199,13 @@ class SockWrapper:
 class Handler:
     def __init__(self, socks = None, callback = None):
         self.ok = True
-        self.socks = set(socks or [])
+        self.socks = socks or []
         if callback:
             self.callback = callback
 
     def pre_select(self, r, w, x):
-        r |= self.socks
+        for i in self.socks:
+            _add(r, i)
 
     def callback(self):
         log('--no callback defined-- %r\n' % self)
@@ -181,7 +214,7 @@ class Handler:
             v = s.recv(4096)
             if not v:
                 log('--closed-- %r\n' % self)
-                self.socks = set()
+                self.socks = []
                 self.ok = False
 
 
@@ -194,20 +227,20 @@ class Proxy(Handler):
 
     def pre_select(self, r, w, x):
         if self.wrap1.connect_to:
-            w.add(self.wrap1.rsock)
+            _add(w, self.wrap1.rsock)
         elif self.wrap1.buf:
             if not self.wrap2.too_full():
-                w.add(self.wrap2.wsock)
+                _add(w, self.wrap2.wsock)
         elif not self.wrap1.shut_read:
-            r.add(self.wrap1.rsock)
+            _add(r, self.wrap1.rsock)
 
         if self.wrap2.connect_to:
-            w.add(self.wrap2.rsock)
+            _add(w, self.wrap2.rsock)
         elif self.wrap2.buf:
             if not self.wrap1.too_full():
-                w.add(self.wrap1.wsock)
+                _add(w, self.wrap1.wsock)
         elif not self.wrap2.shut_read:
-            r.add(self.wrap2.rsock)
+            _add(r, self.wrap2.rsock)
 
     def callback(self):
         self.wrap1.try_connect()
@@ -216,6 +249,12 @@ class Proxy(Handler):
         self.wrap2.fill()
         self.wrap1.copy_to(self.wrap2)
         self.wrap2.copy_to(self.wrap1)
+        if self.wrap1.buf and self.wrap2.shut_write:
+            self.wrap1.buf = []
+            self.wrap1.noread()
+        if self.wrap2.buf and self.wrap1.shut_write:
+            self.wrap2.buf = []
+            self.wrap2.noread()
         if (self.wrap1.shut_read and self.wrap2.shut_read and
             not self.wrap1.buf and not self.wrap2.buf):
             self.ok = False
@@ -247,7 +286,10 @@ class Mux(Handler):
                 return self.chani
 
     def amount_queued(self):
-        return sum(len(b) for b in self.outbuf)
+        total = 0
+        for b in self.outbuf:
+            total += len(b)
+        return total
             
     def check_fullness(self):
         if self.fullness > 32768:
@@ -346,9 +388,9 @@ class Mux(Handler):
                 break
 
     def pre_select(self, r, w, x):
-        r.add(self.rsock)
+        _add(r, self.rsock)
         if self.outbuf:
-            w.add(self.wsock)
+            _add(w, self.wsock)
 
     def callback(self):
         (r,w,x) = select.select([self.rsock], [self.wsock], [], 0)
@@ -420,3 +462,28 @@ def connect_dst(ip, port):
     return SockWrapper(outsock, outsock,
                        connect_to = (ip,port),
                        peername = '%s:%d' % (ip,port))
+
+
+def runonce(handlers, mux):
+    r = []
+    w = []
+    x = []
+    handlers = filter(lambda s: s.ok, handlers)
+    for s in handlers:
+        s.pre_select(r,w,x)
+    debug2('Waiting: %d r=%r w=%r x=%r (fullness=%d/%d)\n' 
+            % (len(handlers), _fds(r), _fds(w), _fds(x),
+               mux.fullness, mux.too_full))
+    (r,w,x) = select.select(r,w,x)
+    debug2('  Ready: %d r=%r w=%r x=%r\n' 
+        % (len(handlers), _fds(r), _fds(w), _fds(x)))
+    ready = r+w+x
+    did = {}
+    for h in handlers:
+        for s in h.socks:
+            if s in ready:
+                h.callback()
+                did[s] = 1
+    for s in ready:
+        if not s in did:
+            raise Fatal('socket %r was not used by any handler' % s)