intercept-https-with-python-mitmproxy.md

Intercepts HTTPs Traffic with Python & mitmproxy

Introduction

Modern applications usually make use of back-end API servers to provide their services. With a non-transparent HTTPs proxy, which intercepts the communication between clients and servers (aka the man-in-the-middle scheme), you can easily manipulate both API requests and responses.

This manual helps you create your own proxy with Python and mitmproxy/libmproxy. Mitmproxy ships with both a standalone command-line tool (mitmproxy) and a Python library (libmproxy).

Requirements

• Python, >= 2.7.3
  • install through brew install python on Mac OS X
• mitmproxy
  1. git clone https://github.com/cortesi/mitmproxy.git
  2. cd mitmproxy
  3. sudo python setup.py install
• netlib, version matching mitmproxy
  1. git clone https://github.com/cortesi/netlib.git
  2. cd netlib
  3. sudo python setup.py install
• PyOpenSSL, >= 0.13
  1. install OpenSSL development package (through sudo apt-get install libssl-dev on Ubuntu)
  2. download pyOpenSSL-0.13.tar.gz from pyOpenSSL project page on PyPI website
  3. tar xvf pyOpenSSL-0.13.tar.gz
  4. cd pyOpenSSL-0.13
  5. python setup.py build
  6. sudo python setup.py install
• pyasn1, >= 0.1.2
  • pip install pyasn1

Note: In my experience, mitmproxy depends on the latest netlib and PyOpenSSL, which cannot be installed from Pip. You may download the source tarball and install them manually.

Generate SSL Private Key and Certificate

1. openssl genrsa -out mitmproxy.key 2048
2. openssl req -new -x509 -key mitmproxy.key -out mitmproxy.crt -days 3650 -subj /CN=MitmProxy
3. cat mitmproxy.key mitmproxy.crt > mitmproxy.pem
4. install mitmproxy.crt on you device (desktop browser, iPhone, Android, etc.)

Example of proxy.py

#!/usr/bin/env python
# -*- encoding: utf-8 -*-
from libmproxy import controller, proxy
import os, sys, re, datetime, json

class RequestHacks:
  @staticmethod
  def example_com (msg):
    # tamper outgoing requests for https://example.com/api/v2
    if ('example.org' in msg.host) and ('action=login' in msg.content):
      fake_lat, fake_lng = 25.0333, 121.5333
      tampered = re.sub('lat=([\d.]+)&lng=([\d.]+)', 'lat=%s&lng=%s' % (fake_lat, fake_lng), msg.content)
      msg.content = tampered
      print '[RequestHacks][Example.com] Fake location (%s, %s) sent when logging in' % (fake_lat, fake_lng)


class ResponseHacks:
  @staticmethod
  def example_org (msg):
    # simple substitution for https://example.org/api/users/:id.json
    if 'example.org' in msg.request.host:
      regex = re.compile('/api/users/(\d+).json')
      match = regex.search(msg.request.path)
      if match and msg.content:
        c = msg.replace(''private_data_accessible':false', ''private_data_accessible':true')
        if c > 0:
          user_id = match.groups()[0]
          print '[ResponseHacks][Example.org] Private info of user #%s revealed' % user_id

  @staticmethod
  def example_com (msg):
    # JSON manipulation for https://example.com/api/v2
    if ('example.com' in msg.request.host) and ('action=user_profile' in msg.request.content):
      msg.decode() # need to decode the message first
      data = json.loads(msg.content) # parse JSON with decompressed content
      data['access_granted'] = true
      msg.content = json.dumps(data) # write back our changes
      print '[ResponseHacks][Example.com] Access granted of user profile #%s' % data['id']

  @staticmethod
  def example_net (msg):
    # Response inspection for https://example.net
    if 'example.net' in msg.request.host:
      data = msg.get_decoded_content() # read decompressed content without modifying msg
      print '[ResponseHacks][Example.net] Respones: %s' % data


class InterceptingMaster (controller.Master):
  def __init__ (self, server):
    controller.Master.__init__(self, server)

  def run (self):
    while True:
      try:
        controller.Master.run(self)
      except KeyboardInterrupt:
        print 'KeyboardInterrupt received. Shutting down'
        self.shutdown()
        sys.exit(0)
      except Exception:
        print 'Exception catched. Intercepting proxy restarted'
        pass

  def handle_request (self, msg):
    timestamp = datetime.datetime.today().strftime('%Y/%m/%d %H:%M:%S')
    client_ip = msg.client_conn.address[0]
    request_url = '%s://%s%s' % (msg.scheme, .msg.host, msg.path)
    print '[%s %s] %s %s' % (timestamp, client_ip, msg.method, request_url)

    RequestHacks.example_com(msg)
    msg.reply()

  def handle_response (self, msg):
    ResponseHacks.example_org(msg)
    ResponseHacks.example_com(msg)
    ResponseHacks.example_net(msg)
    msg.reply()


def main (argv):
  config = proxy.ProxyConfig(
    cacert = os.path.expanduser('./mitmproxy.pem'),
  )
  server = proxy.ProxyServer(config, 8080)
  print 'Intercepting Proxy listening on 8080'
  m = InterceptingMaster(server)
  m.run()

if __name__ == '__main__':
  main(sys.argv)