| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436 |
- #
- # A higher level module for using sockets (or Windows named pipes)
- #
- # processing/connection.py
- #
- # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
- #
- __all__ = [ 'Client', 'Listener', 'Pipe' ]
- import os
- import sys
- import socket
- import time
- import tempfile
- import itertools
- from processing import _processing, currentProcess
- from processing.finalize import Finalize
- from processing.logger import subDebug
- from processing.reduction import connections_are_picklable
- try:
- from processing._processing import win32
- except ImportError:
- pass
- #
- #
- #
- BUFSIZE = 8192
- _nextId = itertools.count().next
- default_family = 'AF_INET'
- families = ['AF_INET']
- if hasattr(socket, 'AF_UNIX'):
- default_family = 'AF_UNIX'
- families += ['AF_UNIX']
- if sys.platform == 'win32':
- default_family = 'AF_PIPE'
- families += ['AF_PIPE']
- #
- #
- #
- def arbitraryAddress(family):
- '''
- Return an arbitrary free address for the given family
- '''
- if family == 'AF_INET':
- return ('localhost', 0)
- elif family == 'AF_UNIX':
- return tempfile.mktemp(prefix='pyc-%d-%d-' % (os.getpid(), _nextId()))
- elif family == 'AF_PIPE':
- return tempfile.mktemp(
- prefix=r'\\.\pipe\pyc-%d-%d-' % (os.getpid(), _nextId())
- )
- else:
- raise ValueError, 'unrecognized family'
- def addressType(address):
- '''
- Return the types of the address
- This can be 'AF_INET', 'AF_UNIX', or 'AF_PIPE'
- '''
- if type(address) == tuple:
- return 'AF_INET'
- elif type(address) is str and address.startswith('\\\\'):
- return 'AF_PIPE'
- elif type(address) is str:
- return 'AF_UNIX'
- else:
- raise ValueError, 'address type of %r unrecognized' % address
- #
- # Public functions
- #
- class Listener(object):
- '''
- Returns a listener object.
- This is a wrapper for a bound socket which is 'listening' for
- connections, or for a Windows named pipe.
- '''
- def __init__(self, address=None, family=None, backlog=1,
- authenticate=False, authkey=None):
- '''
- `address`
- The address to be used by the bound socket
- or named pipe of `self`.
- `family`
- The type of the socket or named pipe to use.
- This can be one of the strings 'AF_INET' (for a TCP
- socket), 'AF_UNIX' (for a Unix domain socket) or 'AF_PIPE'
- (for a Windows named pipe). Of these only the first is
- guaranteed to be available.
- If `family` is None than the family is inferred by the
- format of `address`. If `address` is unspecified then
- a default is chosen which is dependent on the platform.
- This default is the family which is assumed to be the
- fastest available.
- `backlog`
- If the `self` uses a socket then this is passed to the
- `listen()` method of the socket once it has been bound.
- `authenticate`
- If this is true then digest authentication is used even if
- `authkey` is` None`.
- `authkey`
- If `authkey` is a string then it will be used as the
- authentication key; otherwise it must be `None`.
- If `authkey` is `None` and `authenticate` is true then
- `currentProcess.getAuthKey()` is used as the authentication
- key.
- If `authkey` is `None` and `authentication` is false then
- no authentication is done.
- '''
- family = family or (address and addressType(address)) \
- or default_family
- address = address or arbitraryAddress(family)
- if family == 'AF_PIPE':
- self._listener = PipeListener(address, backlog)
- else:
- self._listener = SocketListener(address, family, backlog)
- if authenticate and authkey is None:
- authkey = currentProcess().getAuthKey()
- elif authenticate:
- assert type(authkey) is str
- self._authkey = authkey
- def accept(self):
- '''
- Accept a connection on the bound socket or named pipe of `self`.
- Returns a `Connection` object.
- '''
- c = self._listener.accept()
- if self._authkey:
- deliverChallenge(c, self._authkey)
- answerChallenge(c, self._authkey)
- return c
- def close(self):
- '''
- Close the bound socket or named pipe of `self`.
- '''
- return self._listener.close()
- address = property(lambda self: self._listener._address)
- last_accepted = property(lambda self: self._listener._last_accepted)
- def Client(address, family=None, authenticate=False, authkey=None):
- '''
- Returns a connection to the address of a `Listener`
- '''
- family = family or addressType(address)
- if family == 'AF_PIPE':
- c = PipeClient(address)
- else:
- c = SocketClient(address)
- if authenticate and authkey is None:
- authkey = currentProcess().getAuthKey()
- elif authenticate:
- assert type(authkey) is str
- if authkey is not None:
- answerChallenge(c, authkey)
- deliverChallenge(c, authkey)
- return c
- if sys.platform != 'win32':
- def Pipe(duplex=True):
- '''
- Returns pair of connection objects at either end of a pipe
- '''
- if duplex:
- s1, s2 = socket.socketpair()
- c1 = _processing.Connection(s1.fileno())
- c2 = _processing.Connection(s2.fileno())
- s1.close()
- s2.close()
- else:
- fd1, fd2 = os.pipe()
- c1 = _processing.Connection(fd1, duplicate=False)
- c2 = _processing.Connection(fd2, duplicate=False)
- return c1, c2
-
- else:
- def Pipe(duplex=True):
- '''
- Returns pair of connection objects at either end of a pipe
- '''
- address = arbitraryAddress('AF_PIPE')
- if duplex:
- openmode = win32.PIPE_ACCESS_DUPLEX
- access = win32.GENERIC_READ | win32.GENERIC_WRITE
- obsize, ibsize = BUFSIZE, BUFSIZE
- else:
- openmode = win32.PIPE_ACCESS_INBOUND
- access = win32.GENERIC_WRITE
- obsize, ibsize = 0, BUFSIZE
- h1 = win32.CreateNamedPipe(
- address, openmode,
- win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
- win32.PIPE_WAIT,
- 1, obsize, ibsize, win32.NMPWAIT_WAIT_FOREVER, win32.NULL
- )
- h2 = win32.CreateFile(
- address, access, 0, win32.NULL, win32.OPEN_EXISTING, 0, win32.NULL
- )
- win32.SetNamedPipeHandleState(
- h2, win32.PIPE_READMODE_MESSAGE, None, None
- )
- try:
- win32.ConnectNamedPipe(h1, win32.NULL)
- except WindowsError, e:
- if e.args[0] != win32.ERROR_PIPE_CONNECTED:
- raise
- c1 = _processing.PipeConnection(h1, duplicate=False)
- c2 = _processing.PipeConnection(h2, duplicate=False)
-
- return c1, c2
- #
- # Definitions for connections based on sockets
- #
- class SocketListener(object):
- '''
- Represtation of a socket which is bound to an address and listening
- '''
- def __init__(self, address, family, backlog=1):
- self._socket = socket.socket(getattr(socket, family))
- self._socket.bind(address)
- if family == 'AF_UNIX':
- os.chmod(address, int('0600', 8)) # readable/writable only by user
- self._socket.listen(backlog)
- address = self._socket.getsockname()
- if type(address) is tuple:
- address = (socket.getfqdn(address[0]),) + address[1:]
- self._address = address
- self._family = family
- self._last_accepted = None
- subDebug('listener bound to address %r', self._address)
- if family == 'AF_UNIX':
- self._unlink = Finalize(
- self, os.unlink, args=(self._address,), exitpriority=0
- )
- else:
- self._unlink = None
- def accept(self):
- s, self._last_accepted = self._socket.accept()
- conn = _processing.Connection(s.fileno())
- s.close()
- return conn
- def close(self):
- self._socket.close()
- if self._unlink is not None:
- self._unlink()
- def SocketClient(address):
- '''
- Return a connection object connected to the socket given by `address`
- '''
- family = addressType(address)
- s = socket.socket( getattr(socket, family) )
- endtime = time.time() + 10
- while endtime > time.time():
- try:
- s.connect(address)
- except socket.error, e:
- if e.args[0] != 10061: # 10061 => connection refused
- raise
- time.sleep(0.01)
- else:
- break
- else:
- raise
- conn = _processing.Connection(s.fileno())
- s.close()
- return conn
- #
- # Definitions for connections based on named pipes
- #
- if sys.platform == 'win32':
- class PipeListener(object):
- '''
- Representation of a named pipe
- '''
- def __init__(self, address, backlog=None):
- self._address = address
- handle = win32.CreateNamedPipe(
- address, win32.PIPE_ACCESS_DUPLEX,
- win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
- win32.PIPE_WAIT,
- win32.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
- win32.NMPWAIT_WAIT_FOREVER, win32.NULL
- )
- self._handle_queue = [handle]
- self._last_accepted = None
-
- subDebug('listener created with address=%r', self._address)
- self.close = Finalize(
- self, PipeListener._finalize_pipelistener,
- args=(self._handle_queue, self._address), exitpriority=0
- )
-
- def accept(self):
- newhandle = win32.CreateNamedPipe(
- self._address, win32.PIPE_ACCESS_DUPLEX,
- win32.PIPE_TYPE_MESSAGE | win32.PIPE_READMODE_MESSAGE |
- win32.PIPE_WAIT,
- win32.PIPE_UNLIMITED_INSTANCES, BUFSIZE, BUFSIZE,
- win32.NMPWAIT_WAIT_FOREVER, win32.NULL
- )
- self._handle_queue.append(newhandle)
- handle = self._handle_queue.pop(0)
- try:
- win32.ConnectNamedPipe(handle, win32.NULL)
- except WindowsError, e:
- if e.args[0] != win32.ERROR_PIPE_CONNECTED:
- raise
- return _processing.PipeConnection(handle, duplicate=False)
- @staticmethod
- def _finalize_pipelistener(queue, address):
- subDebug('closing listener with address=%r', address)
- for handle in queue:
- win32.CloseHandle(handle)
-
- def PipeClient(address):
- '''
- Return a connection object connected to the pipe given by `address`
- '''
- endtime = time.time() + 10
- while endtime > time.time():
- try:
- win32.WaitNamedPipe(address, 1000)
- h = win32.CreateFile(
- address, win32.GENERIC_READ | win32.GENERIC_WRITE,
- 0, win32.NULL, win32.OPEN_EXISTING, 0, win32.NULL
- )
- except WindowsError, e:
- if e.args[0] not in (win32.ERROR_SEM_TIMEOUT,
- win32.ERROR_PIPE_BUSY):
- raise
- else:
- break
- else:
- raise
- win32.SetNamedPipeHandleState(
- h, win32.PIPE_READMODE_MESSAGE, None, None
- )
- return _processing.PipeConnection(h, duplicate=False)
- #
- # Authentication stuff
- #
- class AuthenticationError(Exception):
- pass
- def deliverChallenge(connection, authkey):
- import hmac, sha
- assert type(authkey) is str, '%r is not a string' % authkey
- try:
- message = os.urandom(20)
- except AttributeError:
- import random
- message = ''.join(str(random.randrange(256)) for i in range(20))
- connection.sendBytes('#CHALLENGE:' + message)
- digest = hmac.new(authkey, message, sha).digest()
- response = connection.recvBytes()
- if response == digest:
- connection.sendBytes('#WELCOME')
- else:
- connection.sendBytes('#AUTHENTICATION_FAILED')
- raise AuthenticationError, 'digest received was wrong'
- def answerChallenge(connection, authkey):
- import hmac, sha
- assert type(authkey) is str, '%r is not a string' % authkey
- message = connection.recvBytes()
- assert message[:11] == '#CHALLENGE:', 'message = %r' % message
- message = message[11:]
- digest = hmac.new(authkey, message, sha).digest()
- connection.sendBytes(digest)
- response = connection.recvBytes()
- if response != '#WELCOME':
- raise AuthenticationError, 'digest sent was rejected'
- # deprecated
- deliver_Challenge = deliverChallenge
- answer_challenge = answerChallenge
|