[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',
}

write_items = {
    0x00: '1.4 Bksv - Receiver KSV',
    0x08: '1.4 Ri\' - Link Verification',
    0x0a: '1.4 Pj\' - Enhanced Link Verification',
    0x10: '1.4 Aksv - Transmitter KSV',
    0x15: '1.4 Ainfo - Transmitter KSV',
    0x18: '1.4 An - Session random number',
    0x20: '1.4 V\'H0',
    0x24: '1.4 V\'H1',
    0x28: '1.4 V\'H2',
    0x2c: '1.4 V\'H3',
    0x30: '1.4 V\'H4',
    0x40: '1.4 Bcaps',
    0x41: '1.4 Bstatus',
    0x43: '1.4 KSV FIFO',
    0x50: 'HDCP2Version',
    0x60: 'Write_Message',
    0x70: 'RxStatus',
    0x80: 'Read_Message',
}

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