Intercept and manipulate HTTPs traffic with Python and mitmproxy

intercept-https-with-python-mitmproxy.md

Intercepts HTTPs Traffic with Python & mitmproxy

Warning

This Gist is created in 2014, and it's highliy outdated now, according to one of mitmproxy's manjor contributor (check his comment below). Thanks for letting us know, @mhils!

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)

I'd like to point out that this gist is highly outdated at this time. Our main repo has long moved to https://github.com/mitmproxy/mitmproxy and we have considerably easier installation instructions in the docs on https://mitmproxy.org/. Also, you don't need to generate your own certificates or subclass the mitmproxy Master class.

(Writing this as I still receive Google Alerts for forks from time to time)

Hi @mhils,

Thanks for letting us know.
I have added a Warning section in the front of this Gist.

As this is one of the links that comes at the top if you search for intercepting network requests through python, updating gist for the newer version of mitmproxy:

from mitmproxy import flowfilter


class Intercept:
    def __init__(self, domain):
        self.intercept = domain

    def response(self, flow):
        if flow.request.host == self.intercept
           flow.intercept()
        #    print(flow.request.query)
           flow.request.query['whatever parameter you want to change'] = value
           flow.resume()
        #    print(flow.response)
       
def start():
    return Intercept("www.insertdomainyouwanttointercept.com")

To run this script, save it in a file and run it using mitmdump -q -s file.py. -q flag is for quiet mode.