*update_transaction_codes.py* parses Android services source code to find out transaction codes and saves them in YAML format for later use. *transaction_codes.yaml* is an example of the resulting file. *service_call_test.py* shows how it can be used

service_call_test.py

#!/usr/bin/python

import os
import sys
import time
import yaml
import subprocess
import re
import struct

ADB_PATH = '/android/platform-tools/adb'
YAML_FILE = 'transaction_codes.yaml'
TRANSACTION_CODES = {svc:{} for svc in 'phone iphonesubinfo isms usb wifi window input connectivity clipboard'.split()}

def load_TRANSACTION_CODES():
    global TRANSACTION_CODES, YAML_FILE
    try:
        with open(YAML_FILE, 'r') as f:
            TRANSACTION_CODES = yaml.safe_load(f.read())
    except Exception as e:
        pass

def adbdevices():
    global ADB_PATH
    return set([device.split('\t')[0] for device in subprocess.check_output([ADB_PATH, 'devices']).splitlines() if device.endswith('\tdevice')])

def adbshell(command):
    global SERIAL, ADB_PATH
    return subprocess.check_output([ADB_PATH, '-s', SERIAL, 'shell', command]).decode('utf-8').replace('\r\n', '\n').rstrip('\n')

def get_android_version():
    return str(adbshell('getprop ro.build.version.release'))

def build_service_call_command(svc, func, *args):
    global VERSION, TRANSACTION_CODES
    code = TRANSACTION_CODES.get(svc, {}).get(func.lower(), {}).get(VERSION, 0)
    if code == 0:
        raise Exception('Unknown service function!')

    cmd = 'service call {} {}'.format(svc, code)
    for arg in args:
        if arg is None:
            cmd += ' null'
        elif isinstance(arg, int):
            cmd += ' i32 {}'.format(arg)
        elif isinstance(arg, (str, unicode)):
            cmd += ' s16 "{}"'.format(arg)
    return cmd

class Parcel(object):
    def __init__(self, text):
        if text.startswith('Result: Parcel(') and text.endswith('\')'):
            self.data = ''.join([ struct.pack('<L', int(x, 16)) for x in re.findall('([0-9a-f]{8}) ', text)])
            self.resultcode = self.get_int(0)
        else:
            raise Exception('Unexpected input!')

    def __nonzero__(self):
        return self.resultcode == 0

    def __repr__(self):
        return 'Parcel(\'{}\')'.format(self.data.encode('hex'))

    def get_int(self, offset=4):
        return int(struct.unpack('<L', buffer(self.data, offset, 4))[0])

    def get_utf16(self, offset=4):
        return str(buffer(self.data, offset + 4, self.get_int(offset) * 2)).decode('utf-16')

load_TRANSACTION_CODES()
SERIAL = sorted(adbdevices())[0]
VERSION = get_android_version()

print 'rotation =', Parcel(adbshell(build_service_call_command('window', 'getrotation'))).get_int()
print 'IMEI =', Parcel(adbshell(build_service_call_command('iphonesubinfo', 'getdeviceid'))).get_utf16()

update_transaction_codes.py

#!/usr/bin/python

import os
import sys
import time
import yaml
import requests
import subprocess
import base64
import re

ADB_PATH = 'adb'
GOOGLE_SOURCE_URL = 'https://android.googlesource.com'
BASE_REPO = 'platform/frameworks/base'
YAML_FILE = 'transaction_codes.yaml'
TRANSACTION_CODES = {svc:{} for svc in 'package phone'.split()}


def adbdevices():
    global ADB_PATH
    return set([device.split('\t')[0] for device in subprocess.check_output([ADB_PATH, 'devices']).splitlines() if device.endswith('\tdevice')])

def adbshell(command):
    global SERIAL, ADB_PATH
    return subprocess.check_output([ADB_PATH, '-s', SERIAL, 'shell', command]).decode('utf-8').replace('\r\n', '\n').rstrip('\n')

def load_yaml():
    global TRANSACTION_CODES, YAML_FILE
    try:
        with open(YAML_FILE, 'r') as f:
            TRANSACTION_CODES = yaml.safe_load(f.read())
    except Exception as e:
        pass

def save_yaml():
    global TRANSACTION_CODES, YAML_FILE
    try:
        with open(YAML_FILE, 'w') as f:
            f.write(yaml.dump(TRANSACTION_CODES, default_flow_style=False))
    except Exception as e:
        pass

def get_google_source(url, format='text', decode=True):
    try:
        req = requests.get(url, params={'format': format})
        if req.status_code == 200:
            if decode:
                return base64.decodestring(req.content)
            return req.content
    except Exception as e:
        pass
    return ''

def get_service_package_names():
    return dict(re.findall(r'^\d+\s+(\S+): \[(\S*)\]', str(adbshell('service list')), re.M))

def get_google_source_link(link, tag='+/master'):
    global GOOGLE_SOURCE_URL, BASE_REPO
    return link if link.startswith (('http:', 'https:')) else '{}/{}/{}/{}'.format(GOOGLE_SOURCE_URL, BASE_REPO, tag, link)

def get_service_package_links():
    global TAG
    return [aidl for aidl in ''.join([line for line in get_google_source(get_google_source_link('Android.mk', TAG)).replace('\\\n', '').splitlines()
            if line.startswith('LOCAL_SRC_FILES ')]).split() if aidl.endswith('.aidl')]

def get_android_version():
    return str(adbshell('getprop ro.build.version.release'))

def get_all_android_versions():
    return sorted(re.findall(r' refs/tags/android-([0-9.]+)_r1$', get_google_source(get_google_source_link('tags/', '+refs'), decode=False), re.M))

def get_android_tag(version):
    for tag in get_all_android_versions():
        if tag.startswith(version):
            return '+/android-{}_r1'.format(tag)

def get_function_names(src):
    src = src.lower()
    src = src[src.find('interface '):].replace('\n', ' ')
    for fn in src.split(';'):
        if fn.endswith(')'):
            yield fn.rsplit('(', 1)[0].rsplit(' ', 1)[-1]

load_yaml()

DEVICES = sorted(adbdevices().intersection(sys.argv[1:])) if len(sys.argv) > 1 else sorted(adbdevices())

for SERIAL in DEVICES:
    VERSION = get_android_version()
    TAG = get_android_tag(VERSION)
    PACKAGE_NAMES = get_service_package_names()
    PACKAGE_URLS = { pkg[0:-5].replace('/', '.'):pkg for pkg in get_service_package_links() }
    for svc_name in sorted(TRANSACTION_CODES.keys()):
        TRANSACTION_CODES[svc_name]['#PACKAGE_NAME{}'.format(TAG)] = PACKAGE_NAMES.get(svc_name, '#UNKNOWN#')
        link = None

        print SERIAL, VERSION, TAG, svc_name, PACKAGE_NAMES.get(svc_name, '#')

        for pkg in sorted(PACKAGE_URLS.keys()):
            if pkg.endswith(PACKAGE_NAMES.get(svc_name, '#')):
                link = PACKAGE_URLS.get(pkg)
                TRANSACTION_CODES[svc_name]['#PACKAGE_URL{}'.format(TAG)] = get_google_source_link(link, TAG)
                break

        if link is None:
            continue

        for i, fn in enumerate(get_function_names(get_google_source(get_google_source_link(link, TAG)))):
            TRANSACTION_CODES[svc_name].setdefault(fn, {})[VERSION] = i + 1
save_yaml()