Skip to content

Instantly share code, notes, and snippets.

@zapstar
Created June 8, 2018 09:01
Show Gist options
  • Save zapstar/a7035795483753f7b71a542559afa83f to your computer and use it in GitHub Desktop.
Save zapstar/a7035795483753f7b71a542559afa83f to your computer and use it in GitHub Desktop.
Python Asyncio SSL client and server examples
#!/usr/bin/env python3
import asyncio
import ssl
@asyncio.coroutine
async def echo_client(data, loop):
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_ctx.options |= ssl.OP_NO_TLSv1
ssl_ctx.options |= ssl.OP_NO_TLSv1_1
ssl_ctx.load_cert_chain('client_cert.pem', keyfile='client_key.pem')
ssl_ctx.load_verify_locations(cafile='client_ca.pem')
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.VerifyMode.CERT_REQUIRED
ssl_ctx.set_ciphers('ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384')
reader, writer = await asyncio.open_connection('127.0.0.1', 8080, ssl=ssl_ctx, loop=loop)
print('Sending: {}'.format(data))
writer.write(len(data).to_bytes(4, byteorder='big'))
writer.write(data.encode())
await writer.drain()
size_bytes = await reader.readexactly(4)
size = int.from_bytes(size_bytes, byteorder='big')
echo_data = await reader.readexactly(size)
print('Received: {}'.format(echo_data))
writer.close()
if __name__ == '__main__':
async_loop = asyncio.get_event_loop()
send_data = "Ping!!!"
async_loop.run_until_complete(echo_client(send_data, async_loop))
async_loop.close()
#!/usr/bin/env python3
import asyncio
import ssl
@asyncio.coroutine
async def handle_connection(reader, writer):
addr = writer.get_extra_info('peername')
print('Connection established with {}'.format(addr))
while True:
# Read the marker
try:
size_bytes = await reader.readexactly(4)
if not size_bytes:
print('Connection terminated with {}'.format(addr))
break
except asyncio.IncompleteReadError:
print('Connection terminated with {}'.format(addr))
break
size = int.from_bytes(size_bytes, byteorder='big')
# Read the data
try:
data = await reader.readexactly(size)
if not size_bytes:
print('Connection terminated with {}'.format(addr))
break
except asyncio.IncompleteReadError:
print('Connection terminated with {}'.format(addr))
break
print('Read {} bytes from the client: {}'.format(size, addr))
# Reverse the string
echo_data = ''.join(reversed(data.decode()))
# Send the marker
writer.write(len(echo_data).to_bytes(4, byteorder='big'))
# Send the data itself
writer.write(echo_data.encode())
# Wait for the data to be written back
await writer.drain()
print('Finished sending {} bytes to the client: {}'.format(size, addr))
def setup_server():
ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_ctx.options |= ssl.OP_NO_TLSv1
ssl_ctx.options |= ssl.OP_NO_TLSv1_1
ssl_ctx.options |= ssl.OP_SINGLE_DH_USE
ssl_ctx.options |= ssl.OP_SINGLE_ECDH_USE
ssl_ctx.load_cert_chain('server_cert.pem', keyfile='server_key.pem')
ssl_ctx.load_verify_locations(cafile='server_ca.pem')
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.VerifyMode.CERT_REQUIRED
ssl_ctx.set_ciphers('ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384')
loop = asyncio.get_event_loop()
coroutine = asyncio.start_server(handle_connection,
'127.0.0.1',
8080,
ssl=ssl_ctx,
loop=loop)
server = loop.run_until_complete(coroutine)
print('Serving on {}'.format(server.sockets[0].getsockname()))
loop.run_forever()
if __name__ == '__main__':
setup_server()
@mohak-pam
Copy link

I’m working on creating a proxy between a PostgreSQL client and server that intercepts and reads all communications. I’ve successfully implemented this with asyncio for non-SSL connections. However, in production, SSL is enabled, and everything is encrypted, which prevents me from reading the data.

To address this, I’m considering setting up two separate SSL connections: one between the PostgreSQL client and the proxy, and another between the proxy and the PostgreSQL server. Unfortunately, because PostgreSQL uses the Wire Protocol, I’m having difficulty establishing these two independent SSL connections.

Could you guide me on how to achieve this using asyncio?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment