warcprox/tests/single-threaded-proxy.py

#!/usr/bin/env python
"""
tests/single-threaded-proxy.py - single-threaded MITM proxy, useful for
debugging, does not write warcs

Copyright (C) 2015-2016 Internet Archive

This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; either version 2
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 General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
USA.
"""

from __future__ import absolute_import

import warcprox
import logging
import sys
import argparse
import certauth
import queue
import socket
import os

class FakeQueue(object):
    logger = logging.getLogger("FakeQueue")
    def __init__(self, maxsize=0): pass
    def join(self): pass
    def qsize(self): return 0
    def empty(self): return True
    def full(self): return False
    def get(self, block=True, timeout=None): raise queue.Empty
    def put_nowait(self, item): return self.put(item, block=False)
    def get_nowait(self): return self.get(block=False)
    def put(self, recorded_url, block=True, timeout=None):
        logging.info("{} {} {} {} {} size={} {}".format(
            recorded_url.client_ip, recorded_url.status, recorded_url.method,
            recorded_url.url.decode("utf-8"), recorded_url.mimetype,
            recorded_url.size, warcprox.digest_str(recorded_url.response_recorder.payload_digest, False).decode('utf-8')))

def parse_args():
    prog = os.path.basename(sys.argv[0])
    arg_parser = argparse.ArgumentParser(prog=prog,
        description='%s - single threaded mitm http/s proxy, for debugging' % prog,
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)
    arg_parser.add_argument('-p', '--port', dest='port', default='8000',
        type=int, help='port to listen on')
    arg_parser.add_argument('-b', '--address', dest='address',
        default='localhost', help='address to listen on')
    arg_parser.add_argument('-c', '--cacert', dest='cacert',
        default='./{0}-warcprox-ca.pem'.format(socket.gethostname()),
        help='CA certificate file; if file does not exist, it will be created')
    arg_parser.add_argument('--certs-dir', dest='certs_dir',
        default='./{0}-warcprox-ca'.format(socket.gethostname()),
        help='where to store and load generated certificates')
    arg_parser.add_argument('--onion-tor-socks-proxy', dest='onion_tor_socks_proxy',
        default=None, help='host:port of tor socks proxy, used only to connect to .onion sites')
    arg_parser.add_argument('--version', action='version',
        version="warcprox {}".format(warcprox.__version__))
    arg_parser.add_argument('-v', '--verbose', dest='verbose', action='store_true')
    arg_parser.add_argument('-q', '--quiet', dest='quiet', action='store_true')

    return arg_parser.parse_args(args=sys.argv[1:])

def init_logging(verbose):
    if args.verbose:
        loglevel = logging.DEBUG
    elif args.quiet:
        loglevel = logging.WARNING
    else:
        loglevel = logging.INFO

    logging.basicConfig(stream=sys.stdout, level=loglevel,
            format='%(asctime)s %(process)d %(levelname)s %(threadName)s %(name)s.%(funcName)s(%(filename)s:%(lineno)d) %(message)s')
            # format='%(asctime)s %(funcName) 21s() %(filename)15s:%(lineno)05d %(message)s')

def init_proxy(args):
    ca_name = 'Warcprox CA on {}'.format(socket.gethostname())[:64]
    ca = certauth.certauth.CertificateAuthority(args.cacert, args.certs_dir,
        ca_name=ca_name)
    options = warcprox.Options(**vars(args))
    proxy = warcprox.warcproxy.SingleThreadedWarcProxy(ca,
        recorded_url_q=FakeQueue(), options=options)
    return proxy

if __name__ == "__main__":
    args = parse_args()
    init_logging(args.verbose)
    proxy = init_proxy(args)

    proxy.serve_forever()