import os
import threading
from typing import Optional
from olc.encryptor.encryptor import Encryptor
class OlcEncryptorRegistry:
"""
加密器注册表(单例模式)
提供全局加密器管理机制,只存储一个解密器实例。
允许在应用启动时通过编程方式注入自定义加密器。
"""
_encryptor: Optional[Encryptor] = None
_lock = threading.RLock()
@classmethod
def set(cls, encryptor: Encryptor) -> None:
"""
设置加密器实例
Args:
encryptor: 加密器实例,必须实现 Encryptor 接口
Raises:
TypeError: 如果 encryptor 没有实现 Encryptor 接口
Example:
>>> encryptor = AesEncryptor("password")
>>> OlcEncryptorRegistry.set(encryptor)
"""
if not isinstance(encryptor, Encryptor):
raise TypeError("Encryptor must implement the Encryptor interface")
with cls._lock:
cls._encryptor = encryptor
@classmethod
def get(cls) -> Optional[Encryptor]:
"""
获取加密器实例
Returns:
Encryptor: 加密器实例,如果未设置返回 None
Example:
>>> encryptor = OlcEncryptorRegistry.get()
>>> if encryptor:
... encrypted = encryptor.encrypt("data")
"""
with cls._lock:
return cls._encryptor
@classmethod
def clear(cls) -> None:
"""
清除加密器实例(主要用于测试)
Example:
>>> OlcEncryptorRegistry.clear()
"""
with cls._lock:
cls._encryptor = None
@classmethod
def is_set(cls) -> bool:
"""
检查加密器是否已设置
Returns:
bool: 如果已设置返回 True,否则返回 False
Example:
>>> if OlcEncryptorRegistry.is_set():
... print("加密器已设置")
"""
with cls._lock:
return cls._encryptor is not None
@classmethod
def create_aes(cls, password: str) -> Encryptor:
"""
创建 AES 加密器并设置为全局加密器
Args:
password: 加密密码
Returns:
AesEncryptor: 创建的 AES 加密器实例
Example:
>>> encryptor = OlcEncryptorRegistry.create_aes("my_password")
"""
from olc.encryptor.aes_encryptor import AesEncryptor
encryptor = AesEncryptor(password)
cls.set(encryptor)
return encryptor
@classmethod
def create_aes_from_env(cls, env_var: str, default_password: str = None) -> Encryptor:
"""
从环境变量创建 AES 加密器并设置为全局加密器
从指定的环境变量读取密码,如果环境变量不存在则使用默认密码。
如果都没有设置,会抛出 ValueError。
Args:
env_var: 环境变量名
default_password: 默认密码(环境变量不存在时使用)
Returns:
AesEncryptor: 创建的 AES 加密器实例
Raises:
ValueError: 环境变量未设置且没有提供默认密码
Example:
>>> # 从环境变量读取密码
>>> encryptor = OlcEncryptorRegistry.create_aes_from_env("OLC_ENCRYPTOR_PASSWORD")
>>>
>>> # 带默认值的环境变量
>>> encryptor = OlcEncryptorRegistry.create_aes_from_env(
... "OLC_ENCRYPTOR_PASSWORD",
... default_password="fallback_password"
... )
"""
password = os.environ.get(env_var, default_password)
if not password:
raise ValueError(
f"Environment variable '{env_var}' is not set and no default password provided. "
f"Please set the environment variable or provide a default password."
)
from olc.encryptor.aes_encryptor import AesEncryptor
encryptor = AesEncryptor(password)
cls.set(encryptor)
return encryptor