decoders/hdcp/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2018 Dave Craig<dcraig@brightsign.biz>
   5 ##
   6 ## This program is free software; you can redistribute it and/or modify
   7 ## it under the terms of the GNU General Public License as published by
   8 ## the Free Software Foundation; either version 2 of the License, or
   9 ## (at your option) any later version.
  10 ##
  11 ## This program is distributed in the hope that it will be useful,
  12 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ## GNU General Public License for more details.
  15 ##
  16 ## You should have received a copy of the GNU General Public License
  17 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 ##
  19
  20 import sigrokdecode as srd
  21
  22 msg_ids = {
  23     2: 'AKE_Init',
  24     3: 'AKE_Send_Cert',
  25     4: 'AKE_No_stored_km',
  26     5: 'AKE_Stored_km',
  27
  28     7: 'AKE_Send_H_prime',
  29     8: 'AKE_Send_Pairing_Info',
  30
  31     9: 'LC_Init',
  32     10: 'LC_Send_L_prime',
  33
  34     11: 'SKE_Send_Eks',
  35     12: 'RepeaterAuth_Send_ReceiverID_List',
  36
  37     15: 'RepeaterAuth_Send_Ack',
  38     16: 'RepeaterAuth_Stream_Manage',
  39     17: 'RepeaterAuth_Stream_Ready',
  40 }
  41
  42 class Decoder(srd.Decoder):
  43     api_version = 3
  44     id = 'hdcp'
  45     name = 'HDCP'
  46     longname = 'HDCP over HDMI'
  47     desc = 'HDCP protocol over HDMI.'
  48     license = 'gplv2+'
  49     inputs = ['i2c']
  50     outputs = ['hdcp']
  51     tags = ['PC', 'Security/crypto']
  52     annotations = \
  53         tuple(('message-0x%02X' % i, 'Message 0x%02X' % i) for i in range(18)) + (
  54         ('summary', 'Summary'),
  55         ('warnings', 'Warnings'),
  56     )
  57     annotation_rows = (
  58         ('messages', 'Messages', tuple(range(18))),
  59         ('summary', 'Summary', (18,)),
  60         ('warnings', 'Warnings', (19,)),
  61     )
  62
  63     def __init__(self):
  64         self.reset()
  65
  66     def reset(self):
  67         self.state = 'IDLE'
  68         self.stack = []
  69         self.msg = -1
  70         self.ss = self.es = self.ss_block = self.es_block = 0
  71         self.init_seq = []
  72         self.valid = 0
  73         self.type = ''
  74
  75     def start(self):
  76         self.out_ann = self.register(srd.OUTPUT_ANN)
  77
  78     def putb(self, data):
  79         self.put(self.ss_block, self.es_block, self.out_ann, data)
  80
  81     def decode(self, ss, es, data):
  82         cmd, databyte = data
  83
  84         # Collect the 'BITS' packet, then return. The next packet is
  85         # guaranteed to belong to these bits we just stored.
  86         if cmd == 'BITS':
  87             self.bits = databyte
  88             return
  89
  90         self.ss, self.es = ss, es
  91
  92         # State machine.
  93         if self.state == 'IDLE':
  94             # Wait for an I2C START condition.
  95             if cmd == 'START':
  96                 self.reset()
  97                 self.ss_block = ss
  98             elif cmd != 'START REPEAT':
  99                 return
 100             self.state = 'GET SLAVE ADDR'
 101
 102         elif self.state == 'GET SLAVE ADDR':
 103             if cmd == 'ADDRESS READ':
 104                 self.state = 'BUFFER DATA'
 105                 if databyte != 0x3a:
 106                     self.state = 'IDLE'
 107             elif cmd == 'ADDRESS WRITE':
 108                 self.state = 'WRITE OFFSET'
 109                 if databyte != 0x3a:
 110                     self.state = 'IDLE'
 111
 112         elif self.state == 'WRITE OFFSET':
 113             if cmd == 'DATA WRITE':
 114                 if databyte == 0x00:
 115                     self.type = '1.4 Bksv - Receiver KSV'
 116                 elif databyte == 0x08:
 117                     self.type = '1.4 Ri\' - Link Verification'
 118                 elif databyte == 0x0a:
 119                     self.type = '1.4 Pj\' - Enhanced Link Verification'
 120                 elif databyte == 0x10:
 121                     self.type = '1.4 Aksv - Transmitter KSV'
 122                     self.state = 'BUFFER DATA'
 123                 elif databyte == 0x15:
 124                     self.type = '1.4 Ainfo- Transmitter KSV'
 125                     self.state = 'BUFFER DATA'
 126                 elif databyte == 0x18:
 127                     self.type = '1.4 An- Session random number'
 128                     self.state = 'BUFFER DATA'
 129                 elif databyte == 0x20:
 130                     self.type = '1.4 V\'H0'
 131                 elif databyte == 0x24:
 132                     self.type = '1.4 V\'H1'
 133                 elif databyte == 0x28:
 134                     self.type = '1.4 V\'H2'
 135                 elif databyte == 0x2c:
 136                     self.type = '1.4 V\'H3'
 137                 elif databyte == 0x30:
 138                     self.type = '1.4 V\'H4'
 139                 elif databyte == 0x40:
 140                     self.type = '1.4 Bcaps'
 141                 elif databyte == 0x41:
 142                     self.type = '1.4 Bstatus'
 143                 elif databyte == 0x43:
 144                     self.type = '1.4 KSV FIFO'
 145                 elif databyte == 0x50:
 146                     self.type = 'HDCP2Version'
 147                 elif databyte == 0x60:
 148                     self.type = 'Write_Message'
 149                     self.state = 'BUFFER DATA'
 150                 elif databyte == 0x70:
 151                     self.type = 'RxStatus'
 152                 elif databyte == 0x80:
 153                     self.type = 'Read_Message'
 154
 155                 # If we are reading, then jump back to IDLE for a start repeat.
 156                 # If we are writing, then just continue onwards.
 157                 if self.state == 'BUFFER DATA':
 158                     self.state = 'BUFFER DATA'
 159                 elif self.type != '':
 160                     self.state = 'IDLE'
 161
 162         elif self.state == 'BUFFER DATA':
 163             if (cmd == 'STOP') or (cmd == 'NACK'):
 164                 self.es_block = es
 165                 self.state = 'IDLE'
 166                 if self.type != '':
 167                     if self.stack:
 168                         if self.type == 'RxStatus':
 169                             rxstatus = (self.stack.pop() << 8) | self.stack.pop()
 170                             reauth_req = (rxstatus & 0x800) != 0
 171                             ready = (rxstatus & 0x400) != 0
 172                             length = rxstatus & 0x3ff
 173                             text = '%s, reauth %s, ready %s, length %s' % (self.type, reauth_req, ready, length)
 174                             self.putb([18, [text]])
 175                         elif self.type == '1.4 Bstatus':
 176                             bstatus = (self.stack.pop() << 8) | self.stack.pop()
 177                             device_count = bstatus & 0x7f
 178                             max_devs_exceeded = (bstatus & 0x80) != 0
 179                             depth = ((bstatus & 0x700) >> 8)
 180                             max_cascase_exceeded = bstatus & 0x800
 181                             hdmi_mode = (bstatus & 0x1000) != 0
 182                             text = '%s, %s devices, depth %s, hdmi mode %s' % (self.type, device_count, depth, hdmi_mode)
 183                             self.putb([18, [text]])
 184                         elif self.type == 'Read_Message':
 185                             msg = self.stack.pop(0)
 186                             self.putb([msg, ['%s, %s' % (self.type, msg_ids.get(msg, 'Invalid'))]])
 187                         elif self.type == 'Write_Message':
 188                             msg = self.stack.pop(0)
 189                             self.putb([msg, ['%s, %s' % (self.type, msg_ids.get(msg, 'Invalid'))]])
 190                         elif self.type == 'HDCP2Version':
 191                             version = self.stack.pop(0)
 192                             if (version & 0x4):
 193                                 self.putb([18, ['HDCP2']])
 194                             else:
 195                                 self.putb([18, ['NOT HDCP2']])
 196                         else:
 197                             self.putb([18, ['%s' % (self.type)]])
 198                     else:
 199                         self.putb([18, ['%s' % (self.type)]])
 200             elif cmd == 'DATA READ':
 201                 # Stack up our data bytes.
 202                 self.stack.append(databyte)
 203             elif cmd == 'DATA WRITE':
 204                 # Stack up our data bytes.
 205                 self.stack.append(databyte)