| """ |
| Simple encryption utilities for Pipedream credential profiles. |
| """ |
|
|
| import os |
| import base64 |
| from cryptography.fernet import Fernet |
| from utils.logger import logger |
|
|
|
|
| def get_encryption_key() -> bytes: |
| """Get or create encryption key for credentials.""" |
| key_env = os.getenv("MCP_CREDENTIAL_ENCRYPTION_KEY") |
| |
| if key_env: |
| try: |
| if isinstance(key_env, str): |
| return key_env.encode('utf-8') |
| else: |
| return key_env |
| except Exception as e: |
| logger.error(f"Invalid encryption key: {e}") |
| |
| |
| logger.warning("No encryption key found, generating new key for this session") |
| key = Fernet.generate_key() |
| logger.info(f"Generated new encryption key. Set this in your environment:") |
| logger.info(f"MCP_CREDENTIAL_ENCRYPTION_KEY={key.decode()}") |
| return key |
|
|
|
|
| def encrypt_data(data: str) -> str: |
| """ |
| Encrypt a string and return base64 encoded encrypted data. |
| |
| Args: |
| data: String data to encrypt |
| |
| Returns: |
| Base64 encoded encrypted string |
| """ |
| encryption_key = get_encryption_key() |
| cipher = Fernet(encryption_key) |
| |
| |
| data_bytes = data.encode('utf-8') |
| |
| |
| encrypted_bytes = cipher.encrypt(data_bytes) |
| |
| |
| return base64.b64encode(encrypted_bytes).decode('utf-8') |
|
|
|
|
| def decrypt_data(encrypted_data: str) -> str: |
| """ |
| Decrypt base64 encoded encrypted data and return the original string. |
| |
| Args: |
| encrypted_data: Base64 encoded encrypted string |
| |
| Returns: |
| Decrypted string |
| """ |
| encryption_key = get_encryption_key() |
| cipher = Fernet(encryption_key) |
| |
| |
| encrypted_bytes = base64.b64decode(encrypted_data.encode('utf-8')) |
| |
| |
| decrypted_bytes = cipher.decrypt(encrypted_bytes) |
| |
| |
| return decrypted_bytes.decode('utf-8') |
