Skip to content

Instantly share code, notes, and snippets.

@arantius
Created June 6, 2011 00:52
Show Gist options
  • Save arantius/1009602 to your computer and use it in GitHub Desktop.
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.
#!/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 ...'
#!/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')
#!/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')
@nevski99
Copy link

Hi,

I came across your python scripts to use tivo on ipad. This works great!
What would need to be changed to allow two tivo boxes to work with this?

Thanks

@arantius
Copy link
Author

arantius commented Jun 21, 2012 via email

@nevski99
Copy link

nevski99 commented Jun 21, 2012 via email

@Eric2XU
Copy link

Eric2XU commented Mar 22, 2016

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?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment