code.vuplus.com Git - vuplus_dvbapp-plugin/blob - genuinedreambox/src/plugin.py

   1 # -*- coding: utf-8 -*-
   2 ###########################################################################
   3 #
   4 # http://newnigma2.to
   5 #
   6 # $Id:
   7 #
   8 # Copyright (C) 2009 by
   9 # <nixkoenner@newnigma2.to>
  10 #
  11 #          License: GPL
  12 #
  13 #          This program is free software; you can redistribute it and/or modify
  14 #          it under the terms of the GNU General Public License as published by
  15 #          the Free Software Foundation; either version 2 of the License, or
  16 #          (at your option) any later version.
  17 #
  18 #          This program is distributed in the hope that it will be useful,
  19 #          but WITHOUT ANY WARRANTY; without even the implied warranty of
  20 #          MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  21 #          GNU General Public License for more details.
  22 #
  23 #          You should have received a copy of the GNU General Public License
  24 #          along with this program; if not, write to the Free Software
  25 #          Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  26 #
  27 ###########################################################################
  28 #
  29 # thx to <kayshadow@newnigma2.to> for painting the icon
  30 #
  31 from Plugins.Plugin import PluginDescriptor
  32 from Screens.Screen import Screen
  33 from Screens.MessageBox import MessageBox
  34
  35 from Components.ActionMap import ActionMap
  36 from Components.Button import Button
  37 from Components.Label import Label
  38
  39 import socket
  40 import struct
  41 import base64
  42 import os
  43
  44 from twisted.web.client import getPage
  45
  46 TPMD_DT_RESERVED = 0x00
  47 TPMD_DT_PROTOCOL_VERSION = 0x01
  48 TPMD_DT_TPM_VERSION     = 0x02
  49 TPMD_DT_SERIAL = 0x03
  50 TPMD_DT_LEVEL2_CERT = 0x04
  51 TPMD_DT_LEVEL3_CERT     = 0x05
  52 TPMD_DT_FAB_CA_CERT     = 0x06
  53 TPMD_DT_DATABLOCK_SIGNED = 0x07
  54 TPMD_CMD_RESERVED       = 0x0000
  55 TPMD_CMD_GET_DATA       = 0x0001
  56 TPMD_CMD_APDU   = 0x0002
  57 TPMD_CMD_COMPUTE_SIGNATURE = 0x0003
  58 TPMD_CMD_APP_CERT = 0x0004
  59 TPMD_PV_2 = 0x02
  60
  61 class genuineDreambox(Screen):
  62         skin = """
  63                 <screen position="center,center" size="620,420" title="%s" >
  64                 <widget name="infotext" position="10,20" zPosition="1" size="600,150" font="Regular;20" halign="center" valign="center" />
  65                 <widget name="resulttext" position="10,160" zPosition="1" size="600,110" font="Regular;20" halign="center" valign="center" />
  66                 <widget name="infotext2" position="10,280" zPosition="1" size="600,80" font="Regular;20" halign="center" valign="center" />
  67                 <widget name="kRed" position="185,365" zPosition="5" size="140,40" valign="center" halign="center" font="Regular;21" transparent="1" foregroundColor="white" shadowColor="black" shadowOffset="-1,-1" />
  68                 <ePixmap name="red" position="185,365" zPosition="4" size="140,40" pixmap="skin_default/buttons/red.png" transparent="1" alphatest="on" />
  69                 <widget name="kGreen" position="330,365" zPosition="5" size="140,40" valign="center" halign="center" font="Regular;21" transparent="1" foregroundColor="white" shadowColor="black" shadowOffset="-1,-1" />
  70                 <ePixmap name="green" position="330,365" zPosition="4" size="140,40" pixmap="skin_default/buttons/green.png" transparent="1" alphatest="on" />
  71                 </screen>"""% _("Genuine Dreambox")
  72
  73         def __init__(self, session):
  74                 Screen.__init__(self, session)
  75                 self["actions"] = ActionMap(["SetupActions", "ColorActions"],
  76                 {
  77                         "green": self.restart,
  78                         "cancel": self.exit,
  79                  }, -1)
  80                 self["kGreen"] = Button(_("Test again"))
  81                 self["kRed"] = Button(_("Cancel"))
  82                 self["infotext"] = Label("With this plugin you can verify the authenticity of your Dreambox.\nFor additional information, \nplease visit our website \nhttps://www.dream-multimedia-tv.de.")
  83                 self["resulttext"] = Label("... Please wait ...")
  84                 self["infotext2"] = Label("Please visit our website and follow the instructions.\nAlternatively you can call our customer service hotline.")
  85                 self.onLayoutFinish.append(self.start)
  86
  87         def restart(self):
  88                 if not self.isStart:
  89                         self.start()
  90
  91         def start(self):
  92                 udsError = False
  93                 self.isStart = True
  94                 try:
  95                         self["resulttext"].setText("Please wait (Step 1)")
  96                         self.uds = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
  97                         self.uds.connect(("/var/run/tpmd_socket"))
  98                         self.uds.settimeout(5.0)
  99                 except:
 100                         self["resulttext"].setText("Security service not running.")
 101                         udsError = True
 102                 if not udsError:
 103                         if (self.stepFirst(TPMD_CMD_GET_DATA,[TPMD_DT_PROTOCOL_VERSION,TPMD_DT_TPM_VERSION,TPMD_DT_SERIAL])):
 104                                 try:
 105                                         url = ("https://www.dream-multimedia-tv.de/verify/challenge?serial=%s&version=%s" % (self.serial,self.tpmdVersion))
 106                                         getPage(url).addCallback(self._gotPageLoadRandom).addErrback(self.errorLoad)
 107                                 except:
 108                                         self["resulttext"].setText(_("Can't connect to server. Please check your network!"))
 109
 110         def needsTPMUpdate(self):
 111                 return self.level3_cert is None
 112
 113         def updateCallback(self, result):
 114                 if result:
 115                         self.isStart = True
 116                         url = self.buildUrlUpdate()
 117                         #url = ("https://www.dream-multimedia-tv.de/verify/challenge?serial=%s&version=%s" % (self.serial,self.tpmdVersion))
 118                         self["resulttext"].setText(_("Updating, please wait..."))
 119                         getPage(url).addCallback(self._gotPageLoadUpdate).addErrback(self.errorLoad)
 120                 else:
 121                         print "not updating"
 122
 123         def _gotPageLoad(self, data):
 124                 authcode = data.strip().replace('+', '')
 125                 if len(authcode) == 12:
 126                         self.finish = "%s-%s-%s" % (authcode[0:4], authcode[4:8], authcode[8:12])
 127                         self["resulttext"].setText(self.finish)
 128                         if self.needsTPMUpdate():
 129                                 if int(self.protocolVersion) >= TPMD_PV_2:
 130                                         self.session.openWithCallback(self.updateCallback, MessageBox, _("There's a certificate update available for your dreambox. Would you like to apply this update now?"))
 131                 else:
 132                         self["resulttext"].setText(_("Invalid response from server."))
 133                 self.closeUds()
 134                 self.isStart = False
 135
 136         def _gotPageLoadRandom(self, data):
 137                 self["resulttext"].setText(_("Please wait (Step 2)"))
 138                 self.back = data.strip()
 139                 self.random = (self.formatList(base64.b64decode(self.back)))
 140                 self.level2_cert = None
 141                 self.level3_cert = None
 142                 if (self.stepSecond(TPMD_CMD_GET_DATA,[TPMD_DT_PROTOCOL_VERSION,TPMD_DT_TPM_VERSION,TPMD_DT_SERIAL,TPMD_DT_LEVEL2_CERT,
 143                                 TPMD_DT_LEVEL3_CERT,TPMD_DT_FAB_CA_CERT,TPMD_DT_DATABLOCK_SIGNED] )):
 144                         url = self.buildUrl()
 145                         getPage(url).addCallback(self._gotPageLoad).addErrback(self.errorLoad)
 146
 147         def _gotPageLoadUpdate(self, data):
 148                 updatedata = base64.decodestring(data)
 149                 if len(updatedata) != 409:
 150                         self["resulttext"].setText(_("Updating failed. Nothing is broken, just the update couldn't be applied."))
 151                         self.isStart = False
 152                 else:
 153                         udsError = False
 154
 155                         try:
 156                                 self.uds = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 157                                 self.uds.connect(("/var/run/tpmd_socket"))
 158                                 self.uds.settimeout(5.0)
 159                         except:
 160                                 self["resulttext"].setText(_("Security service not running."))
 161                                 udsError = True
 162                         if not udsError:
 163                                 data = [chr(0)]
 164                                 data.extend(updatedata)
 165                                 data = self.formatList(data)
 166                                 self.udsSend(TPMD_CMD_APP_CERT, data, 410)
 167                                 self["resulttext"].setText(_("Update done..."))
 168                                 self.closeUds()
 169                                 self.isStart = False
 170                                 self.session.openWithCallback(self.updateFinished, MessageBox, _("Update done... The genuine dreambox test will now be rerun and should not ask you to update again."), MessageBox.TYPE_INFO)
 171
 172         def updateFinished(self, result):
 173                 self.start()
 174
 175         def errorLoad(self, error):
 176                 print str(error)
 177                 self["resulttext"].setText(_("Invalid response from server. Please report: %s") % str(error))
 178
 179         def buildUrl(self):
 180                 # NOTE: this is a modified base64 which uses -_ instead of +/ to avoid the need for escpaing + when using urlencode
 181                 tmpra = ("random=%s" % self.back.replace('+', '-').replace('/', '_'))
 182                 tmpl2 = ("&l2=%s" % base64.b64encode(self.level2_cert).replace('+', '-').replace('/', '_'))
 183                 if self.level3_cert is not None:
 184                         tmpl3 = ("&l3=%s" % base64.b64encode(self.level3_cert).replace('+', '-').replace('/', '_'))
 185                 else:
 186                         tmpl3 = ""
 187                 tmpfa = ("&fab=%s" % base64.b64encode(self.fab_ca_cert).replace('+', '-').replace('/', '_'))
 188                 tmpda = ("&data=%s" % base64.b64encode(self.datablock_signed).replace('+', '-').replace('/', '_'))
 189                 tmpr  = ("&r=%s" % base64.b64encode(self.r).replace('+', '-').replace('/', '_'))
 190                 return("https://www.dream-multimedia-tv.de/verify/challenge?%s%s%s%s%s%s&serial=%s" % (tmpra,tmpl2,tmpl3,tmpfa,tmpda,tmpr,self.serial))
 191
 192         def buildUrlUpdate(self):
 193                 return self.buildUrl() + "&getupdate=true"
 194
 195         def formatList(self,l):
 196                 liste = []
 197                 for x in l:
 198                         liste.append(ord(x))
 199                 return liste
 200
 201         def formatString(self,s):
 202                 myString = ""
 203                 for x in s:
 204                         myString =  myString + chr(x)
 205                 return myString
 206
 207         def stepFirst(self,typ,daten):
 208                 return (self.parseResult (self.udsSend(typ,daten,len(daten))))
 209
 210         def stepSecond(self,typ,daten):
 211                 if (self.parseResult(self.udsSend(typ,daten,len(daten))) == False):
 212                         return False
 213                 if (self.parseSignature(self.udsSend(TPMD_CMD_COMPUTE_SIGNATURE,self.random,8)) == False):
 214                         return False
 215                 return True
 216
 217         def parseResult(self,rbuf):
 218                 if (rbuf != -1):
 219                         buf = self.formatList(rbuf)
 220
 221                         pos = 0
 222                         while pos < len(buf):
 223                                 tag = buf[pos]
 224                                 length = buf[pos + 1]
 225                                 value = buf[pos + 2: pos + 2 + length]
 226                                 if tag == TPMD_DT_PROTOCOL_VERSION:
 227                                         if length != 1:
 228                                                 self.protocolVersion = None
 229                                         else:
 230                                                 self.protocolVersion = "%d" % value[0]
 231                                 elif tag == TPMD_DT_TPM_VERSION:
 232                                         if length != 1:
 233                                                 self.tpmdVersion = None
 234                                         else:
 235                                                 self.tpmdVersion = "%d" % value[0]
 236                                 elif tag == TPMD_DT_SERIAL:
 237                                         if length != 4:
 238                                                 self.serial = None
 239                                         else:
 240                                                 self.serial = "%d" % ((value[0] << 24) | (value[1] << 16) | (value[2] << 8) | value[3])
 241                                 elif tag == TPMD_DT_LEVEL2_CERT:
 242                                         if length != 210:
 243                                                 self.level2_cert = None
 244                                         else:
 245                                                 self.level2_cert = ''.join([chr(x) for x in value])
 246                                 elif tag == TPMD_DT_LEVEL3_CERT:
 247                                         if length != 210:
 248                                                 self.level3_cert = None
 249                                         else:
 250                                                 self.level3_cert = ''.join([chr(x) for x in value])
 251                                 elif tag == TPMD_DT_FAB_CA_CERT:
 252                                         if length != 210:
 253                                                 self.fab_ca_cert = None
 254                                         else:
 255                                                 self.fab_ca_cert = ''.join([chr(x) for x in value])
 256                                 elif tag == TPMD_DT_DATABLOCK_SIGNED:
 257                                         if length != 128:
 258                                                 self.datablock_signed = None
 259                                         else:
 260                                                 self.datablock_signed = ''.join([chr(x) for x in value])
 261                                 else:
 262                                         print "unknown tag:", tag
 263                                 pos += 2 + length
 264
 265                         return True
 266                 else:
 267                         return False
 268
 269         def parseSignature(self, rbuf):
 270                 if (rbuf != -1):
 271                         self.r = self.formatString(self.formatList(rbuf))
 272                         return True
 273                 else:
 274                         return False
 275
 276         def udsSend(self, cmdTyp, data, length):
 277                 udsError = False
 278                 sbuf = [(cmdTyp >> 8) & 0xff,(cmdTyp >> 0) & 0xff,(length >> 8) & 0xff,(length >> 0) & 0xff]
 279                 sbuf.extend(data[:length])
 280                 sbuf = struct.pack(str((length + 4))+"B", *sbuf)
 281                 try:
 282                         self.uds.send(sbuf)
 283                         udsError = False
 284                 except socket.timeout:
 285                         udsError = True
 286                 try:
 287                         rbuf = self.uds.recv(4)
 288                         udsError = False
 289                 except socket.timeout:
 290                         udsError = True
 291
 292                 res = -1
 293
 294                 if (udsError == False):
 295                         leng = [ord(rbuf[2]) << 8 | ord(rbuf[3])]
 296                         if (leng != 4):
 297                                 try:
 298                                         res = self.uds.recv(leng[0])
 299                                 except socket.timeout:
 300                                         udsError = True
 301                         else:
 302                                 return -1
 303                 else:
 304                         self["resulttext"].setText(_("Invalid response from Security service pls restart again"))
 305                         os.system("kill -9 $(pidof tpmd)")
 306                         return -1
 307                 return res
 308
 309         def closeUds(self):
 310                 try:
 311                         self.uds.close()
 312                 except:
 313                         pass
 314
 315         def exit(self):
 316                 self.closeUds()
 317                 self.close()
 318
 319 def main(session, **kwargs):
 320                 session.open(genuineDreambox)
 321
 322 def Plugins(path,**kwargs):
 323                 global plugin_path
 324                 plugin_path = path
 325                 return [
 326                                 PluginDescriptor(name="Genuine Dreambox", description="Genuine Dreambox",where = PluginDescriptor.WHERE_PLUGINMENU, icon="genuine.png", fnc=main),
 327                                 PluginDescriptor(name="Genuine Dreambox", where = PluginDescriptor.WHERE_EXTENSIONSMENU, fnc=main)
 328                                 ]