I'm trying to write a forwarding http proxy with aiohttp, I've currently got it working with http but wanting it to work with https(without decryption).
import asyncio
from aiohttp import web, ClientSession
loop = asyncio.get_event_loop()
async def handler(server_request):
if server_request.method == "CONNECT":
print(await server_request.read())
else:
async with ClientSession() as session:
async with session.request(server_request.method, server_request.raw_path) as request:
response = web.StreamResponse(status=200,
reason='OK',
headers={'Content-Type': 'text/html'})
await response.prepare(server_request)
while True:
chunk = await request.content.read()
if not chunk:
break
response.write(chunk)
return response
server = web.Server(handler)
loop.run_until_complete(loop.create_server(server, "0.0.0.0", 8080))
try:
loop.run_forever()
except KeyboardInterrupt:
loop.close()
pass
I've got to the point where I need to get the raw body to send down the tunnel to the destination but can't seem to access it
If I attempt to read I get an exception:
b''
Unhandled exception
Traceback (most recent call last):
File "/usr/local/lib/python3.6/dist-packages/aiohttp/web_protocol.py", line 434, in start
yield from resp.prepare(request)
AttributeError: 'NoneType' object has no attribute 'prepare'
The basic idea in a HTTPS proxy is to act like a stream proxy, where we read raw data and put raw data to destination. Unfortunately after skimming through aiohttp I could not find anything which would help achieve the same, so you need to us basic asyncio.
Below code snippets shows the main logic behind the https part
if head[0] == 'CONNECT': # https proxy
try:
logger.info('%sBYPASSING <%s %s> (SSL connection)' %
('[%s] ' % ident if verbose >= 1 else '', head[0], head[1]))
m = REGEX_HOST.search(head[1])
host = m.group(1)
port = int(m.group(2))
req_reader, req_writer = yield from asyncio.open_connection(host, port, ssl=False, loop=loop)
client_writer.write(b'HTTP/1.1 200 Connection established\r\n\r\n')
@asyncio.coroutine
def relay_stream(reader, writer):
try:
while True:
line = yield from reader.read(1024)
if len(line) == 0:
break
writer.write(line)
except:
print_exc()
tasks = [
asyncio.async(relay_stream(client_reader, req_writer), loop=loop),
asyncio.async(relay_stream(req_reader, client_writer), loop=loop),
]
yield from asyncio.wait(tasks, loop=loop)
except:
print_exc()
finally:
return
The complete PROXY code is as below
#!/usr/bin/env python3
VERSION = "v0.2.0"
"""
Copyright (c) 2013 devunt
Permission is hereby granted, free of charge, to any person
obtaining a copy of this software and associated documentation
files (the "Software"), to deal in the Software without
restriction, including without limitation the rights to use,
copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following
conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.
"""
import sys
if sys.version_info < (3, 4):
print('Error: You need python 3.4.0 or above.')
exit(1)
from argparse import ArgumentParser
from socket import TCP_NODELAY
from time import time
from traceback import print_exc
import asyncio
import logging
import random
import functools
import re
REGEX_HOST = re.compile(r'(.+?):([0-9]{1,5})')
REGEX_CONTENT_LENGTH = re.compile(r'\r\nContent-Length: ([0-9]+)\r\n', re.IGNORECASE)
REGEX_CONNECTION = re.compile(r'\r\nConnection: (.+)\r\n', re.IGNORECASE)
clients = {}
logging.basicConfig(level=logging.INFO, format='[%(asctime)s] {%(levelname)s} %(message)s')
logging.getLogger('asyncio').setLevel(logging.CRITICAL)
logger = logging.getLogger('warp')
verbose = 0
def accept_client(client_reader, client_writer, *, loop=None):
ident = hex(id(client_reader))[-6:]
task = asyncio.async(process_warp(client_reader, client_writer, loop=loop), loop=loop)
clients[task] = (client_reader, client_writer)
started_time = time()
def client_done(task):
del clients[task]
client_writer.close()
logger.debug('[%s] Connection closed (took %.5f seconds)' % (ident, time() - started_time))
logger.debug('[%s] Connection started' % ident)
task.add_done_callback(client_done)
@asyncio.coroutine
def process_warp(client_reader, client_writer, *, loop=None):
ident = str(hex(id(client_reader)))[-6:]
header = ''
payload = b''
try:
RECV_MAX_RETRY = 3
recvRetry = 0
while True:
line = yield from client_reader.readline()
if not line:
if len(header) == 0 and recvRetry < RECV_MAX_RETRY:
# handle the case when the client make connection but sending data is delayed for some reasons
recvRetry += 1
yield from asyncio.sleep(0.2, loop=loop)
continue
else:
break
if line == b'\r\n':
break
if line != b'':
header += line.decode()
m = REGEX_CONTENT_LENGTH.search(header)
if m:
cl = int(m.group(1))
while (len(payload) < cl):
payload += yield from client_reader.read(1024)
except:
print_exc()
if len(header) == 0:
logger.debug('[%s] !!! Task reject (empty request)' % ident)
return
req = header.split('\r\n')[:-1]
if len(req) < 4:
logger.debug('[%s] !!! Task reject (invalid request)' % ident)
return
head = req[0].split(' ')
if head[0] == 'CONNECT': # https proxy
try:
logger.info('%sBYPASSING <%s %s> (SSL connection)' %
('[%s] ' % ident if verbose >= 1 else '', head[0], head[1]))
m = REGEX_HOST.search(head[1])
host = m.group(1)
port = int(m.group(2))
req_reader, req_writer = yield from asyncio.open_connection(host, port, ssl=False, loop=loop)
client_writer.write(b'HTTP/1.1 200 Connection established\r\n\r\n')
@asyncio.coroutine
def relay_stream(reader, writer):
try:
while True:
line = yield from reader.read(1024)
if len(line) == 0:
break
writer.write(line)
except:
print_exc()
tasks = [
asyncio.async(relay_stream(client_reader, req_writer), loop=loop),
asyncio.async(relay_stream(req_reader, client_writer), loop=loop),
]
yield from asyncio.wait(tasks, loop=loop)
except:
print_exc()
finally:
return
phost = False
sreq = []
sreqHeaderEndIndex = 0
for line in req[1:]:
headerNameAndValue = line.split(': ', 1)
if len(headerNameAndValue) == 2:
headerName, headerValue = headerNameAndValue
else:
headerName, headerValue = headerNameAndValue[0], None
if headerName.lower() == "host":
phost = headerValue
elif headerName.lower() == "connection":
if headerValue.lower() in ('keep-alive', 'persist'):
# current version of this program does not support the HTTP keep-alive feature
sreq.append("Connection: close")
else:
sreq.append(line)
elif headerName.lower() != 'proxy-connection':
sreq.append(line)
if len(line) == 0 and sreqHeaderEndIndex == 0:
sreqHeaderEndIndex = len(sreq) - 1
if sreqHeaderEndIndex == 0:
sreqHeaderEndIndex = len(sreq)
m = REGEX_CONNECTION.search(header)
if not m:
sreq.insert(sreqHeaderEndIndex, "Connection: close")
if not phost:
phost = '127.0.0.1'
path = head[1][len(phost)+7:]
logger.info('%sWARPING <%s %s>' % ('[%s] ' % ident if verbose >= 1 else '', head[0], head[1]))
new_head = ' '.join([head[0], path, head[2]])
m = REGEX_HOST.search(phost)
if m:
host = m.group(1)
port = int(m.group(2))
else:
host = phost
port = 80
try:
req_reader, req_writer = yield from asyncio.open_connection(host, port, flags=TCP_NODELAY, loop=loop)
req_writer.write(('%s\r\n' % new_head).encode())
yield from req_writer.drain()
yield from asyncio.sleep(0.2, loop=loop)
def generate_dummyheaders():
def generate_rndstrs(strings, length):
return ''.join(random.choice(strings) for _ in range(length))
import string
return ['X-%s: %s\r\n' % (generate_rndstrs(string.ascii_uppercase, 16),
generate_rndstrs(string.ascii_letters + string.digits, 128)) for _ in range(32)]
req_writer.writelines(list(map(lambda x: x.encode(), generate_dummyheaders())))
yield from req_writer.drain()
req_writer.write(b'Host: ')
yield from req_writer.drain()
def feed_phost(phost):
i = 1
while phost:
yield random.randrange(2, 4), phost[:i]
phost = phost[i:]
i = random.randrange(2, 5)
for delay, c in feed_phost(phost):
yield from asyncio.sleep(delay / 10.0, loop=loop)
req_writer.write(c.encode())
yield from req_writer.drain()
req_writer.write(b'\r\n')
req_writer.writelines(list(map(lambda x: (x + '\r\n').encode(), sreq)))
req_writer.write(b'\r\n')
if payload != b'':
req_writer.write(payload)
req_writer.write(b'\r\n')
yield from req_writer.drain()
try:
while True:
buf = yield from req_reader.read(1024)
if len(buf) == 0:
break
client_writer.write(buf)
except:
print_exc()
except:
print_exc()
client_writer.close()
@asyncio.coroutine
def start_warp_server(host, port, *, loop = None):
try:
accept = functools.partial(accept_client, loop=loop)
server = yield from asyncio.start_server(accept, host=host, port=port, loop=loop)
except OSError as ex:
logger.critical('!!! Failed to bind server at [%s:%d]: %s' % (host, port, ex.args[1]))
raise
else:
logger.info('Server bound at [%s:%d].' % (host, port))
return server
def main():
"""CLI frontend function. It takes command line options e.g. host,
port and provides `--help` message.
"""
parser = ArgumentParser(description='Simple HTTP transparent proxy')
parser.add_argument('-H', '--host', default='127.0.0.1',
help='Host to listen [default: %(default)s]')
parser.add_argument('-p', '--port', type=int, default=8800,
help='Port to listen [default: %(default)d]')
parser.add_argument('-v', '--verbose', action='count', default=0,
help='Print verbose')
args = parser.parse_args()
if not (1 <= args.port <= 65535):
parser.error('port must be 1-65535')
if args.verbose >= 3:
parser.error('verbose level must be 1-2')
if args.verbose >= 1:
logger.setLevel(logging.DEBUG)
if args.verbose >= 2:
logging.getLogger('warp').setLevel(logging.DEBUG)
logging.getLogger('asyncio').setLevel(logging.DEBUG)
global verbose
verbose = args.verbose
loop = asyncio.get_event_loop()
try:
loop.run_until_complete(start_warp_server(args.host, args.port))
loop.run_forever()
except OSError:
pass
except KeyboardInterrupt:
print('bye')
finally:
loop.close()
if __name__ == '__main__':
exit(main())
PS: The code has been taken from https://github.com/devunt/warp. The complete code is posted here so that if the link becomes invalid in future, the answer is still valid.