Implemented PGP for credentials

requirements.txt

@@ -3,4 +3,5 @@ dagster-webserver>=1.11.16
 dagster-postgres>=0.27.16
 pytest>=8.4.2
 pytest-cov>=7.0.0
-requests>=2.32.0
+requests>=2.32.0
+pgpy>=0.6.0

src/simpl_open_credential_delivery/ops/send_credentials_email.py

@@ -9,6 +9,12 @@ from simpl_open_credential_delivery.services import EmailDeliveryService
 
 
 @op(
+    config_schema={
+        "pgp_public_key_armored": Field(
+            str,
+            description="Public PGP key",
+        )
+    },
     ins={"credentials": In(dict)},
     out=Out(dict, description="Email delivery result"),
     retry_policy=DEFAULT_RETRY_POLICY,
@@ -24,10 +30,12 @@ def send_credentials_email(
         credentials["consumer_email"],
     )
 
+    config = context.op_config
     service = EmailDeliveryService.from_environment()
     delivery_result = service.send_credentials_email(
         recipient=credentials["consumer_email"],
         credentials=credentials,
+        pgp_public_key_armored=config["pgp_public_key_armored"]
     )
 
     context.log.info(

src/simpl_open_credential_delivery/services/email_service.py

@@ -8,6 +8,7 @@ from dataclasses import dataclass
 import os
 import smtplib
 from email.message import EmailMessage
+import pgpy
 
 
 class EmailDeliveryError(RuntimeError):
@@ -78,10 +79,28 @@ class EmailDeliveryService:
             timeout_seconds=int(os.getenv("SMTP_TIMEOUT_SECONDS", "30")),
         )
 
-    def send_credentials_email(self, recipient: str, credentials: dict[str, str]) -> EmailDeliveryResult:
+    def send_credentials_email(self, recipient: str, credentials: dict[str, str], pgp_public_key_armored: str) -> EmailDeliveryResult:
         """Render the email from templates and send it using SMTP."""
-        subject = self.subject_template.format(**credentials)
-        body = self.body_template.format(**credentials)
+
+        if not pgp_public_key_armored or not pgp_public_key_armored.strip():
+            raise EmailDeliveryError("PGP public key is required and cannot be empty.")
+
+        local_credentials = credentials.copy()
+        if "access_token" in local_credentials:
+            try:
+                pub_key, _ = pgpy.PGPKey.from_blob(pgp_public_key_armored)
+                
+                pgp_message = pgpy.PGPMessage.new(local_credentials["access_token"])
+                encrypted_message = pub_key.encrypt(pgp_message)
+                
+                local_credentials["access_token"] = f"\n{str(encrypted_message)}"
+            except Exception as exc:
+                raise EmailDeliveryError(f"Failed to encrypt access_token with PGP: {exc}") from exc
+        else:
+            raise EmailDeliveryError("Missing 'access_token' in the provided credentials.")
+
+        subject = self.subject_template.format(**local_credentials)
+        body = self.body_template.format(**local_credentials)
 
         message = EmailMessage()
         message["To"] = recipient