Skip to content

Instantly share code, notes, and snippets.

@khssnv
Last active December 3, 2021 09:35
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save khssnv/be4fa33e40d40d4da609bcf46ad5658d to your computer and use it in GitHub Desktop.
Save khssnv/be4fa33e40d40d4da609bcf46ad5658d to your computer and use it in GitHub Desktop.
Python SSL Threading TCP Server with custom data passed to RequestsHandler
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import ssl
from SocketServer import TCPServer, ThreadingMixIn, StreamRequestHandler
class SSL_TCPServer(TCPServer):
def __init__(self,
server_address,
RequestHandlerClass,
certfile,
keyfile,
ssl_version=ssl.PROTOCOL_TLSv1,
bind_and_activate=True):
TCPServer.__init__(self, server_address, RequestHandlerClass, bind_and_activate)
self.certfile = certfile
self.keyfile = keyfile
self.ssl_version = ssl_version
def get_request(self):
newsocket, fromaddr = self.socket.accept()
connstream = ssl.wrap_socket(newsocket,
server_side=True,
certfile = self.certfile,
keyfile = self.keyfile,
ssl_version = self.ssl_version)
return connstream, fromaddr
class SSL_ThreadingTCPServer(ThreadingMixIn, SSL_TCPServer):
def __init__(self, host_port_tuple, streamhandler, certfile, keyfile, myparameter, mymethod):
SSL_TCPServer.__init__(self, host_port_tuple, streamhandler, certfile, keyfile)
self.myparameter = myparameter
self.mymethod = mymethod
class IncomingHandler(StreamRequestHandler): # instantiates for each request
def handle(self):
data = self.connection.recv(4096)
print('Incoming: %s' % data)
print(self.server.myparameter)
print(self.server.mymethod)
def filter(self, data):
pass
def write(self, data):
pass
myparameter = 'I am custom server parameter available from request handler!'
def mymethod():
print('I am custrom method using custom parameter from request hander!')
return 'Custom method return'
print(mymethod)
server = SSL_ThreadingTCPServer(('localhost', 3009),
IncomingHandler,
'server.crt',
'server.key',
myparameter,
mymethod)
try:
server.serve_forever()
finally:
server.shutdown()
server.server_close()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment