Draft: Compare changes for Humble integration

sros2/sros2/keystore/_permission.py

@@ -14,8 +14,10 @@
 
 import os
 import pathlib
+import shutil
 
 from lxml import etree
+from mypy.config_parser import toml_config_types
 
 from rclpy.utilities import get_rmw_implementation_identifier
 
@@ -34,6 +36,9 @@ _RMW_WITH_ROS_GRAPH_INFO_TOPIC = (
     'rmw_fastrtps_dynamic_cpp'
 )
 
+_TPM_PCR_SELECTIONS_DIR = "pcr"
+
+
 
 def create_permission(
         keystore_path: pathlib.Path,
@@ -42,6 +47,95 @@ def create_permission(
     policy_element = _policy.get_policy(identity, policy_file_path)
     create_permissions_from_policy_element(keystore_path, identity, policy_element)
 
+def extend_permission_with_pcr(
+        keystore_path: pathlib.Path,
+        identity: str,
+        system_identifier: str,
+        subject: str
+):
+    relative_path = os.path.normpath(identity.lstrip('/'))
+    key_dir = _keystore.get_keystore_enclaves_dir(keystore_path).joinpath(relative_path)
+    key_tpm_dir = _keystore.get_keystore_enclaves_dir(keystore_path).joinpath(relative_path).joinpath("tpm")
+    permissions_path = key_dir.joinpath('permissions.xml')
+
+    tpm_dir = _keystore.get_keystore_tpm_dir(keystore_path)
+    tpm_system_dir = tpm_dir.joinpath(system_identifier)
+
+    # Create XML elements to add to the permission.xml
+    pcr_selection_children = []
+    config_parsed = False
+
+    total_pcr_selection = []
+    total_bank_selection = []
+    final_pcr_selection_path = key_tpm_dir.joinpath("pcr_selection.txt")
+
+    for _, dirs, _ in os.walk(tpm_system_dir):
+        for config_dir in dirs:
+            config_path = tpm_system_dir.joinpath(config_dir)
+            tpm_config_pcr_values_path = config_path.joinpath("pcr_values.yaml")
+            tpm_config_pcr_selection_path = config_path.joinpath("pcr_selection.txt")
+
+            if (not tpm_config_pcr_values_path.is_file()) or (not tpm_config_pcr_selection_path.is_file()):
+                continue
+
+            with open(tpm_config_pcr_values_path, "r") as f:
+                pcr_values_data = f.read().splitlines()
+                pcr_selection_children.append(*_create_permission_xml_for_value(pcr_values_data))
+
+
+            with open(tpm_config_pcr_selection_path, "r") as f:
+                new_pcr_selection = f.read()
+                new_raw_banks, new_raw_selection = new_pcr_selection.split(":", 1)
+                new_banks = new_raw_banks.split(",")
+                new_selection = new_raw_selection.split(",")
+                total_pcr_selection.extend([s for s in new_selection if s not in total_pcr_selection])
+                total_bank_selection.extend([s for s in new_banks if s not in total_bank_selection])
+
+
+            config_parsed = True
+
+        break
+
+    with open(final_pcr_selection_path, "w") as f:
+        f.write(",".join(total_bank_selection) + ":" + ",".join(sorted(total_pcr_selection, key=int)))
+
+    if not config_parsed:
+        raise sros2.errors.ConfigParsingError(system_identifier)
+
+
+
+    tree = etree.parse(permissions_path)
+    root = tree.getroot()
+    for grant in root.iter("grant"):
+        for pcr_selections in grant.iter("pcr_selections"):
+            grant.remove(pcr_selections)
+
+        new_pcr_selections = etree.SubElement(grant, "pcr_selections")
+
+        subject_name_element = etree.SubElement(new_pcr_selections, "subject_name")
+        subject_name_element.text = subject
+
+        # Add new elements with the pcr values
+        for pcr_selection_child in pcr_selection_children:
+            new_pcr_selections.append(pcr_selection_child)
+
+    tree = etree.ElementTree(root)
+    tree.write(permissions_path, pretty_print=True, xml_declaration=True, encoding="UTF-8")
+
+    # Sign Permissions.xml
+    signed_permissions_path = os.path.join(key_dir, 'permissions.p7s')
+    keystore_permissions_ca_cert_path = os.path.join(
+        _keystore.get_keystore_public_dir(keystore_path), 'permissions_ca.cert.pem')
+    keystore_permissions_ca_key_path = os.path.join(
+        _keystore.get_keystore_private_dir(keystore_path), 'permissions_ca.key.pem')
+
+    _utilities.create_smime_signed_file(
+        keystore_permissions_ca_cert_path,
+        keystore_permissions_ca_key_path,
+        permissions_path,
+        signed_permissions_path
+    )
+
 
 def create_permissions_from_policy_element(
         keystore_path: pathlib.Path,
@@ -93,3 +187,33 @@ def create_permission_file(path: pathlib.Path, domain_id, policy_element) -> Non
 
     with open(path, 'wb') as f:
         f.write(etree.tostring(permissions_xml, pretty_print=True))
+
+def _create_permission_xml_for_value(pcr_value: list) -> etree.Element:
+    active_bank = None
+    current_data = ""
+    pcr_selection_children = []
+
+    for line in pcr_value:
+        if line.strip().startswith("s"): # All TPM banks hash algorithms start with s
+            bank = line.replace(":", "").strip()
+            if active_bank and current_data:
+                xml_element = etree.Element("pcr_selection", bank=active_bank)
+                xml_element.text = "\n" + current_data
+                pcr_selection_children.append(xml_element)
+
+
+            active_bank = bank
+            current_data = ""
+        elif line.startswith(" "):
+            data = line.strip()
+            if active_bank:
+                current_data += data + "\n"
+        else:
+            pass
+
+    if current_data:
+        xml_element = etree.Element("pcr_selection", bank=active_bank)
+        xml_element.text = "\n" + current_data
+        pcr_selection_children.append(xml_element)
+
+    return pcr_selection_children
\ No newline at end of file