first version
This commit is contained in:
@@ -0,0 +1,13 @@
|
||||
"""Initialization file for the python_encrypt_code package."""
|
||||
|
||||
from .data_zipper import DataZipper
|
||||
from .file_encrypter import FileEncrypter
|
||||
from .password_provider import PasswordProvider
|
||||
from .module_importer import ModuleImporter
|
||||
|
||||
__all__ = [
|
||||
"DataZipper",
|
||||
"FileEncrypter",
|
||||
"PasswordProvider",
|
||||
"ModuleImporter",
|
||||
]
|
||||
@@ -0,0 +1,160 @@
|
||||
"""Command-line interface for python-encrypt-code package."""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from python_encrypt_code import ModuleImporter
|
||||
from python_encrypt_code.utils import (
|
||||
generate_password,
|
||||
encrypt_folder,
|
||||
decrypt_to_disk,
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Main CLI function."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Encrypt and run folders securely", prog="python-encrypt-code"
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
||||
|
||||
# Generate password command
|
||||
_ = subparsers.add_parser(
|
||||
name="generate-password",
|
||||
help="Generate a secure password",
|
||||
)
|
||||
|
||||
# Encrypt command
|
||||
encrypt_parser = subparsers.add_parser(
|
||||
name="encrypt",
|
||||
help="Encrypt a folder",
|
||||
)
|
||||
encrypt_parser.add_argument(
|
||||
"folder",
|
||||
type=Path,
|
||||
help="Path to the folder to encrypt",
|
||||
)
|
||||
encrypt_parser.add_argument(
|
||||
"-o",
|
||||
"--output",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="Path where the encrypted file will be saved",
|
||||
)
|
||||
encrypt_parser.add_argument(
|
||||
"-p",
|
||||
"--password",
|
||||
type=str,
|
||||
help="Password to use for encryption",
|
||||
)
|
||||
encrypt_parser.add_argument(
|
||||
"-aad",
|
||||
"--additional-authenticated-data",
|
||||
type=str,
|
||||
default=None,
|
||||
help="JSON-string with optional metadata for encrypted file",
|
||||
)
|
||||
|
||||
# Decrypt command
|
||||
decrypt_parser = subparsers.add_parser(
|
||||
"decrypt",
|
||||
help="Decrypt a file to disk",
|
||||
)
|
||||
decrypt_parser.add_argument(
|
||||
"file",
|
||||
type=Path,
|
||||
help="Path to the encrypted file",
|
||||
)
|
||||
decrypt_parser.add_argument(
|
||||
"-o",
|
||||
"--output",
|
||||
type=Path,
|
||||
required=True,
|
||||
help="Path where the decrypted folder will be saved",
|
||||
)
|
||||
|
||||
# Run insecure command
|
||||
decrypt_run_parser = subparsers.add_parser(
|
||||
"run-insecure",
|
||||
help="Decrypt to temporary folder and run a script from an encrypted file",
|
||||
)
|
||||
decrypt_run_parser.add_argument(
|
||||
"file",
|
||||
type=Path,
|
||||
help="Path to the encrypted file",
|
||||
)
|
||||
decrypt_run_parser.add_argument(
|
||||
"-s",
|
||||
"--script",
|
||||
type=str,
|
||||
default="main.py",
|
||||
help="Name of the script to run (default: main.py)",
|
||||
)
|
||||
|
||||
# Run command
|
||||
run_parser = subparsers.add_parser(
|
||||
"run", help="Decrypt to memory and run a script from an encrypted file"
|
||||
)
|
||||
run_parser.add_argument("file", type=Path, help="Path to the encrypted file")
|
||||
run_parser.add_argument(
|
||||
"-s",
|
||||
"--script",
|
||||
type=str,
|
||||
default="main.py",
|
||||
help="Name of the script to run (default: main.py)",
|
||||
)
|
||||
|
||||
# Parse arguments
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
return
|
||||
|
||||
try:
|
||||
if args.command == "generate-password":
|
||||
generate_password()
|
||||
|
||||
elif args.command == "encrypt":
|
||||
encrypt_folder(
|
||||
folder_path=args.folder,
|
||||
output_path=args.output,
|
||||
password=args.password,
|
||||
aad=args.additional_authenticated_data,
|
||||
)
|
||||
|
||||
elif args.command == "decrypt":
|
||||
decrypt_to_disk(
|
||||
encrypted_file_path=args.file,
|
||||
output_path=args.output,
|
||||
)
|
||||
|
||||
elif args.command == "run-insecure":
|
||||
# Create temporary directory for decrypted files
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
temp_path = Path(temp_dir)
|
||||
|
||||
decrypt_to_disk(
|
||||
encrypted_file_path=args.file,
|
||||
output_path=temp_path,
|
||||
)
|
||||
|
||||
# Import and run the specified module
|
||||
ModuleImporter.import_module_from_disk(
|
||||
module_path=temp_path,
|
||||
main_module=args.script,
|
||||
)
|
||||
|
||||
elif args.command == "run":
|
||||
raise NotImplementedError("The 'run' command is not implemented yet.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Data zipper module."""
|
||||
|
||||
from .data_zipper import DataZipper
|
||||
|
||||
__all__ = ["DataZipper"]
|
||||
@@ -0,0 +1,98 @@
|
||||
"""Definition of DataZipper class."""
|
||||
|
||||
from io import BytesIO
|
||||
import os
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from typing import Dict
|
||||
|
||||
from .data_zipper_interface import DataZipperInterface
|
||||
|
||||
|
||||
class DataZipper(DataZipperInterface):
|
||||
"""Class to handle zipping of data folders."""
|
||||
|
||||
@staticmethod
|
||||
def zip_data(
|
||||
input_path: Path,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
"""Zips the contents of a folder into a zip file.
|
||||
|
||||
Args:
|
||||
folder_path (Path): The path to the folder to be zipped.
|
||||
output_path (Path): The path where the zip file will be saved.
|
||||
"""
|
||||
|
||||
# Check inputs
|
||||
if not isinstance(input_path, Path):
|
||||
raise TypeError("input_path must be a Path object")
|
||||
if not input_path.exists():
|
||||
raise FileNotFoundError(f"input_path {input_path} does not exist")
|
||||
if not input_path.is_dir():
|
||||
raise ValueError("input_path must be a directory")
|
||||
if not isinstance(output_path, Path):
|
||||
raise TypeError("output_path must be a Path object")
|
||||
|
||||
# Zip the folder
|
||||
with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as zipf:
|
||||
for root, _, files in os.walk(input_path):
|
||||
for file in files:
|
||||
file_path = Path(root) / file
|
||||
zipf.write(file_path, arcname=file_path.relative_to(input_path))
|
||||
|
||||
@staticmethod
|
||||
def unzip_data(
|
||||
input_data: BytesIO,
|
||||
) -> Dict[str, BytesIO]:
|
||||
"""Unzips the data of a zip file into memory.
|
||||
|
||||
Args:
|
||||
input_data (BytesIO): The data of the zip file to be unzipped.
|
||||
|
||||
Returns:
|
||||
Dict[str, BytesIO]: Dictionary mapping file names to BytesIO objects.
|
||||
"""
|
||||
|
||||
# Check inputs
|
||||
if not isinstance(input_data, BytesIO):
|
||||
raise TypeError("input_data must be a BytesIO object")
|
||||
|
||||
if input_data.getbuffer().nbytes == 0:
|
||||
raise ValueError("input_data must not be empty")
|
||||
|
||||
# Unzip the file into memory
|
||||
extracted_files = {}
|
||||
with zipfile.ZipFile(input_data, "r") as zipf:
|
||||
for file_info in zipf.filelist:
|
||||
if not file_info.is_dir(): # Skip directories
|
||||
file_content = zipf.read(file_info.filename)
|
||||
extracted_files[file_info.filename] = BytesIO(file_content)
|
||||
|
||||
return extracted_files
|
||||
|
||||
@staticmethod
|
||||
def unzip_data_to_disk(
|
||||
input_data: BytesIO,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
"""Unzips the data of a zip file to disk.
|
||||
|
||||
Args:
|
||||
input_data (BytesIO): The data of the zip file to be unzipped.
|
||||
output_path (Path): The path where the unzipped files will be saved.
|
||||
"""
|
||||
|
||||
# Check inputs
|
||||
if not isinstance(input_data, BytesIO):
|
||||
raise TypeError("input_data must be a BytesIO object")
|
||||
|
||||
if not isinstance(output_path, Path):
|
||||
raise TypeError("output_path must be a Path object")
|
||||
|
||||
if input_data.getbuffer().nbytes == 0:
|
||||
raise ValueError("input_data must not be empty")
|
||||
|
||||
# Unzip the file to disk
|
||||
with zipfile.ZipFile(input_data, "r") as zipf:
|
||||
zipf.extractall(output_path)
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Definition of DataZipperInterface."""
|
||||
|
||||
from io import BytesIO
|
||||
from typing import Dict
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class DataZipperInterface(ABC):
|
||||
"""Interface for data zipper implementations."""
|
||||
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def zip_data(
|
||||
input_path: Path,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
"""Zips data from the input path and saves it to the output path.
|
||||
|
||||
Args:
|
||||
input_path (Path): The path to the data to be zipped.
|
||||
output_path (Path): The path where the zipped data will be saved.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def unzip_data(
|
||||
input_data: BytesIO,
|
||||
) -> Dict[str, BytesIO]:
|
||||
"""Unzips data from the input path and saves it in memory.
|
||||
|
||||
Args:
|
||||
input_data (BytesIO): The data of the zipped file.
|
||||
Returns:
|
||||
Dict[str, BytesIO]: A dictionary mapping file names to BytesIO objects.
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def unzip_data_to_disk(
|
||||
input_data: BytesIO,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
"""Unzips data from the input path and saves it to the output path on disk.
|
||||
|
||||
Args:
|
||||
input_data (BytesIO): The data of the zipped file.
|
||||
output_path (Path): The path where the unzipped data will be saved.
|
||||
"""
|
||||
@@ -0,0 +1,5 @@
|
||||
"""File encrypter module."""
|
||||
|
||||
from .file_encrypter import FileEncrypter
|
||||
|
||||
__all__ = ["FileEncrypter"]
|
||||
@@ -0,0 +1,174 @@
|
||||
"""Definiition of FileEncrypter class."""
|
||||
|
||||
from io import BytesIO
|
||||
import os
|
||||
import base64
|
||||
import json
|
||||
from pathlib import Path
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
|
||||
from .file_encrypter_interface import FileEncrypterInterface
|
||||
|
||||
|
||||
class FileEncrypter(FileEncrypterInterface):
|
||||
"""Class to handle file encryption and decryption."""
|
||||
|
||||
_encoding: str = "utf-8"
|
||||
_salt_length: int = 16 # Length of the salt in bytes
|
||||
_nonce_length: int = 12 # Length of the nonce in bytes
|
||||
|
||||
@classmethod
|
||||
def encrypt_file(
|
||||
cls,
|
||||
input_file_path: Path,
|
||||
output_file_path: Path,
|
||||
password: str,
|
||||
additional_data: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
"""Encrypt a file using the provided password.
|
||||
|
||||
Args:
|
||||
input_file_path (str): The path to the input file to be encrypted.
|
||||
output_file_path (str): The path where the encrypted file will be saved.
|
||||
password (str): The password used for encryption.
|
||||
additional_data (dict[str, str] | None): Optional additional authenticated data (AAD).
|
||||
"""
|
||||
|
||||
# Check inputs
|
||||
if not isinstance(input_file_path, Path):
|
||||
raise TypeError("input_file_path must be a Path object")
|
||||
if not input_file_path.exists():
|
||||
raise FileNotFoundError(f"input_file_path {input_file_path} does not exist")
|
||||
if not input_file_path.is_file():
|
||||
raise ValueError("input_file_path must be a file")
|
||||
if not isinstance(output_file_path, Path):
|
||||
raise TypeError("output_file_path must be a Path object")
|
||||
if not isinstance(password, str):
|
||||
raise TypeError("password must be a string")
|
||||
if len(password) == 0:
|
||||
raise ValueError("password cannot be empty")
|
||||
password_bytes = base64.urlsafe_b64decode(password.encode(cls._encoding))
|
||||
if len(password_bytes) != 32:
|
||||
raise ValueError("Password must decode to 32 bytes for AES-256")
|
||||
if additional_data is not None and not isinstance(additional_data, dict):
|
||||
raise TypeError("additional_data must be a dict if provided")
|
||||
|
||||
# Prepare additional authenticated data (AAD) if provided
|
||||
additional_data_bytes = None
|
||||
if additional_data is not None:
|
||||
additional_data_bytes = json.dumps(additional_data).encode(cls._encoding)
|
||||
|
||||
# Generate a random Number Used Once (nonce)
|
||||
nonce = os.urandom(cls._nonce_length)
|
||||
|
||||
# Initialize AESGCM with derived key
|
||||
aesgcm = AESGCM(password_bytes)
|
||||
|
||||
# Read the input file
|
||||
with open(input_file_path, "rb") as file:
|
||||
original_data = file.read()
|
||||
|
||||
# Encrypt the data
|
||||
encrypted_data = aesgcm.encrypt(nonce, original_data, additional_data_bytes)
|
||||
|
||||
# Write the salt, nonce, and encrypted data to the output file
|
||||
with open(output_file_path, "wb") as file:
|
||||
file.write(nonce + encrypted_data)
|
||||
|
||||
@classmethod
|
||||
def decrypt_file(
|
||||
cls,
|
||||
input_file_path: Path,
|
||||
password: str,
|
||||
additional_data: dict[str, str] | None = None,
|
||||
) -> BytesIO:
|
||||
"""Decrypt a file using the provided password.
|
||||
|
||||
Args:
|
||||
input_file_path (str): The path to the input file to be decrypted.
|
||||
password (str): The password used for decryption.
|
||||
additional_data (dict[str, str] | None): Optional additional authenticated data (AAD).
|
||||
|
||||
Returns:
|
||||
BytesIO: The decrypted file data in memory.
|
||||
"""
|
||||
|
||||
# Check inputs
|
||||
if not isinstance(input_file_path, Path):
|
||||
raise TypeError("input_file_path must be a Path object")
|
||||
if not input_file_path.exists():
|
||||
raise FileNotFoundError(f"input_file_path {input_file_path} does not exist")
|
||||
if not input_file_path.is_file():
|
||||
raise ValueError("input_file_path must be a file")
|
||||
if not isinstance(password, str):
|
||||
raise TypeError("password must be a string")
|
||||
if len(password) == 0:
|
||||
raise ValueError("password cannot be empty")
|
||||
password_bytes = base64.urlsafe_b64decode(password.encode(cls._encoding))
|
||||
if len(password_bytes) != 32:
|
||||
raise ValueError("Password must decode to 32 bytes for AES-256")
|
||||
if additional_data is not None and not isinstance(additional_data, dict):
|
||||
raise TypeError("additional_data must be a dict if provided")
|
||||
|
||||
# Prepare additional authenticated data (AAD) if provided
|
||||
additional_data_bytes = None
|
||||
if additional_data is not None:
|
||||
additional_data_bytes = json.dumps(additional_data).encode(cls._encoding)
|
||||
|
||||
# Initialize AESGCM with derived key
|
||||
aesgcm = AESGCM(password_bytes)
|
||||
|
||||
# Read the input file
|
||||
with open(input_file_path, "rb") as file:
|
||||
file_data = file.read()
|
||||
|
||||
# Split the nonce and encrypted data
|
||||
nonce = file_data[: cls._nonce_length]
|
||||
encrypted_data = file_data[cls._nonce_length :]
|
||||
|
||||
# Decrypt the data
|
||||
decrypted_data = aesgcm.decrypt(nonce, encrypted_data, additional_data_bytes)
|
||||
|
||||
# # Extract the salt and encrypted data
|
||||
# salt = file_data[:16]
|
||||
# encrypted_data = file_data[16:]
|
||||
|
||||
# # Derive the key from the password using PBKDF2
|
||||
# kdf = PBKDF2HMAC(
|
||||
# algorithm=hashes.SHA256(),
|
||||
# length=32,
|
||||
# salt=salt,
|
||||
# iterations=480000, # OWASP recommended minimum for 2023+
|
||||
# )
|
||||
# key = base64.urlsafe_b64encode(kdf.derive(password.encode("utf-8")))
|
||||
# fernet = Fernet(key)
|
||||
|
||||
# # Decrypt the data
|
||||
# decrypted_data = fernet.decrypt(encrypted_data)
|
||||
|
||||
return BytesIO(decrypted_data)
|
||||
|
||||
@classmethod
|
||||
def decrypt_to_disk(
|
||||
cls,
|
||||
input_file_path: Path,
|
||||
output_file_path: Path,
|
||||
password: str,
|
||||
additional_data: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
"""Decrypt a file using the provided password and save to disk.
|
||||
|
||||
Args:
|
||||
input_file_path (str): The path to the input file to be decrypted.
|
||||
output_file_path (str): The path where the decrypted file will be saved.
|
||||
password (str): The password used for decryption.
|
||||
additional_data (dict[str, str] | None): Optional additional authenticated data (AAD).
|
||||
"""
|
||||
|
||||
decrypted_data_io = FileEncrypter.decrypt_file(
|
||||
input_file_path, password, additional_data
|
||||
)
|
||||
|
||||
# Write the decrypted data to the output file
|
||||
with open(output_file_path, "wb") as file:
|
||||
file.write(decrypted_data_io.getbuffer())
|
||||
@@ -0,0 +1,64 @@
|
||||
"""Definition of FileEncrypterInterface."""
|
||||
|
||||
from io import BytesIO
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class FileEncrypterInterface(ABC):
|
||||
"""Interface for file encrypters."""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def encrypt_file(
|
||||
cls,
|
||||
input_file_path: Path,
|
||||
output_file_path: Path,
|
||||
password: str,
|
||||
additional_data: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
"""Encrypt a file using the provided password.
|
||||
|
||||
Args:
|
||||
input_file_path (str): The path to the input file to be encrypted.
|
||||
output_file_path (str): The path where the encrypted file will be saved.
|
||||
password (str): The password used for encryption.
|
||||
additional_data (dict[str, str] | None): Optional additional authenticated data (AAD).
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def decrypt_file(
|
||||
cls,
|
||||
input_file_path: Path,
|
||||
password: str,
|
||||
additional_data: dict[str, str] | None = None,
|
||||
) -> BytesIO:
|
||||
"""Decrypt a file using the provided password.
|
||||
|
||||
Args:
|
||||
input_file_path (str): The path to the input file to be decrypted.
|
||||
password (str): The password used for decryption.
|
||||
additional_data (dict[str, str] | None): Optional additional authenticated data (AAD).
|
||||
|
||||
Returns:
|
||||
BytesIO: The decrypted file data in memory.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def decrypt_to_disk(
|
||||
cls,
|
||||
input_file_path: Path,
|
||||
output_file_path: Path,
|
||||
password: str,
|
||||
additional_data: dict[str, str] | None = None,
|
||||
) -> None:
|
||||
"""Decrypt a file and save it to disk.
|
||||
|
||||
Args:
|
||||
input_file_path (str): The path to the input file to be decrypted.
|
||||
output_file_path (str): The path where the decrypted file will be saved.
|
||||
password (str): The password used for decryption.
|
||||
additional_data (dict[str, str] | None): Optional additional authenticated data (AAD).
|
||||
"""
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Module importer for dynamically loading modules from encrypted files."""
|
||||
|
||||
from .module_importer import ModuleImporter
|
||||
|
||||
__all__ = ["ModuleImporter"]
|
||||
@@ -0,0 +1,144 @@
|
||||
"""Definition of ModuleImporter class."""
|
||||
|
||||
from io import BytesIO
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
import sys
|
||||
|
||||
from .module_importer_interface import ModuleImporterInterface
|
||||
|
||||
|
||||
class ModuleImporter(ModuleImporterInterface):
|
||||
"""Class to handle importing Python modules from various sources."""
|
||||
|
||||
@classmethod
|
||||
def import_module(
|
||||
cls,
|
||||
module_data: dict[str, BytesIO],
|
||||
main_module: str | None = None,
|
||||
) -> None:
|
||||
"""Imports a Python module from in-memory data.
|
||||
|
||||
Args:
|
||||
module_data (dict[str, BytesIO]): A dictionary mapping module names to their bytecode streams.
|
||||
main_module (str | None): The name of the main module to execute, if any.
|
||||
"""
|
||||
raise NotImplementedError("import_module is not implemented yet.")
|
||||
|
||||
@classmethod
|
||||
def import_module_from_disk(
|
||||
cls,
|
||||
module_path: Path,
|
||||
main_module: str | None = None,
|
||||
) -> None:
|
||||
"""Imports a Python module or package from a directory on disk.
|
||||
|
||||
Args:
|
||||
module_path (Path): The directory path containing Python modules to import.
|
||||
main_module (str | None): The filename (e.g., 'main.py') to treat as the main module.
|
||||
If specified, this module will have __name__ set to '__main__'.
|
||||
"""
|
||||
|
||||
# Check inputs
|
||||
if not isinstance(module_path, Path):
|
||||
raise TypeError("module_path must be a Path object")
|
||||
if not module_path.exists():
|
||||
raise FileNotFoundError(f"module_path {module_path} does not exist")
|
||||
if not module_path.is_dir():
|
||||
raise ValueError("module_path must be a directory")
|
||||
|
||||
# Find all Python files in the directory and subdirectories
|
||||
python_files = list(module_path.rglob("*.py"))
|
||||
if not python_files:
|
||||
raise ValueError(f"No Python files found in {module_path}")
|
||||
|
||||
# Add the module directory to sys.path so imports work
|
||||
module_path_str = str(module_path)
|
||||
path_added = False
|
||||
if module_path_str not in sys.path:
|
||||
sys.path.insert(0, module_path_str)
|
||||
path_added = True
|
||||
|
||||
try:
|
||||
# Import all Python files as modules (dependencies first, main module last)
|
||||
main_module_file = None
|
||||
init_files = []
|
||||
|
||||
# First pass: Load all non-main, non-__init__ modules (dependencies)
|
||||
for py_file in python_files:
|
||||
relative_path = py_file.relative_to(module_path)
|
||||
|
||||
# Check if this file is the main module
|
||||
is_main_module = main_module and relative_path.name == main_module
|
||||
# Check if this is an __init__.py file
|
||||
is_init_file = relative_path.name == "__init__.py"
|
||||
|
||||
if is_main_module:
|
||||
# Store the main module file for later execution
|
||||
main_module_file = py_file
|
||||
continue
|
||||
elif is_init_file:
|
||||
# Store __init__.py files for later execution
|
||||
init_files.append(py_file)
|
||||
continue
|
||||
|
||||
# Load regular dependency modules first
|
||||
cls._load_module(py_file, module_path, is_main=False)
|
||||
|
||||
# Second pass: Load __init__.py files (after their contents are available)
|
||||
for init_file in init_files:
|
||||
cls._load_module(init_file, module_path, is_main=False)
|
||||
|
||||
# Third pass: Load and execute the main module last (if specified)
|
||||
if main_module_file:
|
||||
cls._load_module(main_module_file, module_path, is_main=True)
|
||||
|
||||
finally:
|
||||
# Clean up sys.path
|
||||
if path_added and module_path_str in sys.path:
|
||||
sys.path.remove(module_path_str)
|
||||
|
||||
@staticmethod
|
||||
def _load_module(py_file: Path, module_path: Path, is_main: bool = False) -> None:
|
||||
"""Load a single Python module.
|
||||
|
||||
Args:
|
||||
py_file (Path): Path to the Python file to load
|
||||
module_path (Path): Base path for the module directory
|
||||
is_main (bool): Whether this is the main module
|
||||
"""
|
||||
relative_path = py_file.relative_to(module_path)
|
||||
|
||||
# Create module name from relative path
|
||||
if len(relative_path.parts) > 1:
|
||||
# Nested module: subdir/module.py -> subdir.module
|
||||
parts = list(relative_path.parts[:-1]) + [relative_path.stem]
|
||||
module_name = ".".join(parts)
|
||||
else:
|
||||
# Top-level module
|
||||
module_name = relative_path.stem
|
||||
|
||||
# Use "__main__" as the spec name for main modules
|
||||
spec_name = "__main__" if is_main else module_name
|
||||
|
||||
# Import the module from the file
|
||||
spec = importlib.util.spec_from_file_location(spec_name, py_file)
|
||||
if spec is None:
|
||||
raise ImportError(f"Could not create spec for module {module_name}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
|
||||
# Set module attributes
|
||||
module.__file__ = str(py_file)
|
||||
module.__package__ = (
|
||||
".".join(relative_path.parts[:-1]) if len(relative_path.parts) > 1 else None
|
||||
)
|
||||
|
||||
# Execute the module
|
||||
if spec.loader is None:
|
||||
raise ImportError(f"No loader found for module {module_name}")
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
# Add to sys.modules
|
||||
if is_main:
|
||||
sys.modules["__main__"] = module
|
||||
sys.modules[module_name] = module # Also keep regular name for imports
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Definition of ModuleImporterInterface."""
|
||||
|
||||
from io import BytesIO
|
||||
from abc import ABC, abstractmethod
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class ModuleImporterInterface(ABC):
|
||||
"""Interface for module importer implementations."""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def import_module(
|
||||
cls,
|
||||
module_data: dict[str, BytesIO],
|
||||
main_module: str | None = None,
|
||||
) -> None:
|
||||
"""Imports a Python module from in-memory data.
|
||||
|
||||
Args:
|
||||
module_data (dict[str, BytesIO]): A dictionary mapping module names to their bytecode streams.
|
||||
main_module (str | None): The name of the main module to execute, if any.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def import_module_from_disk(
|
||||
cls,
|
||||
module_path: Path,
|
||||
main_module: str | None = None,
|
||||
) -> None:
|
||||
"""Imports a Python module from a file on disk.
|
||||
|
||||
Args:
|
||||
module_path (Path): The file path of the module on disk.
|
||||
main_module (str | None): The name of the main module to execute, if any.
|
||||
"""
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Password provider module."""
|
||||
|
||||
from .password_provider import PasswordProviderExample as PasswordProvider
|
||||
|
||||
__all__ = ["PasswordProvider"]
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Definition of PasswordProvider class."""
|
||||
|
||||
import os
|
||||
import json
|
||||
import base64
|
||||
import secrets
|
||||
from typing import Tuple
|
||||
|
||||
from .password_provider_interface import PasswordProviderInterface
|
||||
|
||||
|
||||
class PasswordProviderExample(PasswordProviderInterface):
|
||||
"""Class to provide password for encryption.
|
||||
Please extend this class to implement your own password storage/retrieval logic.
|
||||
"""
|
||||
|
||||
_encoding: str = "utf-8"
|
||||
|
||||
@classmethod
|
||||
def generate_password(cls) -> str:
|
||||
"""Generate a new password AES-256-GCM compliant key (32 bytes, base64-url encoded).
|
||||
|
||||
Returns:
|
||||
str: The generated password.
|
||||
"""
|
||||
|
||||
# AES-256 requires a 32-byte key
|
||||
key_bytes = secrets.token_bytes(32)
|
||||
url_safe_string = base64.urlsafe_b64encode(key_bytes).decode(cls._encoding)
|
||||
return url_safe_string
|
||||
|
||||
@classmethod
|
||||
def get_decryption_keys(cls) -> Tuple[str, dict[str, str] | None]:
|
||||
"""Placeholder function to get decryption keys. Implement your own logic here.
|
||||
|
||||
Returns:
|
||||
password: str: The password fetched from the authentication provider.
|
||||
aad: dict[str, str] | None: The additional authenticated data.
|
||||
"""
|
||||
|
||||
# Simulate using metadata to authenticate and get decryption keys
|
||||
password_str = cls._get_password()
|
||||
|
||||
# Simulate using metadata to authenticate and get AAD
|
||||
aad_dict = cls._get_additional_authenticated_data()
|
||||
|
||||
return password_str, aad_dict
|
||||
|
||||
@classmethod
|
||||
def _collect_metadata(cls) -> dict[str, str]:
|
||||
"""Placeholder function to collect metadata. Implement your own logic here.
|
||||
|
||||
Returns:
|
||||
dict[str, str]: The metadata fetched from e.g. environment variables.
|
||||
"""
|
||||
|
||||
# Simulate fetching metadata from environment variables
|
||||
return {"DEPLOYED_MODULE_VERSION": "0.1.0"}
|
||||
|
||||
@classmethod
|
||||
def _get_additional_authenticated_data(cls) -> dict[str, str] | None:
|
||||
"""Placeholder function to get optional additional authenticated data (AAD). Implement your own logic here.
|
||||
|
||||
Returns:
|
||||
dict[str, str] | None: The additional authenticated data.
|
||||
"""
|
||||
|
||||
# Simulate collecting metadata about self to authorize against external provider
|
||||
_ = cls._collect_metadata()
|
||||
|
||||
# Simulate fetching AAD from an external provider
|
||||
aad_str = os.getenv("PEC_AAD", None)
|
||||
if aad_str is None:
|
||||
return None
|
||||
aad_dict = {str(k): str(v) for k, v in json.loads(aad_str).items()}
|
||||
|
||||
return aad_dict
|
||||
|
||||
@classmethod
|
||||
def _get_password(cls) -> str:
|
||||
"""Placeholder function to get password. Implement your own logic here.
|
||||
|
||||
Returns:
|
||||
str: The password fetched from the authentication provider.
|
||||
"""
|
||||
|
||||
# Simulate collecting metadata about self to authorize against external provider
|
||||
_ = cls._collect_metadata()
|
||||
|
||||
# Simulate fetching password from an external authentication provider
|
||||
if password_str := os.getenv("PEC_PASSWORD"):
|
||||
return password_str
|
||||
|
||||
raise ValueError(
|
||||
"PEC_PASSWORD environment variable not set. "
|
||||
"Please set it to use this PasswordProvider."
|
||||
)
|
||||
@@ -0,0 +1,26 @@
|
||||
"""Definition of PasswordProviderInterface."""
|
||||
|
||||
from typing import Tuple
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class PasswordProviderInterface(ABC):
|
||||
"""Interface for password providers."""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def generate_password(cls) -> str:
|
||||
"""Generate a new password for encryption and decryption.
|
||||
|
||||
Returns:
|
||||
str: The generated password.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def get_decryption_keys(cls) -> Tuple[str, dict[str, str] | None]:
|
||||
"""Get decryption keys for decryption.
|
||||
|
||||
Returns:
|
||||
dict: The decryption keys.
|
||||
"""
|
||||
@@ -0,0 +1,12 @@
|
||||
"""Utility functions for encryption and decryption."""
|
||||
|
||||
from .decrypt_to_disk import decrypt_to_disk
|
||||
from .encrypt_folder import encrypt_folder
|
||||
from .generate_password import generate_password
|
||||
|
||||
|
||||
__all__ = [
|
||||
"decrypt_to_disk",
|
||||
"encrypt_folder",
|
||||
"generate_password",
|
||||
]
|
||||
@@ -0,0 +1,56 @@
|
||||
"""Definition of decrypt_file_to_disk function."""
|
||||
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from python_encrypt_code import (
|
||||
DataZipper,
|
||||
FileEncrypter,
|
||||
PasswordProvider,
|
||||
)
|
||||
|
||||
|
||||
def decrypt_to_disk(
|
||||
encrypted_file_path: Path,
|
||||
output_path: Path,
|
||||
) -> None:
|
||||
"""Decrypt a file and save the contents to disk.
|
||||
|
||||
Args:
|
||||
encrypted_file_path: Path to the encrypted file
|
||||
output_path: Path where the decrypted folder will be saved
|
||||
"""
|
||||
|
||||
# Validate inputs
|
||||
if not encrypted_file_path.exists():
|
||||
raise FileNotFoundError(f"Encrypted file not found: {encrypted_file_path}")
|
||||
|
||||
if not encrypted_file_path.is_file():
|
||||
raise ValueError(f"Path is not a file: {encrypted_file_path}")
|
||||
|
||||
# Get decryption keys
|
||||
password, additional_data = PasswordProvider.get_decryption_keys()
|
||||
|
||||
# Create temporary zip file
|
||||
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip:
|
||||
temp_zip_path = Path(temp_zip.name)
|
||||
|
||||
try:
|
||||
# Step 1: Decrypt the file
|
||||
print(f"Decrypting: {encrypted_file_path}")
|
||||
decrypted_data = FileEncrypter.decrypt_file(
|
||||
input_file_path=encrypted_file_path,
|
||||
password=password,
|
||||
additional_data=additional_data,
|
||||
)
|
||||
|
||||
# Step 2: Extract the zip file to disk
|
||||
print(f"Extracting to: {output_path}")
|
||||
DataZipper.unzip_data_to_disk(decrypted_data, output_path)
|
||||
|
||||
print("✅ File decrypted and extracted successfully!")
|
||||
|
||||
finally:
|
||||
# Clean up temporary zip file
|
||||
if temp_zip_path.exists():
|
||||
temp_zip_path.unlink()
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Definition of encrypt_folder function."""
|
||||
|
||||
import tempfile
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from python_encrypt_code import (
|
||||
DataZipper,
|
||||
FileEncrypter,
|
||||
)
|
||||
|
||||
|
||||
def encrypt_folder(
|
||||
folder_path: Path,
|
||||
output_path: Path,
|
||||
password: str,
|
||||
aad: str | None = None,
|
||||
) -> None:
|
||||
"""Encrypt a folder.
|
||||
|
||||
Args:
|
||||
folder_path (Path): Path to the folder to encrypt
|
||||
output_path (Path): Path where the encrypted file will be saved
|
||||
password (str): Optional password to use. If None, generates a new one.
|
||||
aad (str | None): Optional JSON-formatted metadata that will be stored in with the encrypted file.
|
||||
"""
|
||||
|
||||
# Validate inputs
|
||||
if not isinstance(folder_path, Path):
|
||||
raise TypeError(f"folder_path must be a Path object: {folder_path}")
|
||||
if not folder_path.exists():
|
||||
raise FileNotFoundError(f"Folder not found: {folder_path}")
|
||||
if not folder_path.is_dir():
|
||||
raise ValueError(f"Path is not a directory: {folder_path}")
|
||||
if not isinstance(output_path, Path):
|
||||
raise TypeError(f"output_path must be a Path object: {output_path}")
|
||||
if not output_path.parent.exists():
|
||||
raise FileNotFoundError(
|
||||
f"Output directory does not exist: {output_path.parent}"
|
||||
)
|
||||
if not isinstance(password, str):
|
||||
raise TypeError("password must be a string")
|
||||
if len(password) == 0:
|
||||
raise ValueError("password cannot be an empty string")
|
||||
if aad is not None and not isinstance(aad, str):
|
||||
raise TypeError("aad must be a string or None")
|
||||
|
||||
# Convert additional_data to if provided
|
||||
if isinstance(aad, str):
|
||||
additional_data = dict(json.loads(aad))
|
||||
else:
|
||||
additional_data = None
|
||||
|
||||
# Create temporary zip file
|
||||
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip:
|
||||
temp_zip_path = Path(temp_zip.name)
|
||||
|
||||
try:
|
||||
# Step 1: Zip the folder
|
||||
print(f"Zipping folder: {folder_path}")
|
||||
DataZipper.zip_data(folder_path, temp_zip_path)
|
||||
|
||||
# Step 2: Encrypt the zip file
|
||||
print(f"Encrypting to: {output_path}")
|
||||
FileEncrypter.encrypt_file(
|
||||
input_file_path=temp_zip_path,
|
||||
output_file_path=output_path,
|
||||
password=password,
|
||||
additional_data=additional_data,
|
||||
)
|
||||
|
||||
print("✅ Folder encrypted successfully!")
|
||||
|
||||
finally:
|
||||
# Clean up temporary zip file
|
||||
if temp_zip_path.exists():
|
||||
temp_zip_path.unlink()
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Definition of generate_password function."""
|
||||
|
||||
from python_encrypt_code import PasswordProvider
|
||||
|
||||
|
||||
def generate_password() -> None:
|
||||
"""Generate and print a secure password."""
|
||||
|
||||
# Generate and print a secure password
|
||||
password = PasswordProvider.generate_password()
|
||||
print(password)
|
||||
Reference in New Issue
Block a user