Adds public_key to universal core capability

Author: marshallmcdonnell

src/intersect_sdk/capability/universal_capability/universal_capability.py

@@ -6,13 +6,16 @@
 """
 
 import datetime
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives import serialization
 import os
 import time
 from typing import final
 
 import psutil
 
-from ...service_definitions import intersect_status
+from ...service_definitions import intersect_message, intersect_status
+from ..._internal.encryption.models import IntersectEncryptionPublicKey  
 from ..base import IntersectBaseCapabilityImplementation
 from .status import IntersectCoreStatus
 
@@ -36,6 +39,25 @@ def __init__(self) -> None:  # noqa: D107
         self.process = psutil.Process(os.getpid())
         """psutil.Process caches most functions it calls after it calls the function once, so just save the object itself"""
 
+        # Generate a key pair for encryption
+        self._private_key: rsa.RSAPrivateKey = rsa.generate_private_key(
+            public_exponent=65537, key_size=2048
+        )
+        self._public_key = self._private_key.public_key()
+
+        # Get the PEM encoded public key
+        self._public_key_pem = self._public_key.public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        ).decode()
+
+        self._private_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+
+
     @intersect_status
     def system_capability(self) -> IntersectCoreStatus:
         """The status of this Capability reflects core system information which is okay to broadcast across the INTERSECT-SDK system.
@@ -58,3 +80,9 @@ def system_capability(self) -> IntersectCoreStatus:
             disk_total=disk_info.total,
             disk_usage_percentage=disk_info.percent,
         )
+    
+    @intersect_message()
+    def get_public_key(self) -> Dict[str, str]:
+        """Returns the public key for clients / services to use for encryption"""
+        return IntersectEncryptionPublicKey(public_key=self._public_key_pem)
+