[libsigrokdecode.git] / decoders / hdcp / pd.py

##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2018 Dave Craig<dcraig@brightsign.biz>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see <http://www.gnu.org/licenses/>.
##

import sigrokdecode as srd

msg_ids = {
    2: 'AKE_Init',
    3: 'AKE_Send_Cert',
    4: 'AKE_No_stored_km',
    5: 'AKE_Stored_km',

    7: 'AKE_Send_H_prime',
    8: 'AKE_Send_Pairing_Info',

    9: 'LC_Init',
    10: 'LC_Send_L_prime',

    11: 'SKE_Send_Eks',
    12: 'RepeaterAuth_Send_ReceiverID_List',

    15: 'RepeaterAuth_Send_Ack',
    16: 'RepeaterAuth_Stream_Manage',
    17: 'RepeaterAuth_Stream_Ready',
}

class Decoder(srd.Decoder):
    api_version = 3
    id = 'hdcp'
    name = 'HDCP'
    longname = 'HDCP over HDMI'
    desc = 'HDCP protocol over HDMI.'
    license = 'gplv2+'
    inputs = ['i2c']
    outputs = ['hdcp']
    tags = ['PC', 'Security/crypto']
    annotations = \
        tuple(('message-0x%02X' % i, 'Message 0x%02X' % i) for i in range(18)) + (
        ('summary', 'Summary'),
        ('warnings', 'Warnings'),
    )
    annotation_rows = (
        ('messages', 'Messages', tuple(range(18))),
        ('summary', 'Summary', (18,)),
        ('warnings', 'Warnings', (19,)),
    )

    def __init__(self):
        self.reset()

    def reset(self):
        self.state = 'IDLE'
        self.stack = []
        self.msg = -1
        self.ss = self.es = self.ss_block = self.es_block = 0
        self.init_seq = []
        self.valid = 0
        self.type = ''

    def start(self):
        self.out_ann = self.register(srd.OUTPUT_ANN)

    def putb(self, data):
        self.put(self.ss_block, self.es_block, self.out_ann, data)

    def decode(self, ss, es, data):
        cmd, databyte = data

        # Collect the 'BITS' packet, then return. The next packet is
        # guaranteed to belong to these bits we just stored.
        if cmd == 'BITS':
            self.bits = databyte
            return

        self.ss, self.es = ss, es

        # State machine.
        if self.state == 'IDLE':
            # Wait for an I2C START condition.
            if cmd == 'START':
                self.reset()
                self.ss_block = ss
            elif cmd != 'START REPEAT':
                return
            self.state = 'GET SLAVE ADDR'

        elif self.state == 'GET SLAVE ADDR':
            if cmd == 'ADDRESS READ':
                self.state = 'BUFFER DATA'
                if databyte != 0x3a:
                    self.state = 'IDLE'
            elif cmd == 'ADDRESS WRITE':
                self.state = 'WRITE OFFSET'
                if databyte != 0x3a:
                    self.state = 'IDLE'

        elif self.state == 'WRITE OFFSET':
            if cmd == 'DATA WRITE':
                if databyte == 0x00:
                    self.type = '1.4 Bksv - Receiver KSV'
                elif databyte == 0x08:
                    self.type = '1.4 Ri\' - Link Verification'
                elif databyte == 0x0a:
                    self.type = '1.4 Pj\' - Enhanced Link Verification'
                elif databyte == 0x10:
                    self.type = '1.4 Aksv - Transmitter KSV'
                    self.state = 'BUFFER DATA'
                elif databyte == 0x15:
                    self.type = '1.4 Ainfo- Transmitter KSV'
                    self.state = 'BUFFER DATA'
                elif databyte == 0x18:
                    self.type = '1.4 An- Session random number'
                    self.state = 'BUFFER DATA'
                elif databyte == 0x20:
                    self.type = '1.4 V\'H0'
                elif databyte == 0x24:
                    self.type = '1.4 V\'H1'
                elif databyte == 0x28:
                    self.type = '1.4 V\'H2'
                elif databyte == 0x2c:
                    self.type = '1.4 V\'H3'
                elif databyte == 0x30:
                    self.type = '1.4 V\'H4'
                elif databyte == 0x40:
                    self.type = '1.4 Bcaps'
                elif databyte == 0x41:
                    self.type = '1.4 Bstatus'
                elif databyte == 0x43:
                    self.type = '1.4 KSV FIFO'
                elif databyte == 0x50:
                    self.type = 'HDCP2Version'
                elif databyte == 0x60:
                    self.type = 'Write_Message'
                    self.state = 'BUFFER DATA'
                elif databyte == 0x70:
                    self.type = 'RxStatus'
                elif databyte == 0x80:
                    self.type = 'Read_Message'

                # If we are reading, then jump back to IDLE for a start repeat.
                # If we are writing, then just continue onwards.
                if self.state == 'BUFFER DATA':
                    self.state = 'BUFFER DATA'
                elif self.type != '':
                    self.state = 'IDLE'

        elif self.state == 'BUFFER DATA':
            if (cmd == 'STOP') or (cmd == 'NACK'):
                self.es_block = es
                self.state = 'IDLE'
                if self.type != '':
                    if self.stack:
                        if self.type == 'RxStatus':
                            rxstatus = (self.stack.pop() << 8) | self.stack.pop()
                            reauth_req = (rxstatus & 0x800) != 0
                            ready = (rxstatus & 0x400) != 0
                            length = rxstatus & 0x3ff
                            text = '%s, reauth %s, ready %s, length %s' % (self.type, reauth_req, ready, length)
                            self.putb([18, [text]])
                        elif self.type == '1.4 Bstatus':
                            bstatus = (self.stack.pop() << 8) | self.stack.pop()
                            device_count = bstatus & 0x7f
                            max_devs_exceeded = (bstatus & 0x80) != 0
                            depth = ((bstatus & 0x700) >> 8)
                            max_cascase_exceeded = bstatus & 0x800
                            hdmi_mode = (bstatus & 0x1000) != 0
                            text = '%s, %s devices, depth %s, hdmi mode %s' % (self.type, device_count, depth, hdmi_mode)
                            self.putb([18, [text]])
                        elif self.type == 'Read_Message':
                            msg = self.stack.pop(0)
                            self.putb([msg, ['%s, %s' % (self.type, msg_ids.get(msg, 'Invalid'))]])
                        elif self.type == 'Write_Message':
                            msg = self.stack.pop(0)
                            self.putb([msg, ['%s, %s' % (self.type, msg_ids.get(msg, 'Invalid'))]])
                        elif self.type == 'HDCP2Version':
                            version = self.stack.pop(0)
                            if (version & 0x4):
                                self.putb([18, ['HDCP2']])
                            else:
                                self.putb([18, ['NOT HDCP2']])
                        else:
                            self.putb([18, ['%s' % (self.type)]])
                    else:
                        self.putb([18, ['%s' % (self.type)]])
            elif cmd == 'DATA READ':
                # Stack up our data bytes.
                self.stack.append(databyte)
            elif cmd == 'DATA WRITE':
                # Stack up our data bytes.
                self.stack.append(databyte)
Commit	Line	Data
e27dd4b6 DC	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']
3e5bd72e	51	tags = ['PC', 'Security/crypto']
e27dd4b6 DC	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)