Support for blocking and nonblocking sockets, using Asynchronous*Channel support in Java 7. Next step is to consider how to implement SSLEngine in conjunction.

blocking_aio.py

from java.net import InetSocketAddress
from java.nio.channels import AsynchronousSocketChannel as ASC
from java.nio import ByteBuffer

from array import array

remote = "jim-baker.com"

s = ASC.open()
s.connect(InetSocketAddress(remote, 80)).get()
s.write(ByteBuffer.wrap("""GET / HTTP/1.0\r
Host: jim-baker.com\r\n\r\n""")).get()

# could i just do a flip here?

dest = ByteBuffer.allocate(1024)
s.read(dest).get()
x = dest.array()[:dest.position()].tostring()
print len(x), x
s.close()

design.md

socket

• Python type - socket._socketobject
• Wraps an AsynchronousSocketChannel (if used with connect) or AsynchronousServerSocketChannel (if used with listen)
• Can be switched between blocking and nonblocking behavior as desired; when blocking, timeouts can be specified. Defaults to blocking.
• Each socket has a lock and corresponding condition variable
• Blocking operations use the condition variable, with timeouts on the await as desired
• Maintains a list of listeners, using CopyOnWriteArrayList so to avoid locking/unlocking each socket with select/poll.register. Note that listeners do not signal the socket's condition variable. We do listeners even for blocking, since it's possible for select to be on a blocking socket. Listeners will be a subclass of Listener, and include SelectListener and PollListener, to provide the necessary differentiated behavior.
• Buffers incoming, outgoing data with ArrayDeque, protected by a lock, of ByteBuffer (maybe we can reuse, probably not) Uses state variables to implement state machine re blocking, connect/listen, etc

select.select, implemented via a object that wraps the following:

• Condition variable for notification
• dict for each set of rlist, wlist, xlist sockets (or channels generally, should revisit files)
• SelectListener, as invoked by a the socket's completion handler for an op, goes through its listener list, sets the dicts as appropriate, then signals the condition

Upon waking up from await, select function then listifies each dict

select.poll Create a poll object that wraps a BlockingQueue

poll.register(socket, eventmask) - adds to the socket's listener a PollListener(eventmask) poll.unregister(socket) - checks return value of CopyOnArrayWriteList.remove for socket, returns KeyError is not removed poll.poll([timeout]) - simply performs blockingqueue.poll(timeout, timeunit)

ssl_engine_state_machine.py

# does not yet work!

import time
from array import array

from java.net import InetSocketAddress
from java.nio.channels import AsynchronousSocketChannel as ASC
from java.nio import ByteBuffer
from javax.net.ssl import SSLContext
from javax.net.ssl.SSLEngineResult import Status as ResultStatus
from javax.net.ssl.SSLEngineResult import HandshakeStatus


# so ideally this layer will do the following
# it will process the socket until it has data the user can actually use (although maybe we can still have underflow)
# in particular it will handle *all* handshaking, wrap/unrwap

# maybe we should use ArrayDeque to avoid unnecessary reallocs, then
# scatter/gather into buffers, using toArray to integrate
# probably not

# need to take in account that reading data on the channel might have more app data... need to pump that through

# need to init handshake for other operations, such as determining
# negotiated keys; otherwise wrap/unwrap will implicitly do this



def stringify_bb(x):
    return x.array()[:x.position()].tostring()


class SSLChannel(object):

    # preliminary blocking only version

    def __init__(self, channel, context=None, client_mode=True):
        self.channel = channel
        if context is None:
            context = SSLContext.getDefault()
        self.engine = context.createSSLEngine()

        self.src = self.alloc_buffer()
        self.wrapped_src = self.alloc_buffer()
        self.wrapped_dest = self.alloc_buffer()
        self.dest = self.alloc_buffer()
        self.dummy = ByteBuffer.allocate(0);
        self.engine.useClientMode = client_mode

    def alloc_buffer(self):
        return ByteBuffer.allocate(32768)  # be especially generous for now to avoid realloc

    def read(self):
        # unwrap network data
        return unwrap(self.channel.read(...))

    def write(self, data):
        self.src.clear()
        self.src.put(data)
        self.src.flip()
        self.process(self.engine.wrap(self.src, self.wrapped_src))  # handles actual channel write

    def process(self, engine_state):
        # per docs and with minor paraphrasing to make clearer:
        #
        # Data moves through the engine by calling wrap() or unwrap()
        # on outbound or inbound data, respectively.
        #
        # Depending on the state of the SSLEngine:
        #
        # * wrap() may consume application data from the source buffer and
        #   may produce network data in the destination buffer. The
        #   outbound data may contain application and/or handshake data.
        #
        # * unwrap() will examine the source buffer and may advance the
        #   handshake if the data is handshaking information, or may place
        #   application data in the destination buffer if the data is
        #   application.
        #
        # The state of the underlying SSL/TLS algorithm
        # will determine when data is consumed and produced."

        # should not allocate buffers all the time. but we will for this
        # first pass of code.

        while True:
            rs = engine_state.status
            if rs == ResultStatus.OK:
                hs = engine_state.handshakeStatus
                if hs == HandshakeStatus.FINISHED:
                    print "finished handshake"
                    break

                elif hs == HandshakeStatus.NEED_TASK:
                    # let's exec each task in the thread pool regardless
                    # of blocking status, for now just do it in this
                    # thread; note "Multiple delegated tasks can be run in
                    # parallel." per docs
                    print "Need to run task(s)"
                    while True:
                        task = engine.getDelegatedTask()
                        if task is None:
                            print "No more tasks"
                            break
                        print "Running task", task
                        task.run()
                        print "Completed task", task
                    continue

                elif hs == HandshakeStatus.NEED_UNWRAP:
                    print "need to unwrap handshake"  # FIXME do we need to unwrap some bytes sent from the client?
                    self.channel.read(self.wrapped_dest).get()
                    engine_state = engine.unwrap(b, dest)
                    continue

                elif hs == HandshakeStatus.NEED_WRAP:
                    # flush buffer

                elif hs == HandshakeStatus.NOT_HANDSHAKING:
                    pass  # is this an error state

            elif rs = ResultStatus.BUFFER_UNDERFLOW:
                print "Need more data from peer"
                return rs  # get more data

            elif rs = ResultStatus.BUFFER_OVERFLOW:
                # FIXME should alloc larger buffers in next version of
                # code instead of overly large preallocated buffers
                raise Exception("need to alloc a larger buffer")


def main():
    remote_host = "www.verisign.com"
    http_req = """GET / HTTP/1.0\r
Host: {}\r\n\r\n""".format(remote_host)
    
    channel = ASC.open()
    channel.connect(InetSocketAddress(remote, 443)).get()
    ssl_channel = SSLChannel(channel)
    ssl_channel.write(http_req)
    print ssl_channel.read()

    # channel.close()
    # need to take in account the closing process re handshaking, etc
    #engine.closeInbound()
    #engine.closeOutbound()


if __name__ == "__main__":
    main()