[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']
    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']
	51	annotations = \
	52	tuple(('message-0x%02X' % i, 'Message 0x%02X' % i) for i in range(18)) + (
	53	('summary', 'Summary'),
	54	('warnings', 'Warnings'),
	55	)
	56	annotation_rows = (
	57	('messages', 'Messages', tuple(range(18))),
	58	('summary', 'Summary', (18,)),
	59	('warnings', 'Warnings', (19,)),
	60	)
	61
	62	def __init__(self):
	63	self.reset()
	64
65	def reset(self):
66	self.state = 'IDLE'
67	self.stack = []
68	self.msg = -1
69	self.ss = self.es = self.ss_block = self.es_block = 0
70	self.init_seq = []
71	self.valid = 0
72	self.type = ''
73
74	def start(self):
75	self.out_ann = self.register(srd.OUTPUT_ANN)
76
77	def putb(self, data):
78	self.put(self.ss_block, self.es_block, self.out_ann, data)
79
80	def decode(self, ss, es, data):
81	cmd, databyte = data
82
83	# Collect the 'BITS' packet, then return. The next packet is
84	# guaranteed to belong to these bits we just stored.
85	if cmd == 'BITS':
86	self.bits = databyte
87	return
88
89	self.ss, self.es = ss, es
90
91	# State machine.
92	if self.state == 'IDLE':
93	# Wait for an I2C START condition.
94	if cmd == 'START':
95	self.reset()
96	self.ss_block = ss
97	elif cmd != 'START REPEAT':
98	return
99	self.state = 'GET SLAVE ADDR'
100
101	elif self.state == 'GET SLAVE ADDR':
102	if cmd == 'ADDRESS READ':
103	self.state = 'BUFFER DATA'
104	if databyte != 0x3a:
105	self.state = 'IDLE'
106	elif cmd == 'ADDRESS WRITE':
107	self.state = 'WRITE OFFSET'
108	if databyte != 0x3a:
109	self.state = 'IDLE'
110
111	elif self.state == 'WRITE OFFSET':
112	if cmd == 'DATA WRITE':
113	if databyte == 0x00:
114	self.type = '1.4 Bksv - Receiver KSV'
115	elif databyte == 0x08:
116	self.type = '1.4 Ri\' - Link Verification'
117	elif databyte == 0x0a:
118	self.type = '1.4 Pj\' - Enhanced Link Verification'
119	elif databyte == 0x10:
120	self.type = '1.4 Aksv - Transmitter KSV'
121	self.state = 'BUFFER DATA'
122	elif databyte == 0x15:
123	self.type = '1.4 Ainfo- Transmitter KSV'
124	self.state = 'BUFFER DATA'
125	elif databyte == 0x18:
126	self.type = '1.4 An- Session random number'
127	self.state = 'BUFFER DATA'
128	elif databyte == 0x20:
129	self.type = '1.4 V\'H0'
130	elif databyte == 0x24:
131	self.type = '1.4 V\'H1'
132	elif databyte == 0x28:
133	self.type = '1.4 V\'H2'
134	elif databyte == 0x2c:
135	self.type = '1.4 V\'H3'
136	elif databyte == 0x30:
137	self.type = '1.4 V\'H4'
138	elif databyte == 0x40:
139	self.type = '1.4 Bcaps'
140	elif databyte == 0x41:
141	self.type = '1.4 Bstatus'
142	elif databyte == 0x43:
143	self.type = '1.4 KSV FIFO'
144	elif databyte == 0x50:
145	self.type = 'HDCP2Version'
146	elif databyte == 0x60:
147	self.type = 'Write_Message'
148	self.state = 'BUFFER DATA'
149	elif databyte == 0x70:
150	self.type = 'RxStatus'
151	elif databyte == 0x80:
152	self.type = 'Read_Message'
153
154	# If we are reading, then jump back to IDLE for a start repeat.
155	# If we are writing, then just continue onwards.
156	if self.state == 'BUFFER DATA':
157	self.state = 'BUFFER DATA'
158	elif self.type != '':
159	self.state = 'IDLE'
160
161	elif self.state == 'BUFFER DATA':
162	if (cmd == 'STOP') or (cmd == 'NACK'):
163	self.es_block = es
164	self.state = 'IDLE'
165	if self.type != '':
166	if self.stack:
167	if self.type == 'RxStatus':
168	rxstatus = (self.stack.pop() << 8) \| self.stack.pop()
169	reauth_req = (rxstatus & 0x800) != 0
170	ready = (rxstatus & 0x400) != 0
171	length = rxstatus & 0x3ff
172	text = '%s, reauth %s, ready %s, length %s' % (self.type, reauth_req, ready, length)
173	self.putb([18, [text]])
174	elif self.type == '1.4 Bstatus':
175	bstatus = (self.stack.pop() << 8) \| self.stack.pop()
176	device_count = bstatus & 0x7f
177	max_devs_exceeded = (bstatus & 0x80) != 0
178	depth = ((bstatus & 0x700) >> 8)
179	max_cascase_exceeded = bstatus & 0x800
180	hdmi_mode = (bstatus & 0x1000) != 0
181	text = '%s, %s devices, depth %s, hdmi mode %s' % (self.type, device_count, depth, hdmi_mode)
182	self.putb([18, [text]])
183	elif self.type == 'Read_Message':
184	msg = self.stack.pop(0)
185	self.putb([msg, ['%s, %s' % (self.type, msg_ids.get(msg, 'Invalid'))]])
186	elif self.type == 'Write_Message':
187	msg = self.stack.pop(0)
188	self.putb([msg, ['%s, %s' % (self.type, msg_ids.get(msg, 'Invalid'))]])
189	elif self.type == 'HDCP2Version':
190	version = self.stack.pop(0)
191	if (version & 0x4):
192	self.putb([18, ['HDCP2']])
193	else:
194	self.putb([18, ['NOT HDCP2']])
195	else:
196	self.putb([18, ['%s' % (self.type)]])
197	else:
198	self.putb([18, ['%s' % (self.type)]])
199	elif cmd == 'DATA READ':
200	# Stack up our data bytes.
201	self.stack.append(databyte)
202	elif cmd == 'DATA WRITE':
203	# Stack up our data bytes.
204	self.stack.append(databyte)