Loading...
Searching...
No Matches
network.py
Go to the documentation of this file.
# vim: set fileencoding=utf-8 foldmethod=marker :
# {{{ Copyright 2013-2019 Bas Wijnen <wijnen@debian.org>
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or(at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# }}}
'''@file
Python module for easy networking. This module intends to make networking
easy. It supports tcp and unix domain sockets. Connection targets can be
specified in several ways.
'''
'''@package network Python module for easy networking.
This module intends to make networking easy. It supports tcp and unix domain
sockets. Connection targets can be specified in several ways.
'''
# {{{ Imports.
import math
import sys
import os
import socket
import select
import re
import time
import inspect
import fhs
modulename = 'network'
fhs.module_info(modulename, 'Networking made easy', '0.4', 'Bas Wijnen <wijnen@debian.org>')
fhs.module_option(modulename, 'tls', 'default tls hostname for server sockets. The code may ignore this option. Set to - to request that tls is disabled on the server. If left empty, detects hostname.', default = '')
import traceback
try:
import ssl
have_ssl = True
except:
have_ssl = False
# }}}
# {{{ Interface description
# - connection setup
# - connect to server
# - listen on port
# - when connected
# - send data
# - asynchronous read
# - blocking read for data
# implementation:
# - Server: listener, creating Sockets on accept
# - Socket: used for connection; symmetric
# }}}
if sys.version >= '3':
makestr = lambda x: str(x, 'utf8', 'replace') if isinstance(x, bytes) else x
else:
makestr = lambda x: x
log_output = sys.stderr
log_date = False
def set_log_output(file):
global log_output, log_date
log_output = file
log_date = True
# }}}
def log(*message, filename = None, line = None, funcname = None, depth = 0):
t = time.strftime('%F %T' if log_date else '%T')
source = inspect.currentframe().f_back
for d in range(depth):
source = source.f_back
code = source.f_code
if filename is None:
filename = os.path.basename(code.co_filename)
if funcname is None:
funcname = code.co_name
if line is None:
line = source.f_lineno
for msg in message:
log_output.write(''.join(['%s %s:%s:%d:\t%s\n' % (t, filename, funcname, line, m) for m in str(msg).split('\n')]))
log_output.flush()
# }}}
def lookup(service):
if isinstance(service, int):
return service
try:
return socket.getservbyname(service)
except socket.error:
pass
return int(service)
# }}}
class _Fake:
def __init__(self, i, o = None):
self._i = i
self._o = o if o is not None else i
def close(self):
pass
def sendall(self, data):
while len(data) > 0:
fd = self._o if isinstance(self._o, int) else self._o.fileno()
ret = os.write(fd, data)
if ret >= 0:
data = data[ret:]
continue
log('network.py: Failed to write data')
traceback.print_exc()
def recv(self, maxsize):
#log('recv fake')
return os.read(self._i.fileno(), maxsize)
def fileno(self):
# For reading.
return self._i.fileno()
# }}}
def wrap(i, o = None):
return Socket(_Fake(i, o))
# }}}
class Socket:
def __init__(self, address, tls = None, disconnect_cb = None, remote = None, connections = None):
self.tls = tls
self.remote = remote
self.connections = connections
if self.connections is not None:
self.connections.add(self)
self.socket = None
self._disconnect_cb = disconnect_cb
self._event = None
self._linebuffer = b''
if isinstance(address, (_Fake, socket.socket)):
#log('new %d' % id(address))
self.socket = address
return
if isinstance(address, str):
r = re.match('^(?:([a-z0-9-]+)://)?([^:/?#]+)(?::([^:/?#]+))?([:/?#].*)?$', address)
if r is not None:
# Group 1: protocol or None
# Group 2: hostname
# Group 3: port
# Group 4: everything after the port (address, query string, etc)
protocol = r.group(1)
hostname = r.group(2)
port = r.group(3)
url = r.group(4)
#print('protocol', protocol, 'hostname', hostname, 'port', port, 'url', url)
if self.tls is None:
if protocol is None:
self.tls = False
else:
self.tls = protocol != 'ws' and protocol.endswith('s')
if port is None:
port = protocol
else:
port = None
if address.startswith('./') or address.startswith('/') or (port is None and '/' in address):
# Unix socket.
# TLS is ignored for those.
self.remote = address
self.socket = socket.socket(socket.AF_UNIX)
self.socket.connect(self.remote)
return
elif r is None:
# Probably an error, but attempt to parse address as port.
hostname = 'localhost'
port = address
else:
# Url (with possibly some missing parts)
if port is None:
hostname = 'localhost'
port = address
else:
hostname = 'localhost'
port = address
self.remote = (hostname, lookup(port))
#log('remote %s' % str(self.remote))
self._setup_connection()
# }}}
def _setup_connection(self):
self.socket = socket.create_connection(self.remote)
if self.tls is None:
try:
assert have_ssl
self.socket = ssl.wrap_socket(self.socket, ssl_version = ssl.PROTOCOL_TLSv1)
self.tls = True
except:
self.tls = False
self.socket = socket.create_connection(self.remote)
elif self.tls is True:
try:
assert have_ssl
self.socket = ssl.wrap_socket(self.socket, ssl_version = ssl.PROTOCOL_TLSv1)
except ssl.SSLError as e:
raise TypeError('Socket does not seem to support TLS: ' + str(e))
else:
self.tls = False
# }}}
def disconnect_cb(self, disconnect_cb):
self._disconnect_cb = disconnect_cb
# }}}
def close(self):
if not self.socket:
return b''
data = self.unread()
self.socket.close()
self.socket = None
if self.connections is not None:
self.connections.remove(self)
if self._disconnect_cb:
return self._disconnect_cb(self, data) or b''
return data
# }}}
def send(self, data):
if self.socket is None:
return
#print 'sending %s' % repr(data)
try:
self.socket.sendall(data)
except BrokenPipeError:
self.close()
# }}}
def sendline(self, data):
if self.socket is None:
return
#print 'sending %s' % repr(data)
self.socket.sendall((data + '\n').encode('utf-8'))
# }}}
def recv(self, maxsize = 4096):
if self.socket is None:
log('recv on closed socket')
raise EOFError('recv on closed socket')
ret = b''
try:
ret = self.socket.recv(maxsize)
if hasattr(self.socket, 'pending'):
while self.socket.pending():
ret += self.socket.recv(maxsize)
except:
log('Error reading from socket: %s' % sys.exc_info()[1])
self.close()
return ret
if len(ret) == 0:
ret = self.close()
if not self._disconnect_cb:
raise EOFError('network connection closed')
return ret
# }}}
def rawread(self, callback, error = None):
if self.socket is None:
return b''
ret = self.unread()
self._callback = (callback, None)
self._event = add_read(self.socket, callback, error)
return ret
# }}}
def read(self, callback, error = None, maxsize = 4096):
if self.socket is None:
return b''
first = self.unread()
self._maxsize = maxsize
self._callback = (callback, False)
def cb():
data = self.recv(self._maxsize)
#log('network read %d bytes' % len(data))
if not self._event:
return False
callback(data)
return True
self._event = add_read(self.socket, cb, error)
if first:
callback(first)
# }}}
def readlines(self, callback, error = None, maxsize = 4096):
if self.socket is None:
return
self._linebuffer = self.unread()
self._maxsize = maxsize
self._callback = (callback, True)
self._event = add_read(self.socket, self._line_cb, error)
# }}}
def _line_cb(self):
self._linebuffer += self.recv(self._maxsize)
while b'\n' in self._linebuffer and self._event:
assert self._callback[1] is not None # Going directly from readlines() to rawread() is not allowed.
if self._callback[1]:
line, self._linebuffer = self._linebuffer.split(b'\n', 1)
line = makestr(line)
self._callback[0] (line)
else:
data = makestr(self._linebuffer)
self._linebuffer = b''
self._callback[0](data)
return True
# }}}
def unread(self):
if self._event:
try:
remove_read(self._event)
except ValueError:
# The function already returned False.
pass
self._event = None
ret = self._linebuffer
self._linebuffer = b''
return ret
# }}}
# }}}
class Server:
def __init__(self, port, obj, address = '', backlog = 5, tls = False, disconnect_cb = None):
self._obj = obj
self.port = ''
self.ipv6 = False
self._socket = None
self.tls = tls
self.connections = set()
self.disconnect_cb = disconnect_cb
if isinstance(port, str) and '/' in port:
# Unix socket.
# TLS is ignored for these sockets.
self.tls = False
self._socket = socket.socket(socket.AF_UNIX)
self._socket.bind(port)
self.port = port
self._socket.listen(backlog)
else:
self._tls_init()
port = lookup(port)
self._socket = socket.socket()
self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._socket.bind((address, port))
self._socket.listen(backlog)
if address == '':
self._socket6 = socket.socket(socket.AF_INET6)
self._socket6.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
self._socket6.bind(('::1', port))
self._socket6.listen(backlog)
self.ipv6 = True
self.port = port
self._event = add_read(self._socket, lambda: self._cb(False), lambda: self._cb(False))
if self.ipv6:
self._event = add_read(self._socket6, lambda: self._cb(True), lambda: self._cb(True))
def _cb(self, is_ipv6):
if is_ipv6:
new_socket = self._socket6.accept()
else:
new_socket = self._socket.accept()
#log('Accepted connection from %s; possibly attempting to set up encryption' % repr(new_socket))
if self.tls:
assert have_ssl
try:
new_socket = (ssl.wrap_socket(new_socket[0], ssl_version = ssl.PROTOCOL_TLSv1, server_side = True, certfile = self._tls_cert, keyfile = self._tls_key), new_socket[1])
except ssl.SSLError as e:
log('Rejecting (non-TLS?) connection for %s: %s' % (repr(new_socket[1]), str(e)))
try:
new_socket[0].shutdown(socket.SHUT_RDWR)
except:
# Ignore errors here.
pass
return True
except socket.error as e:
log('Rejecting connection for %s: %s' % (repr(new_socket[1]), str(e)))
try:
new_socket[0].shutdown(socket.SHUT_RDWR)
except:
# Don't care about errors on shutdown.
pass
return True
#log('Accepted TLS connection from %s' % repr(new_socket[1]))
s = Socket(new_socket[0], remote = new_socket[1], disconnect_cb = self.disconnect_cb, connections = self.connections)
self._obj(s)
return True
def close(self):
self._socket.close()
self._socket = None
if self.ipv6:
self._socket6.close()
self._socket6 = None
if isinstance(self.port, str) and '/' in self.port:
os.remove(self.port)
self.port = ''
def __del__(self):
if self._socket is not None:
self.close()
def _tls_init(self):
# Set up members for using tls, if requested.
if self.tls in (False, '-'):
self.tls = False
return
if self.tls in (None, True, ''):
self.tls = fhs.module_get_config('network')['tls']
if self.tls == '':
self.tls = socket.getfqdn()
elif self.tls == '-':
self.tls = False
return
# Use tls.
fc = fhs.read_data(os.path.join('certs', self.tls + os.extsep + 'pem'), opened = False, packagename = 'network')
fk = fhs.read_data(os.path.join('private', self.tls + os.extsep + 'key'), opened = False, packagename = 'network')
if fc is None or fk is None:
# Create new self-signed certificate.
certfile = fhs.write_data(os.path.join('certs', self.tls + os.extsep + 'pem'), opened = False, packagename = 'network')
csrfile = fhs.write_data(os.path.join('csr', self.tls + os.extsep + 'csr'), opened = False, packagename = 'network')
for p in (certfile, csrfile):
path = os.path.dirname(p)
if not os.path.exists(path):
os.makedirs(path)
keyfile = fhs.write_data(os.path.join('private', self.tls + os.extsep + 'key'), opened = False, packagename = 'network')
path = os.path.dirname(keyfile)
if not os.path.exists(path):
os.makedirs(path, 0o700)
os.system('openssl req -x509 -nodes -days 3650 -newkey rsa:4096 -subj "/CN=%s" -keyout "%s" -out "%s"' % (self.tls, keyfile, certfile))
os.system('openssl req -subj "/CN=%s" -new -key "%s" -out "%s"' % (self.tls, keyfile, csrfile))
fc = fhs.read_data(os.path.join('certs', self.tls + os.extsep + 'pem'), opened = False, packagename = 'network')
fk = fhs.read_data(os.path.join('private', self.tls + os.extsep + 'key'), opened = False, packagename = 'network')
self._tls_cert = fc
self._tls_key = fk
#print(fc, fk)
# }}}
_timeouts = []
_abort = False
def _handle_timeouts():
now = time.time()
while not _abort and len(_timeouts) > 0 and _timeouts[0][0] <= now:
_timeouts.pop(0)[1]()
if len(_timeouts) == 0:
return float('inf')
return _timeouts[0][0] - now
# }}}
_fds = [[], []]
def iteration(block = False):
# The documentation says timeout should be omitted, it doesn't mention making it None.
t = _handle_timeouts()
if not block:
t = 0
#log('do select with timeout %f' % t)
if math.isinf(t):
ret = select.select(_fds[0], _fds[1], _fds[0] + _fds[1])
else:
ret = select.select(_fds[0], _fds[1], _fds[0] + _fds[1], t)
#log('select returned %s' % repr(ret))
for f in ret[2]:
if f not in _fds[0] and f not in _fds[1]:
continue
if not f.error():
try:
remove_read(f)
except ValueError:
# The connection was already closed.
pass
if _abort:
return
for f in ret[0]:
if f not in _fds[0]:
continue
if not f.handle():
try:
remove_read(f)
except ValueError:
# The connection was already closed.
pass
if _abort:
return
for f in ret[1]:
if f not in _fds[1]:
continue
if not f.handle():
remove_write(f)
if _abort:
return
_handle_timeouts()
# }}}
_running = False
_idle = []
def fgloop():
global _running
assert not _running
_running = True
try:
while _running:
iteration(len(_idle) == 0)
if not _running:
return False
for i in _idle[:]:
if not i():
remove_idle(i)
if not _running:
break
finally:
_running = False
_abort = False
return False
# }}}
def bgloop():
assert _running == False
if os.getenv('NETWORK_NO_FORK') is None:
if os.fork() != 0:
sys.exit(0)
else:
log('Not backgrounding because NETWORK_NO_FORK is set\n')
fgloop()
# }}}
def endloop(force = False):
global _running, _abort
assert _running
_running = False
if force:
_abort = True
# }}}
class _fd_wrap:
def __init__(self, fd, cb, error):
self.fd = fd
self.handle = cb
if error is not None:
self.error = error
else:
self.error = self.default_error
def fileno(self):
if isinstance(self.fd, int):
return self.fd
else:
return self.fd.fileno()
def default_error(self):
try:
remove_read(self)
log('Error returned from select; removed fd from read list')
except:
try:
remove_write(self)
log('Error returned from select; removed fd from write list')
except:
log('Error returned from select, but fd was not in read or write list')
# }}}
def add_read(fd, cb, error = None):
_fds[0].append(_fd_wrap(fd, cb, error))
#log('add read %s' % repr(_fds[0][-1]))
return _fds[0][-1]
# }}}
def add_write(fd, cb, error = None):
_fds[1].append(_fd_wrap(fd, cb, error))
return _fds[1][-1]
# }}}
def add_timeout(abstime, cb):
_timeouts.append([abstime, cb])
ret = _timeouts[-1]
_timeouts.sort()
return ret
# }}}
def add_idle(cb):
_idle.append(cb)
return _idle[-1]
# }}}
def remove_read(handle):
#log('remove read %s' % repr(handle))
#traceback.print_stack()
_fds[0].remove(handle)
# }}}
def remove_write(handle):
_fds[1].remove(handle)
# }}}
def remove_timeout(handle):
_timeouts.remove(handle)
# }}}
def remove_idle(handle):
_idle.remove(handle)
# }}}
network.Server.tls
bool tls
False or the hostname for which the TLS keys are used.
Definition network.py:502
network.Server.port
str port
Port that is listened on.
Definition network.py:497
network.Server.disconnect_cb
disconnect_cb
Disconnect handler, to be used for new sockets.
Definition network.py:506
network.Server.connections
connections
Currently active connections for this server.
Definition network.py:504
network.Server.close
close(self)
Stop the server.
Definition network.py:565
network.Server.ipv6
bool ipv6
Whether the server listens for IPv6.
Definition network.py:499
network.Server
Listen on a network port and accept connections.
Definition network.py:464
network.Socket.recv
recv(self, maxsize=4096)
Read data from the network.
Definition network.py:346
network.Socket.connections
connections
connections set where this socket is registered.
Definition network.py:210
network.Socket.sendline
sendline(self, data)
Send a line of text.
Definition network.py:328
network.Socket.close
close(self)
Close the network connection.
Definition network.py:296
network.Socket.readlines
readlines(self, callback, error=None, maxsize=4096)
Buffer incoming data until a line is received, then call a function.
Definition network.py:422
network.Socket.send
send(self, data)
Send data over the network.
Definition network.py:313
network.Socket.remote
tuple remote
remote end of the network connection.
Definition network.py:208
network.Socket.rawread
rawread(self, callback, error=None)
Register function to be called when data is ready for reading.
Definition network.py:374
network.Socket.disconnect_cb
disconnect_cb(self, disconnect_cb)
Change the callback for disconnect notification.
Definition network.py:291
network.Socket.unread
unread(self)
Cancel a read() or rawread() callback.
Definition network.py:449
network.Socket.socket
socket
underlying socket object.
Definition network.py:214
network.Socket.read
read(self, callback, error=None, maxsize=4096)
Register function to be called when data is received.
Definition network.py:393
network.Socket.tls
bool tls
read only variable which indicates whether TLS encryption is used on this socket.
Definition network.py:206
network.Socket
Connection object.
Definition network.py:187
network.remove_write
remove_write(handle)
Definition network.py:809
network.bgloop
bgloop()
Like fgloop, but forks to the background.
Definition network.py:728
network.add_timeout
add_timeout(abstime, cb)
Definition network.py:791
network.remove_timeout
remove_timeout(handle)
Definition network.py:813
network.log
log(*message, filename=None, line=None, funcname=None, depth=0)
Log a message.
Definition network.py:102
network.add_read
add_read(fd, cb, error=None)
Definition network.py:780
network.add_idle
add_idle(cb)
Definition network.py:798
network.lookup
lookup(service)
Convert int or str with int or service to int port.
Definition network.py:123
network.remove_idle
remove_idle(handle)
Definition network.py:817
network.wrap
wrap(i, o=None)
Wrap two files into a fake socket.
Definition network.py:181
network.set_log_output
set_log_output(file)
Change target for log().
Definition network.py:85
network.iteration
iteration(block=False)
Do a single iteration of the main loop.
Definition network.py:639
network.remove_read
remove_read(handle)
Definition network.py:803
network.makestr
str makestr
Definition network.py:72
network.add_write
add_write(fd, cb, error=None)
Definition network.py:786
network.endloop
endloop(force=False)
Stop a loop that was started with fgloop() or bgloop().
Definition network.py:741
network.fgloop
fgloop()
Wait for events and handle them.
Definition network.py:694