capture.py

from libmproxy.models import HTTPResponse
from netlib.http.headers import Headers
import cgi
import re
from gzip import GzipFile
import StringIO
import time

XML_OK_RESPONSE = '''<?xml version="1.0" encoding="UTF-8"?>
  <!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
  <plist version="1.0">
   <dict>
     <key>iPhone5,1</key><array><string>powerDiagnostics</string></array><key>iPhone5,2</key><array><string>powerDiagnostics</string></array><key>iPhone5,3</key><array><string>powerDiagnostics</string></array><key>iPhone5,4</key><array><string>powerDiagnostics</string></array><key>iPhone6,1</key><array><string>powerDiagnostics</string></array><key>iPhone6,2</key><array><string>powerDiagnostics</string></array><key>iPhone7,1</key><array><string>powerDiagnostics</string></array><key>iPhone7,2</key><array><string>powerDiagnostics</string></array><key>iPhone8,2</key><array><string>powerDiagnostics</string></array><key>iPhone8,1</key><array><string>powerDiagnostics</string></array>
   </dict>
</plist>'''


def request(context, flow):
    path = flow.request.path
    context.log('Path is %s' % path, 'error')
    #if path == '/ios/TestConfiguration/1.2':
        # context.log('sending OK', 'error')
        # respond(flow, XML_OK_RESPONSE)
    if path == '/MR3Server/ValidateTicket?ticket_number=123456':
        context.log('sending OK', 'error')
        respond(flow, XML_OK_RESPONSE)
    elif path == '/MR3Server/MR3Post':
        context.log('sending OK', 'error')
        saveContent(flow, 'general')
        respond(flow, XML_OK_RESPONSE)
    elif path == '/ios/log/extendedUpload':
        context.log('sending OK', 'error')
        saveContent(flow, 'power')
        respond(flow, XML_OK_RESPONSE)


def saveContent(flow, prefix):
    decodedData = StringIO.StringIO()
    decodedData.write(flow.request.get_decoded_content())

    contentType = flow.request.headers["Content-Type"]
    multipart_boundary_re = re.compile('^multipart/form-data; boundary=(.*)$')
    matches = multipart_boundary_re.match(contentType)

    decodedData.seek(0)

    query = cgi.parse_multipart( decodedData, {"boundary" : matches.group(1)})

    with open("%s-%s.tar.gz" % (prefix, time.strftime("%Y%m%d-%H%M%S")), "w") as logs:
        logs.write(query['log_archive'][0])

def respond(flow, content):
    resp = HTTPResponse([1,1], 200, "OK", Headers(Content_Type="text/xml"), content)
    flow.reply(resp)

it works with the irserver running on the home assistant server, (my irtrans has irdb), thanks,