Add support for install mode.

If a PD report to use SCBK-D in SCS_12, generate a SCBK using
Master Key and set using the KEYSET command.

Signed-off-by: Johan Carlsson <johan.carlsson@teenage.engineering>

Author: Johan Carlsson

osdp/_bus.py

@@ -132,6 +132,9 @@ def process_reply(self, reply: Reply, device: Device):
 			if device.validate_secure_channel_establishment(reply):
 				log.debug("Secure session established.")
 
+		if device.reset_security_after_reply == True:
+			device.reset_security()
+
 		if self._on_reply_received is not None:
 			self._on_reply_received(reply)
 

osdp/_device.py

@@ -2,9 +2,9 @@
 import queue
 import datetime
 
-from ._types import Control
+from ._types import Control, SecurityBlockType
 from ._command import (
-	PollCommand, SecurityInitializationRequestCommand, ServerCryptogramCommand
+	PollCommand, SecurityInitializationRequestCommand, ServerCryptogramCommand, KeySetCommand
 )
 from ._secure_channel import SecureChannel
 
@@ -21,6 +21,7 @@ def __init__(self, address: int, use_crc: bool, use_secure_channel: bool, master
 		self._commands = queue.Queue()
 		self._secure_channel = SecureChannel(master_key)
 		self._last_valid_reply = datetime.datetime.utcfromtimestamp(0)
+		self.reset_security_after_reply = False
 
 	@property
 	def is_security_established(self) -> bool:
@@ -40,6 +41,10 @@ def get_next_command_data(self):
 		if self._use_secure_channel and not self._secure_channel.is_established:
 			return ServerCryptogramCommand(self.address, self._secure_channel.server_cryptogram)
 
+		if self._use_secure_channel and self._secure_channel.is_scbkd:
+			self.reset_security_after_reply = True
+			return KeySetCommand(self.address, self._secure_channel.calculate_scbk())
+
 		if self._commands.empty():
 			return PollCommand(self.address)
 		else:
@@ -54,6 +59,9 @@ def valid_reply_has_been_received(self):
 		self._last_valid_reply = datetime.datetime.now()
 
 	def initialize_secure_channel(self, reply):
+		if reply.security_block_type == SecurityBlockType.SecureConnectionSequenceStep2.value:
+			self._secure_channel.select_scbk(reply.secure_block_data[0])
+
 		reply_data = reply.extract_reply_data
 		self._secure_channel.initialize(reply_data[:8], reply_data[8:16], reply_data[16:32])
 
@@ -69,6 +77,7 @@ def generate_mac(self, message: bytes, is_command: bool):
 		return self._secure_channel.generate_mac(message, is_command)
 
 	def reset_security(self):
+		self.reset_security_after_reply = False
 		self._secure_channel.reset()
 
 	def encrypt_data(self, data: bytes):

osdp/_secure_channel.py

@@ -22,17 +22,25 @@ def __init__(self, master_key: bytes):
 		self.is_initialized = False
 		self.is_established = False
 		self.master_key = master_key
+		self.is_scbkd = False
+		self.cuid = None
+		self.scbk = None
 		self.reset()
 
 	def initialize(self, cuid: bytes, client_random_number: bytes, client_cryptogram: bytes):
+		self.cuid = cuid
+		if self.is_scbkd == True:
+			self.scbk = self.default_secure_channel_key
+		else:
+			self.scbk = self.calculate_scbk()
 		self._enc = self.generate_key(
 			bytes([
 				0x01, 0x82,
 				self.server_random_number[0], self.server_random_number[1], self.server_random_number[2],
 				self.server_random_number[3], self.server_random_number[4], self.server_random_number[5]
 			]),
 			bytes([0x00] * 8),
-			self.default_secure_channel_key
+			self.scbk
 		)
 
 		if client_cryptogram != self.generate_key(self.server_random_number, client_random_number, self._enc):
@@ -45,7 +53,7 @@ def initialize(self, cuid: bytes, client_random_number: bytes, client_cryptogram
 				self.server_random_number[3], self.server_random_number[4], self.server_random_number[5]
 			]),
 			bytes([0x00] * 8),
-			self.default_secure_channel_key
+			self.scbk
 		)
 		self._smac2 = self.generate_key(
 			bytes([
@@ -55,7 +63,7 @@ def initialize(self, cuid: bytes, client_random_number: bytes, client_cryptogram
 				self.server_random_number[4], self.server_random_number[5]
 			]),
 			bytes([0x00] * 8),
-			self.default_secure_channel_key
+			self.scbk
 		)
 		self.server_cryptogram = self.generate_key(
 			client_random_number,
@@ -133,3 +141,19 @@ def reset(self):
 	def generate_key(self, first: bytes, second: bytes, key: bytes) -> bytes:
 		cipher = AES.new(key, AES.MODE_ECB)
 		return cipher.encrypt(first + second)
+
+	def select_scbk(self, byte):
+		if byte == 0x0:
+			self.is_scbkd = True
+		else:
+			self.is_scbkd = False
+
+	def calculate_scbk(self):
+		inv_cuid = bytes([(~b) & 0xFF for b in self.cuid])
+		scbk = self.generate_key(
+			self.cuid,
+			inv_cuid,
+			self.master_key
+		)
+
+		return scbk