| 实践维度 | 核心方法 | 技术实现 | 安全等级 | 适用场景 | |---------|---------|---------|---------|---------| | **存储方式** | 环境变量 | `os.getenv()` + `.env`文件 | ★★★★☆ | 开发/生产环境 | | **配置文件** | 专用配置文件 | JSON/YAML + 加密存储 | ★★★☆☆ | 多环境配置 | | **密钥管理服务** | 云端KMS | AWS KMS/Azure Key Vault | ★★★★★ | 企业级应用 | | **运行时保护** | 内存加密 | `keyring`库 + 进程隔离 | ★★★★☆ | 敏感数据处理 | ## 1. 环境变量配置（推荐基础方案） ### 1.1 使用python-dotenv管理.env文件 ```python # requirements.txt python-dotenv>=1.0.0 # .env 文件配置（必须加入.gitignore） OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxx DATABASE_URL=postgresql://user:pass@localhost/db SECRET_KEY=your-secret-key-here # config.py import os from pathlib import Path from dotenv import load_dotenv # 加载.env文件 env_path = Path('.') / '.env' load_dotenv(dotenv_path=env_path) # 获取环境变量 class Config: OPENAI_API_KEY = os.getenv('OPENAI_API_KEY') DATABASE_URL = os.getenv('DATABASE_URL') SECRET_KEY = os.getenv('SECRET_KEY') # 验证关键配置 @classmethod def validate(cls): required_vars = ['OPENAI_API_KEY', 'SECRET_KEY'] missing = [var for var in required_vars if not getattr(cls, var)] if missing: raise ValueError(f"缺少必需的环境变量: {missing}") ``` ### 1.2 生产环境环境变量设置 ```bash # Linux/Mac export OPENAI_API_KEY="sk-proj-xxxxxxxxxxxxxxxxxxxx" export SECRET_KEY="$(openssl rand -hex 32)" # Windows PowerShell $env:OPENAI_API_KEY = "sk-proj-xxxxxxxxxxxxxxxxxxxx" $env:SECRET_KEY = [System.Convert]::ToBase64String([System.Security.Cryptography.RandomNumberGenerator]::GetBytes(32)) # 使用systemd服务文件（生产环境推荐） # /etc/systemd/system/myapp.service [Service] Environment="OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxx" Environment="SECRET_KEY=generated-secure-key-here" ``` ## 2. 配置文件加密方案 ### 2.1 使用cryptography库加密敏感配置 ```python # requirements.txt cryptography>=42.0.0 # config_encrypt.py from cryptography.fernet import Fernet import json import base64 import os class EncryptedConfig: def __init__(self, key_path='.encryption_key'): self.key_path = key_path self.cipher = self._init_cipher() def _init_cipher(self): """初始化加密器，生成或加载密钥""" if os.path.exists(self.key_path): with open(self.key_path, 'rb') as f: key = f.read() else: key = Fernet.generate_key() with open(self.key_path, 'wb') as f: f.write(key) os.chmod(self.key_path, 0o600) # 设置文件权限为仅所有者可读 return Fernet(key) def encrypt_config(self, config_data: dict, output_path='config.enc'): """加密配置文件""" json_str = json.dumps(config_data) encrypted_data = self.cipher.encrypt(json_str.encode()) with open(output_path, 'wb') as f: f.write(encrypted_data) return output_path def decrypt_config(self, input_path='config.enc'): """解密配置文件""" with open(input_path, 'rb') as f: encrypted_data = f.read() decrypted_data = self.cipher.decrypt(encrypted_data) return json.loads(decrypted_data.decode()) # 使用示例 config_manager = EncryptedConfig() # 加密配置 sensitive_config = { "api_key": "sk-proj-xxxxxxxxxxxxxxxxxxxx", "database_password": "secure_password_123", "jwt_secret": "jwt-secret-key-here" } config_manager.encrypt_config(sensitive_config) # 运行时解密 config = config_manager.decrypt_config() print(f"API Key: {config['api_key']}") ``` ## 3. 密钥管理服务集成 ### 3.1 AWS Secrets Manager集成 ```python # requirements.txt boto3>=1.34.0 # aws_secrets.py import boto3 import json from botocore.exceptions import ClientError class AWSSecretsManager: def __init__(self, region_name='us-east-1'): self.client = boto3.client( 'secretsmanager', region_name=region_name ) def get_secret(self, secret_name): """从AWS Secrets Manager获取密钥""" try: response = self.client.get_secret_value( SecretId=secret_name ) if 'SecretString' in response: secret = response['SecretString'] return json.loads(secret) else: decoded_binary_secret = base64.b64decode( response['SecretBinary'] ) return json.loads(decoded_binary_secret) except ClientError as e: error_code = e.response['Error']['Code'] if error_code == 'ResourceNotFoundException': print(f"密钥 {secret_name} 不存在") elif error_code == 'InvalidRequestException': print(f"请求无效: {e}") elif error_code == 'InvalidParameterException': print(f"参数无效: {e}") raise def create_secret(self, secret_name, secret_value): """创建新的密钥""" if isinstance(secret_value, dict): secret_value = json.dumps(secret_value) try: response = self.client.create_secret( Name=secret_name, SecretString=secret_value, Description='API密钥和敏感配置', Tags=[ {'Key': 'Environment', 'Value': 'Production'}, {'Key': 'Application', 'Value': 'MyPythonApp'} ] ) return response['ARN'] except ClientError as e: print(f"创建密钥失败: {e}") raise # 使用示例 secrets_manager = AWSSecretsManager() # 获取密钥 api_config = secrets_manager.get_secret('prod/api-keys') openai_key = api_config.get('OPENAI_API_KEY') database_url = api_config.get('DATABASE_URL') ``` ### 3.2 HashiCorp Vault集成 ```python # requirements.txt hvac>=2.0.0 # vault_client.py import hvac import os class VaultClient: def __init__(self, vault_url=None, token=None): self.vault_url = vault_url or os.getenv('VAULT_ADDR') self.token = token or os.getenv('VAULT_TOKEN') self.client = hvac.Client( url=self.vault_url, token=self.token ) # 验证连接 if not self.client.is_authenticated(): raise ConnectionError("Vault认证失败") def read_secret(self, path, mount_point='secret'): """读取Vault中的密钥""" try: response = self.client.secrets.kv.v2.read_secret_version( path=path, mount_point=mount_point ) return response['data']['data'] except hvac.exceptions.InvalidPath: print(f"密钥路径不存在: {path}") return None def write_secret(self, path, data, mount_point='secret'): """写入密钥到Vault""" self.client.secrets.kv.v2.create_or_update_secret( path=path, secret=data, mount_point=mount_point ) return True # 使用示例 vault = VaultClient() # 读取配置 app_config = vault.read_secret('data/prod/myapp') if app_config: api_key = app_config.get('api_key') db_config = app_config.get('database') ``` ## 4. 安全最佳实践 ### 4.1 密钥轮换策略 ```python # key_rotation.py import datetime from typing import Optional import hashlib class KeyRotationManager: def __init__(self, key_lifetime_days=90): self.key_lifetime = datetime.timedelta(days=key_lifetime_days) self.key_history = {} # 存储历史密钥用于平滑过渡 def should_rotate(self, key_name: str, creation_date: datetime) -> bool: """检查是否需要轮换密钥""" age = datetime.datetime.now() - creation_date return age > self.key_lifetime def generate_new_key(self, key_name: str) -> dict: """生成新密钥并记录历史""" import secrets import string # 生成随机密钥 alphabet = string.ascii_letters + string.digits + '!@#$%^&*' new_key = ''.join(secrets.choice(alphabet) for _ in range(64)) # 记录到历史 if key_name not in self.key_history: self.key_history[key_name] = [] self.key_history[key_name].append({ 'key': new_key, 'created_at': datetime.datetime.now(), 'hash': hashlib.sha256(new_key.encode()).hexdigest() }) # 保持最近5个历史密钥 if len(self.key_history[key_name]) > 5: self.key_history[key_name] = self.key_history[key_name][-5:] return { 'key': new_key, 'created_at': datetime.datetime.now(), 'expires_at': datetime.datetime.now() + self.key_lifetime } def validate_key(self, key_name: str, provided_key: str) -> bool: """验证密钥是否有效（支持历史密钥过渡）""" if key_name not in self.key_history: return False provided_hash = hashlib.sha256(provided_key.encode()).hexdigest() # 检查当前和历史密钥 for key_record in self.key_history[key_name]: if key_record['hash'] == provided_hash: return True return False ``` ### 4.2 访问控制与权限管理 ```python # access_control.py from functools import wraps import logging from typing import List, Callable class APIAccessControl: def __init__(self): self.allowed_ips = set() self.rate_limits = {} self.access_log = [] def ip_whitelist(self, allowed_ips: List[str]): """IP白名单装饰器""" def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): import socket client_ip = socket.gethostbyname(socket.gethostname()) if client_ip not in allowed_ips: logging.warning(f"未授权的IP访问: {client_ip}") raise PermissionError("IP地址未授权") return func(*args, **kwargs) return wrapper return decorator def rate_limit(self, calls_per_minute: int = 60): """速率限制装饰器""" def decorator(func: Callable): import time call_history = [] @wraps(func) def wrapper(*args, **kwargs): current_time = time.time() # 清理一分钟前的记录 call_history[:] = [ t for t in call_history if current_time - t < 60 ] if len(call_history) >= calls_per_minute: logging.error("API调用频率超限") raise RuntimeError("API调用频率超限，请稍后重试") call_history.append(current_time) return func(*args, **kwargs) return wrapper return decorator # 使用示例 access_control = APIAccessControl() @access_control.ip_whitelist(['192.168.1.100', '10.0.0.1']) @access_control.rate_limit(calls_per_minute=30) def sensitive_operation(api_key: str, data: dict): """受保护的操作""" # 执行需要密钥的操作 return {"status": "success", "data": data} ``` ## 5. 配置验证与监控 ### 5.1 配置完整性检查 ```python # config_validator.py import os import re from typing import Dict, List, Tuple class ConfigValidator: @staticmethod def validate_api_key_format(key: str, provider: str) -> bool: """验证API密钥格式""" patterns = { 'openai': r'^sk-[a-zA-Z0-9]{48}$', 'anthropic': r'^sk-ant-[a-zA-Z0-9]{48}$', 'google': r'^AIza[a-zA-Z0-9_-]{39}$', 'azure': r'^[a-f0-9]{32}$', } if provider not in patterns: return True # 未知提供商，跳过格式验证 pattern = patterns[provider] return bool(re.match(pattern, key)) @staticmethod def check_required_vars(config: Dict) -> Tuple[bool, List[str]]: """检查必需的环境变量""" required = [ 'API_KEY', 'SECRET_KEY', 'DATABASE_URL' ] missing = [] for var in required: if not config.get(var): missing.append(var) return len(missing) == 0, missing @staticmethod def audit_config_access(): """审计配置访问日志""" import logging import inspect logger = logging.getLogger('config_audit') def audit_decorator(func): @wraps(func) def wrapper(*args, **kwargs): caller = inspect.stack()[1] logger.info( f"配置访问 - 函数: {func.__name__}, " f"调用者: {caller.filename}:{caller.lineno}, " f"时间: {datetime.datetime.now()}" ) return func(*args, **kwargs) return wrapper return audit_decorator # 使用示例 validator = ConfigValidator() # 验证配置 config = { 'API_KEY': os.getenv('OPENAI_API_KEY'), 'SECRET_KEY': os.getenv('SECRET_KEY'), 'DATABASE_URL': os.getenv('DATABASE_URL') } is_valid, missing = validator.check_required_vars(config) if not is_valid: raise ValueError(f"缺少必需的配置项: {missing}") # 验证密钥格式 if not validator.validate_api_key_format(config['API_KEY'], 'openai'): raise ValueError("API密钥格式无效") ``` ### 5.2 安全扫描与漏洞检测 ```python # security_scanner.py import ast import os from pathlib import Path class SecurityScanner: def __init__(self): self.sensitive_patterns = [ r'api[_-]?key', r'secret[_-]?key', r'password', r'token', r'credential', r'private[_-]?key' ] def scan_file_for_hardcoded_secrets(self, file_path: Path) -> List[Dict]: """扫描文件中的硬编码密钥""" import re findings = [] try: with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # 检查Python文件中的字符串常量 if file_path.suffix == '.py': tree = ast.parse(content) for node in ast.walk(tree): if isinstance(node, ast.Str): for pattern in self.sensitive_patterns: if re.search(pattern, node.s, re.IGNORECASE): findings.append({ 'file': str(file_path), 'line': node.lineno, 'content': node.s[:50] + '...' if len(node.s) > 50 else node.s, 'type': 'hardcoded_secret' }) # 通用正则匹配 for pattern in self.sensitive_patterns: matches = re.finditer( f'{pattern}\\s*[=:]\\s*[\'\"]([^\'\"]+)[\'\"]', content, re.IGNORECASE ) for match in matches: findings.append({ 'file': str(file_path), 'match': match.group(), 'type': 'potential_secret' }) except Exception as e: print(f"扫描文件 {file_path} 时出错: {e}") return findings def scan_directory(self, directory: Path) -> Dict: """扫描整个目录""" results = { 'files_scanned': 0, 'findings': [], 'vulnerable_files': [] } for file_path in directory.rglob('*.py'): findings = self.scan_file_for_hardcoded_secrets(file_path) results['files_scanned'] += 1 if findings: results['findings'].extend(findings) results['vulnerable_files'].append(str(file_path)) return results # 使用示例 scanner = SecurityScanner() project_dir = Path('.') scan_results = scanner.scan_directory(project_dir) if scan_results['findings']: print(f"发现 {len(scan_results['findings'])} 个潜在安全问题") for finding in scan_results['findings']: print(f"文件: {finding['file']}, 类型: {finding['type']}") ``` ## 6. 多环境配置管理 ### 6.1 环境特定配置 ```python # config_manager.py from enum import Enum import json from typing import Dict, Any class Environment(Enum): DEVELOPMENT = "development" TESTING = "testing" STAGING = "staging" PRODUCTION = "production" class ConfigManager: def __init__(self, env: Environment = None): self.env = env or self._detect_environment() self.configs = self._load_configs() def _detect_environment(self) -> Environment: """检测当前环境""" env_str = os.getenv('APP_ENV', 'development').lower() env_map = { 'dev': Environment.DEVELOPMENT, 'development': Environment.DEVELOPMENT, 'test': Environment.TESTING, 'testing': Environment.TESTING, 'stage': Environment.STAGING, 'staging': Environment.STAGING, 'prod': Environment.PRODUCTION, 'production': Environment.PRODUCTION } return env_map.get(env_str, Environment.DEVELOPMENT) def _load_configs(self) -> Dict[str, Any]: """加载环境特定配置""" base_config = { 'app_name': 'MyPythonApp', 'debug': False, 'log_level': 'INFO' } env_configs = { Environment.DEVELOPMENT: { 'debug': True, 'log_level': 'DEBUG', 'api_endpoint': 'http://localhost:8000', 'database': { 'host': 'localhost', 'port': 5432, 'name': 'myapp_dev' } }, Environment.TESTING: { 'debug': True, 'log_level': 'DEBUG', 'api_endpoint': 'http://test-api.example.com', 'database': { 'host': 'test-db.example.com', 'port': 5432, 'name': 'myapp_test' } }, Environment.PRODUCTION: { 'debug': False, 'log_level': 'WARNING', 'api_endpoint': 'https://api.example.com', 'database': { 'host': 'prod-db.example.com', 'port': 5432, 'name': 'myapp_prod' } } } # 合并基础配置和环境特定配置 config = base_config.copy() config.update(env_configs.get(self.env, {})) # 从环境变量覆盖敏感配置 sensitive_vars = [ ('API_KEY', 'api_key'), ('DATABASE_PASSWORD', 'database.password'), ('SECRET_KEY', 'secret_key') ] for env_var, config_path in sensitive_vars: env_value = os.getenv(env_var) if env_value: self._set_nested_value(config, config_path, env_value) return config def _set_nested_value(self, config: Dict, path: str, value: Any): """设置嵌套配置值""" keys = path.split('.') current = config for key in keys[:-1]: if key not in current: current[key] = {} current = current[key] current[keys[-1]] = value def get(self, key: str, default: Any = None) -> Any: """获取配置值""" keys = key.split('.') value = self.configs try: for k in keys: value = value[k] return value except (KeyError, TypeError): return default # 使用示例 config = ConfigManager(Environment.PRODUCTION) # 获取配置值 api_key = config.get('api_key') db_host = config.get('database.host') debug_mode = config.get('debug', False) ``` 上述方案覆盖了Python密钥配置的完整生命周期管理。环境变量方案适合大多数场景，加密配置提供额外保护，云服务集成适合企业级应用。密钥轮换、访问控制和监控是确保长期安全的关键[ref_1][ref_2][ref_4]。安全扫描和配置验证应集成到CI/CD流程中[ref_6]。多环境管理确保不同部署环境的一致性[ref_5]。