Created
June 6, 2011 00:52
-
-
Save arantius/1009602 to your computer and use it in GitHub Desktop.
The files I used to reverse engineer, and proof-of-concept, the TiVo to iPad protocol.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
"""An mDNS daemon designed to appear, to the iPad app, to be a TiVo Premiere.""" | |
import Zeroconf | |
import socket | |
local_ip = socket.gethostbyname(socket.gethostname()) | |
local_ip = socket.inet_aton(local_ip) | |
# You need the right value here. Either use tcpdump to find the value that | |
# your TiVo device advertises, or navigate to: | |
# Account & System Info | |
# System Information | |
# TiVo Service Number | |
# It should be 15 hex digits (0-9 and a-f) with no dashes. All in uppercase | |
# to be safe. | |
TSN = 'XXXXXXXXXXXXXXX' | |
server = Zeroconf.Zeroconf() | |
server.registerService( | |
Zeroconf.ServiceInfo( | |
'_tivo-device._tcp.local.', | |
'Proxy._tivo-device._tcp.local.', | |
address = local_ip, | |
port = 80, | |
weight = 0, priority=0, | |
properties = { | |
'path': '/', | |
'services': '_tivo-mindrpc._tcp,_tivo-remote._tcp', | |
'platformname': 'TiVo Premiere', | |
'swversion': '14.8.U2-01-3.746', | |
'platform': 'tcd/Series4', | |
'TSN': TSN, | |
} | |
) | |
) | |
server.registerService( | |
Zeroconf.ServiceInfo( | |
'_http._tcp.local.', | |
'Proxy._http._tcp.local.', | |
address = local_ip, | |
port = 80, | |
weight = 0, priority=0, | |
properties = { | |
'path': '/index.html', | |
'swversion': '14.8.U2-01-3.746', | |
'platform': 'tcd/Series4', | |
'TSN': TSN, | |
} | |
) | |
) | |
server.registerService( | |
Zeroconf.ServiceInfo( | |
'_tivo-mindrpc._tcp.local.', | |
'Proxy._tivo-mindrpc._tcp.local.', | |
address = local_ip, | |
port = 1413, | |
weight = 0, priority=0, | |
properties = { | |
'protocol': 'tivo-mindrpc', | |
'path': '/', | |
'swversion': '14.8.U2-01-3.746', | |
'platform': 'tcd/Series4', | |
'TSN': TSN, | |
} | |
) | |
) | |
server.registerService( | |
Zeroconf.ServiceInfo( | |
'_tivo-videos._tcp.local.', | |
'Proxy._tivo-videos._tcp.local.', | |
address = local_ip, | |
port = 443, | |
weight = 0, priority=0, | |
properties = { | |
'protocol': 'https', | |
'path': '/TiVoConnect?Command=QueryContainer&Container=%2FNowPlaying', | |
'swversion': '14.8.U2-01-3.746', | |
'platform': 'tcd/Series4', | |
'TSN': TSN, | |
} | |
) | |
) | |
print 'Running mDNS daemon ...' |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
import logging | |
import socket | |
import ssl | |
import sys | |
import simplejson | |
# Put your appropriate details here. | |
tivo_addr = '255.255.255.255' | |
mak = 'XXXXXXXXXX' | |
rpc_id = 0 | |
def RpcRequest(type, monitor=False, **kwargs): | |
global rpc_id | |
rpc_id += 1 | |
req = dict(**kwargs) | |
req.update({'type': type}) | |
obj = { | |
'type': 'mindRpcRequest', | |
'request': req, | |
'rpcId': rpc_id, | |
'mindVersion': 7, | |
'monitorFutureChanges': monitor, | |
} | |
return obj | |
class Remote(object): | |
def __init__(self): | |
self.buf = '' | |
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
self.ssl_socket = ssl.wrap_socket(self.socket) | |
self.ssl_socket.connect((tivo_addr, 1393)) | |
self.Auth() | |
def Read(self): | |
while True: | |
buf = self.ssl_socket.read(128) | |
self.buf += buf | |
if '\x00' in buf: | |
break | |
strs = self.buf.split('\x00', 1) | |
logging.info('READ %s', strs[0]) | |
obj = simplejson.loads(strs[0]) | |
self.buf = strs[1] | |
return obj | |
def Write(self, obj): | |
str = simplejson.dumps(obj) | |
logging.info('SEND %s', str) | |
self.ssl_socket.send(str + '\x00') | |
def Auth(self): | |
self.Write(RpcRequest('bodyAuthenticate', | |
credential={ | |
'type': 'makCredential', | |
'key': mak, | |
} | |
)) | |
result = self.Read() | |
if result['response']['status'] != 'success': | |
logging.error('Authentication failed! Got: %s', result) | |
sys.exit(1) | |
def Key(self, key): | |
"""Send a key. | |
Supported: | |
Letters 'a' through 'z'. | |
Numbers '0' through '1'. | |
Named keys: | |
actionA, actionB, actionC, actionD, advance, channelDown, | |
channelUp, clear, down, enter, forward, guide, info, left, liveTv, | |
pause, play, record, replay, reverse, right, select, slow, | |
thumbsDown, thumbsUp, tivo, up, zoom | |
""" | |
if key == ' ': | |
key = 'forward' | |
if key.lower() in 'abcdefghijklmnopqrstuvwxyz': | |
req = RpcRequest('keyEventSend', event='ascii', value=ord(key)) | |
elif key in '0123456789': | |
req = RpcRequest('keyEventSend', event='num' + key) | |
else: | |
req = RpcRequest('keyEventSend', event=key) | |
self.Write(req) | |
result = self.Read() | |
if result['response']['type'] != 'success': | |
logging.error('Pause failed! Got: %s', result) | |
sys.exit(1) | |
if __name__ == '__main__': | |
logging.basicConfig(stream=sys.stderr, level=logging.INFO) | |
remote = Remote() | |
remote.Key('pause') | |
for key in 'typed via remote': | |
remote.Key(key) | |
remote.Key('pause') |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
import logging | |
import random | |
import re | |
import socket | |
import ssl | |
import sys | |
import time | |
import simplejson | |
tivo_addr = '255.255.255.255' | |
mak = 'XXXXXXXXXX' | |
body_id = '' | |
rpc_id = 0 | |
session_id = random.randrange(0x26c000, 0x27dc20) | |
def RpcRequest(type, monitor=False, **kwargs): | |
global rpc_id | |
rpc_id += 1 | |
headers = '\r\n'.join(( | |
'Type: request', | |
'RpcId: %d' % rpc_id, | |
'SchemaVersion: 7', | |
'Content-Type: application/json', | |
'RequestType: %s' % type, | |
'ResponseCount: %s' % (monitor and 'multiple' or 'single'), | |
'BodyId: %s' % body_id, | |
'X-ApplicationName: Quicksilver', | |
'X-ApplicationVersion: 1.2', | |
'X-ApplicationSessionId: 0x%x' % session_id, | |
)) + '\r\n' | |
req_obj = dict(**kwargs) | |
req_obj.update({'type': type}) | |
body = simplejson.dumps(req_obj) + '\n' | |
# The "+ 2" is for the '\r\n' we'll add to the headers next. | |
start_line = 'MRPC/2 %d %d' % (len(headers) + 2, len(body)) | |
return '\r\n'.join((start_line, headers, body)) | |
class Remote(object): | |
def __init__(self): | |
self.buf = '' | |
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) | |
self.ssl_socket = ssl.wrap_socket(self.socket) | |
self.ssl_socket.connect((tivo_addr, 1413)) | |
self.Auth() | |
def Read(self): | |
start_line = '' | |
head_len = None | |
body_len = None | |
while True: | |
self.buf += self.ssl_socket.read(16) | |
match = re.match(r'MRPC/2 (\d+) (\d+)\r\n', self.buf) | |
if match: | |
start_line = match.group(0) | |
head_len = int(match.group(1)) | |
body_len = int(match.group(2)) | |
break | |
need_len = len(start_line) + head_len + body_len | |
while len(self.buf) < need_len: | |
self.buf += self.ssl_socket.read(1024) | |
buf = self.buf[:need_len] | |
self.buf = self.buf[need_len:] | |
logging.debug('READ %s', buf) | |
return simplejson.loads(buf[-1 * body_len:]) | |
def Write(self, data): | |
logging.debug('SEND %s', data) | |
self.ssl_socket.send(data) | |
def Auth(self): | |
self.Write(RpcRequest('bodyAuthenticate', | |
credential={ | |
'type': 'makCredential', | |
'key': mak, | |
} | |
)) | |
result = self.Read() | |
if result['status'] != 'success': | |
logging.error('Authentication failed! Got: %s', result) | |
sys.exit(1) | |
def Key(self, key): | |
"""Send a key. | |
Supported: | |
Letters 'a' through 'z'. | |
Numbers '0' through '1'. | |
Named keys: | |
actionA, actionB, actionC, actionD, advance, channelDown, | |
channelUp, clear, down, enter, forward, guide, info, left, liveTv, | |
pause, play, record, replay, reverse, right, select, slow, | |
thumbsDown, thumbsUp, tivo, up, zoom | |
A space is turned into the 'fast forward' button, as that's what the | |
TiVo normally expects, where a space is a valid character to be | |
entering. | |
""" | |
if key == ' ': | |
key = 'forward' | |
if key.lower() in 'abcdefghijklmnopqrstuvwxyz': | |
req = RpcRequest('keyEventSend', event='ascii', value=ord(key)) | |
elif key in '0123456789': | |
req = RpcRequest('keyEventSend', event='num' + key) | |
else: | |
req = RpcRequest('keyEventSend', event=key) | |
self.Write(req) | |
result = self.Read() | |
if result['type'] != 'success': | |
logging.error('Pause failed! Got: %s', result) | |
sys.exit(1) | |
if __name__ == '__main__': | |
logging.basicConfig(stream=sys.stderr, level=logging.INFO) | |
remote = Remote() | |
remote.Key('pause') | |
# for key in 'typed via remote': | |
# remote.Key(key) | |
time.sleep(2) | |
remote.Key('pause') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
I want to use this for testing but I dont know Python. I looks close enough to other languages I may be able to hang in there but I am puzzled about where in the code you specify the p12 or cert/key files for the certificate to authenticate?