大家好，我是扣扣。今天聊一个自动化脚本中非常重要但容易被忽视的话题——环境变量的安全配置。

为什么环境变量配置这么重要？

看几个常见的场景：

1. 数据库密码硬编码

# ❌ 这样写很危险
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='123456',  # 暴露了！
    database='mydb'
)

2. API密钥泄露

# ❌ GitHub上一搜一片
API_KEY = "sk-xxxxxxxxxxxxxxxxxxxxxxxx"

3. 多环境配置混乱

# ❌ 开发生产和测试混在一起
if env == 'dev':
    url = 'http://localhost:8080'
else:
    url = 'https://api.production.com'

这些问题轻则导致安全漏洞，重则账号被盗、数据泄露。今天就来系统性地解决这个问题。

一、.env文件基础

.env文件是存储环境变量的标准方式，配合python-dotenv库使用。

项目结构

myproject/
├── .env              # 敏感配置（不上传git）
├── .env.example      # 配置模板（上传git）
├── .gitignore
├── config.py
└── main.py

.env文件格式

# .env 文件
# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=supersecret123
DB_NAME=mydb

# API密钥
API_KEY=sk-xxxxxxxxxxxxxxxx
API_SECRET=yyyyyyyyyyyyyyyy

# 环境配置
ENV=development
DEBUG=true

# Redis配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=

# 日志级别
LOG_LEVEL=INFO

.gitignore配置

# .gitignore
.env
.env.local
.env.*.local
*.pyc
__pycache__/

.env.example模板

# .env.example - 配置模板
# 复制为.env后填入真实值

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=your_username
DB_PASSWORD=your_password
DB_NAME=your_database

# API密钥
API_KEY=your_api_key
API_SECRET=your_api_secret

# 环境配置
ENV=development
DEBUG=false

# Redis配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=

# 日志级别
LOG_LEVEL=INFO

二、配置管理类实现

"""配置管理模块"""

import os
from pathlib import Path
from dotenv import load_dotenv
from dataclasses import dataclass
from typing import Optional
import json

class Config:
    """环境配置管理类"""
    
    _instance: Optional['Config'] = None
    _loaded: bool = False
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self):
        if not self._loaded:
            self._load_config()
            self._loaded = True
    
    def _load_config(self):
        """加载环境变量"""
        # 查找.env文件（项目根目录）
        env_path = Path.cwd() / '.env'
        
        # 也支持上级目录查找
        if not env_path.exists():
            env_path = Path(__file__).parent / '.env'
        
        if env_path.exists():
            load_dotenv(env_path)
            print(f"已加载配置文件: {env_path}")
        else:
            print("警告: 未找到.env文件，使用系统环境变量")
    
    def get(self, key: str, default: str = None) -> str:
        """获取环境变量"""
        return os.getenv(key, default)
    
    def get_int(self, key: str, default: int = 0) -> int:
        """获取整型环境变量"""
        value = os.getenv(key)
        if value is None:
            return default
        try:
            return int(value)
        except ValueError:
            return default
    
    def get_bool(self, key: str, default: bool = False) -> bool:
        """获取布尔环境变量"""
        value = os.getenv(key, '').lower()
        if value in ('true', '1', 'yes', 'on'):
            return True
        elif value in ('false', '0', 'no', 'off', ''):
            return default
        return default
    
    def get_list(self, key: str, separator: str = ',', default: list = None) -> list:
        """获取列表环境变量"""
        value = os.getenv(key)
        if value is None:
            return default or []
        return [item.strip() for item in value.split(separator) if item.strip()]
    
    def get_json(self, key: str, default: dict = None) -> dict:
        """获取JSON环境变量"""
        value = os.getenv(key)
        if value is None:
            return default or {}
        try:
            return json.loads(value)
        except json.JSONDecodeError:
            return default or {}


@dataclass
class DatabaseConfig:
    """数据库配置"""
    host: str
    port: int
    user: str
    password: str
    database: str
    
    @classmethod
    def from_env(cls) -> 'DatabaseConfig':
        """从环境变量加载"""
        config = Config()
        return cls(
            host=config.get('DB_HOST', 'localhost'),
            port=config.get_int('DB_PORT', 3306),
            user=config.get('DB_USER', 'root'),
            password=config.get('DB_PASSWORD', ''),
            database=config.get('DB_NAME', 'mydb')
        )
    
    def connection_string(self) -> str:
        """生成连接字符串"""
        return f"mysql+pymysql://{self.user}:{self.password}@{self.host}:{self.port}/{self.database}"


@dataclass  
class RedisConfig:
    """Redis配置"""
    host: str
    port: int
    password: Optional[str]
    db: int
    
    @classmethod
    def from_env(cls) -> 'RedisConfig':
        config = Config()
        password = config.get('REDIS_PASSWORD')
        return cls(
            host=config.get('REDIS_HOST', 'localhost'),
            port=config.get_int('REDIS_PORT', 6379),
            password=password if password else None,
            db=config.get_int('REDIS_DB', 0)
        )

三、敏感信息加密存储

.env文件虽然不上传git，但服务器上还是明文存储。有更高安全要求时，可以对敏感字段加密。

加密配置实现

"""敏感信息加密模块"""

import os
import base64
import hashlib
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2
from pathlib import Path

class SecretManager:
    """密钥管理器"""
    
    def __init__(self, password: str = None):
        # 从环境变量或参数获取主密钥
        master_key = password or os.getenv('MASTER_KEY')
        if not master_key:
            raise ValueError("需要设置MASTER_KEY环境变量")
        
        self.cipher = self._create_cipher(master_key)
    
    def _create_cipher(self, password: str) -> Fernet:
        """从密码创建加密器"""
        # 使用PBKDF2从密码派生密钥
        salt = b'fixed_salt_for_demo'  # 生产环境应存储salt
        kdf = PBKDF2(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return Fernet(key)
    
    def encrypt(self, plaintext: str) -> str:
        """加密字符串"""
        encrypted = self.cipher.encrypt(plaintext.encode())
        return base64.urlsafe_b64encode(encrypted).decode()
    
    def decrypt(self, ciphertext: str) -> str:
        """解密字符串"""
        decoded = base64.urlsafe_b64decode(ciphertext.encode())
        decrypted = self.cipher.decrypt(decoded)
        return decrypted.decode()


class EncryptedConfig:
    """加密配置读取器"""
    
    def __init__(self, secrets_file: str = '.secrets'):
        self.secrets_file = Path(secrets_file)
        self._secrets = {}
        self._load_secrets()
    
    def _load_secrets(self):
        """加载加密的配置"""
        if not self.secrets_file.exists():
            return
        
        with open(self.secrets_file, 'r') as f:
            for line in f:
                line = line.strip()
                if '=' in line:
                    key, value = line.split('=', 1)
                    self._secrets[key.strip()] = value.strip()
    
    def get(self, key: str, master_key: str = None) -> str:
        """获取并解密配置"""
        encrypted_value = self._secrets.get(key)
        if encrypted_value and master_key:
            manager = SecretManager(master_key)
            return manager.decrypt(encrypted_value)
        return encrypted_value


# 使用示例
if __name__ == '__main__':
    # 加密配置
    manager = SecretManager('my_secure_password')
    
    encrypted_db_password = manager.encrypt('supersecret123')
    encrypted_api_key = manager.encrypt('sk-api-key-xxxxx')
    
    # 保存到.secrets文件
    with open('.secrets', 'w') as f:
        f.write(f"DB_PASSWORD={encrypted_db_password}\n")
        f.write(f"API_KEY={encrypted_api_key}\n")
    
    print(f"加密后的密码: {encrypted_db_password}")
    print(f"解密后的密码: {manager.decrypt(encrypted_db_password)}")

四、多环境配置

不同环境使用不同的配置：

环境目录结构

config/
├── __init__.py
├── base.py          # 基础配置
├── development.py   # 开发环境
├── staging.py       # 测试环境
└── production.py    # 生产环境

配置实现

"""多环境配置"""

from pathlib import Path
from dataclasses import dataclass
from typing import Dict, Any
import os

class BaseConfig:
    """基础配置"""
    
    # 通用配置
    APP_NAME = "MyApp"
    DEBUG = False
    LOG_LEVEL = "INFO"
    
    # 数据库（默认值）
    DB_HOST = "localhost"
    DB_PORT = 3306
    DB_NAME = "mydb"
    
    # Redis
    REDIS_HOST = "localhost"
    REDIS_PORT = 6379
    REDIS_TTL = 3600
    
    # API配置
    API_TIMEOUT = 30
    API_RETRY = 3
    
    # 文件上传
    UPLOAD_DIR = "./uploads"
    MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB


class DevelopmentConfig(BaseConfig):
    """开发环境配置"""
    DEBUG = True
    LOG_LEVEL = "DEBUG"
    DB_HOST = "localhost"
    DB_NAME = "mydb_dev"


class StagingConfig(BaseConfig):
    """测试环境配置"""
    DEBUG = False
    LOG_LEVEL = "INFO"
    DB_HOST = "staging-db.example.com"
    DB_NAME = "mydb_staging"


class ProductionConfig(BaseConfig):
    """生产环境配置"""
    DEBUG = False
    LOG_LEVEL = "WARNING"
    DB_HOST = "prod-db.example.com"
    DB_NAME = "mydb_production"
    REDIS_HOST = "prod-redis.example.com"
    # 更严格的安全配置
    ALLOWED_HOSTS = ["example.com", "www.example.com"]


# 配置映射
CONFIG_MAP: Dict[str, BaseConfig] = {
    'development': DevelopmentConfig,
    'dev': DevelopmentConfig,
    'staging': StagingConfig,
    'test': StagingConfig,
    'production': ProductionConfig,
    'prod': ProductionConfig,
}


def get_config(env: str = None) -> BaseConfig:
    """获取当前环境的配置"""
    if env is None:
        env = os.getenv('APP_ENV', 'development')
    
    config_class = CONFIG_MAP.get(env.lower(), DevelopmentConfig)
    return config_class()


# 全局配置实例
config = get_config()

使用示例

"""应用主文件"""

from config import config

# 自动根据环境加载配置
print(f"运行环境: {config.APP_NAME}")
print(f"调试模式: {config.DEBUG}")
print(f"数据库: {config.DB_HOST}/{config.DB_NAME}")

# 使用配置连接数据库
def get_db_connection():
    import pymysql
    return pymysql.connect(
        host=config.DB_HOST,
        port=config.DB_PORT,
        user=os.getenv('DB_USER'),
        password=os.getenv('DB_PASSWORD'),
        database=config.DB_NAME
    )

五、环境配置验证

启动时验证配置是否正确：

"""配置验证模块"""

import os
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class ValidationError:
    """验证错误"""
    field: str
    message: str

class ConfigValidator:
    """配置验证器"""
    
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.errors: List[ValidationError] = []
    
    def required(self, *fields: str) -> 'ConfigValidator':
        """验证必填字段"""
        for field in fields:
            if not self.config.get(field):
                self.errors.append(ValidationError(
                    field=field,
                    message=f"缺少必填配置: {field}"
                ))
        return self
    
    def in_range(self, field: str, min_val: int, max_val: int) -> 'ConfigValidator':
        """验证数值范围"""
        value = self.config.get(field)
        if value is not None:
            try:
                num = int(value)
                if not min_val <= num <= max_val:
                    self.errors.append(ValidationError(
                        field=field,
                        message=f"{field}值 {num} 超出范围 [{min_val}, {max_val}]"
                    ))
            except (ValueError, TypeError):
                self.errors.append(ValidationError(
                    field=field,
                    message=f"{field}必须是整数"
                ))
        return self
    
    def one_of(self, field: str, choices: List[str]) -> 'ConfigValidator':
        """验证枚举值"""
        value = self.config.get(field)
        if value and value not in choices:
            self.errors.append(ValidationError(
                field=field,
                message=f"{field}值必须是 {choices} 之一"
            ))
        return self
    
    def validate(self) -> bool:
        """执行验证"""
        if self.errors:
            for error in self.errors:
                print(f"配置错误 [{error.field}]: {error.message}")
            return False
        return True


# 启动时验证
def validate_startup_config():
    """验证启动配置"""
    from config import config
    
    validator = ConfigValidator(vars(config))
    
    validator \
        .required('DB_HOST', 'DB_NAME') \
        .in_range('DB_PORT', 1, 65535) \
        .one_of('LOG_LEVEL', ['DEBUG', 'INFO', 'WARNING', 'ERROR'])
    
    if not validator.validate():
        raise ValueError("配置验证失败，无法启动应用")
    
    print("✓ 配置验证通过")


if __name__ == '__main__':
    validate_startup_config()

六、Docker/K8s环境变量

在容器化部署时，配置方式略有不同：

# Dockerfile
FROM python:3.10-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

# 构建时不传密码，构建时用ARG
# ARG DB_PASSWORD
# ENV DB_PASSWORD=${DB_PASSWORD}

COPY . .
CMD ["python", "main.py"]

# docker-compose.yml
version: '3.8'
services:
  app:
    build: .
    environment:
      - DB_HOST=db
      - DB_PORT=3306
      - DB_NAME=mydb
      # 从.env文件加载（仅用于本地开发）
      env_file:
        - .env
    secrets:
      - db_password
    depends_on:
      - db

secrets:
  db_password:
    file: ./secrets/db_password.txt

# Kubernetes ConfigMap
apiVersion: v1
kind: ConfigMap
metadata:
  name: app-config
data:
  APP_ENV: "production"
  LOG_LEVEL: "INFO"
---
# Kubernetes Secret
apiVersion: v1
kind: Secret
metadata:
  name: app-secrets
type: Opaque
stringData:
  DB_PASSWORD: "your-password-here"
  API_KEY: "your-api-key-here"

总结

1. 基础配置：使用.env文件 + python-dotenv
2. 配置类：用dataclass组织，类型安全
3. 敏感信息：加密存储，设置MASTER_KEY
4. 多环境：按环境分离配置类
5. 验证：启动时验证必要配置
6. 容器化：使用ConfigMap和Secret

环境配置虽小，但做好了对项目的可维护性和安全性至关重要。我是扣扣，有问题留言~🙃