from pydantic import BaseModel, validator
import re

class ScriptInput(BaseModel):
    filename: str
    
    @validator('filename')
    def validate_filename(cls, v):
        if not re.match(r'^[\w.\- ]+$', v):
            raise ValueError('非法文件名')
        if '..' in v:
            raise ValueError('禁止路径遍历')
        return v

# 使用
try:
    input_data = ScriptInput(filename=user_input)
except ValueError as e:
    print(f"验证失败: {e}")
    exit(1)

# ❌ 危险：硬编码密钥
API_KEY = "sk_live_1234567890abcdef"  # 提交到GitHub就完了
DATABASE_PASSWORD = "P@ssw0rd123!"

# 方案1：环境变量（配合.env文件，gitignore保护）
import os
from dotenv import load_dotenv

load_dotenv()  # 加载 .env 文件（不提交到Git）

API_KEY = os.environ.get('API_KEY')
if not API_KEY:
    raise EnvironmentError("未设置API_KEY环境变量")

# 方案2：使用密钥管理服务（以HashiCorp Vault为例）
import hvac

vault = hvac.Client(
    url=os.environ['VAULT_ADDR'],
    token=os.environ['VAULT_TOKEN']
)

secret = vault.secrets.kv.v2.read_secret_version(
    path='myapp/api-key'
)
API_KEY = secret['data']['data']['key']

# 方案3：云平台密钥管理（AWS Secrets Manager）
import boto3
from botocore.exceptions import ClientError

def get_secret(secret_name):
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name='us-east-1'
    )
    
    try:
        response = client.get_secret_value(SecretId=secret_name)
        return response['SecretString']
    except ClientError as e:
        print(f"密钥获取失败: {e}")
        exit(1)

# ❌ 危险：泄露敏感信息
try:
    result = api_call()
except Exception as e:
    print(f"错误详情: {e}")  # 可能包含API密钥、数据库连接字符串
    raise

import logging
import sys
from typing import Optional

class SecureLogger:
    def __init__(self, log_file: str = '/var/log/myapp.log'):
        self.logger = logging.getLogger('myapp')
        self.logger.setLevel(logging.INFO)
        
        # 文件处理器（限制访问）
        file_handler = logging.FileHandler(log_file)
        file_handler.setLevel(logging.INFO)
        
        # 控制台处理器（开发环境）
        console_handler = logging.StreamHandler(sys.stdout)
        console_handler.setLevel(logging.WARNING)
        
        # 自定义格式化器，过滤敏感信息
        class SecureFormatter(logging.Formatter):
            SENSITIVE_PATTERNS = [
                (r'password=([^&\s]+)', 'password=***'),
                (r'key=([^&\s]+)', 'key=***'),
                (r'token=([^&\s]+)', 'token=***'),
                (r'Authorization: Bearer ([^\s]+)', 'Authorization: Bearer ***'),
            ]
            
            def format(self, record):
                msg = super().format(record)
                for pattern, replacement in self.SENSITIVE_PATTERNS:
                    msg = re.sub(pattern, replacement, msg, flags=re.IGNORECASE)
                return msg
        
        formatter = SecureFormatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)
        
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)
    
    def safe_log(self, level: int, message: str, 
                 sensitive_data: Optional[dict] = None):
        """安全的日志记录方法"""
        # 在日志前清理敏感数据
        safe_message = self._sanitize(message)
        self.logger.log(level, safe_message)
    
    def _sanitize(self, text: str) -> str:
        # 移除潜在的敏感信息
        import re
        patterns = [
            (r'\b\d{4}[*-]\d{4}[*-]\d{4}[*-]\d{4}\b', '****-****-****-****'),  # 信用卡
            (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '***@***.***'),  # 邮箱
            (r'Bearer\s+[A-Za-z0-9\-._~+/]+=*', 'Bearer ***'),  # JWT
        ]
        for pattern, replacement in patterns:
            text = re.sub(pattern, replacement, text)
        return text

# 使用
logger = SecureLogger()
try:
    process_data(api_key=os.getenv('API_KEY'))
except Exception as e:
    logger.safe_log(logging.ERROR, f"处理失败: {str(e)}")
    sys.exit(1)  # 退出码而不是堆栈跟踪

#!/usr/bin/env python3
"""
依赖安全检查与自动更新脚本
"""
import subprocess
import json
from packaging import version

def check_vulnerabilities():
    """检查已知漏洞"""
    result = subprocess.run(
        ['pip-audit', '--format', 'json'],
        capture_output=True,
        text=True
    )
    
    vulnerabilities = json.loads(result.stdout)
    if vulnerabilities:
        print(f"发现 {len(vulnerabilities)} 个漏洞:")
        for vuln in vulnerabilities:
            print(f"  - {vuln['name']} {vuln['version']}: {vuln['vuln_id']}")
        return False
    return True

def update_dependencies():
    """安全地更新依赖"""
    # 使用 pip-review 或 pip-upgrader
    subprocess.run(['pip-review', '--auto'], check=True)
    
    # 更新后重新测试
    subprocess.run(['pytest'], check=True)

if __name__ == '__main__':
    if not check_vulnerabilities():
        print("建议更新依赖项...")
        response = input("是否立即更新？(y/N): ")
        if response.lower() == 'y':
            update_dependencies()

#!/usr/bin/env python3
"""
受限环境执行脚本
"""
import resource
import sys
import os

def set_resource_limits():
    """设置资源限制，防止DoS攻击"""
    # CPU时间限制（秒）
    resource.setrlimit(resource.RLIMIT_CPU, (30, 60))
    
    # 内存限制（字节）
    mem_limit = 256 * 1024 * 1024  # 256MB
    resource.setrlimit(resource.RLIMIT_AS, (mem_limit, mem_limit))
    
    # 文件大小限制
    resource.setrlimit(resource.RLIMIT_FSIZE, (10 * 1024 * 1024, 10 * 1024 * 1024))
    
    # 进程数限制
    resource.setrlimit(resource.RLIMIT_NPROC, (10, 20))

# 在main之前调用
if __name__ == '__main__':
    set_resource_limits()
    
    # 设置安全的umask
    os.umask(0o077)
    
    # 执行主逻辑
    main()

import json
import logging
from datetime import datetime

class AuditLogger:
    def __init__(self, log_file: str = '/var/log/audit.json'):
        self.log_file = log_file
        
    def log_event(self, event_type: str, user: str, 
                  resource: str, action: str, 
                  status: str, details: dict = None):
        """记录审计事件"""
        event = {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'event_type': event_type,
            'user': user,
            'resource': resource,
            'action': action,
            'status': status,
            'details': details or {},
            'source_ip': os.environ.get('REMOTE_ADDR', 'localhost')
        }
        
        with open(self.log_file, 'a') as f:
            f.write(json.dumps(event) + '\n')
    
    def log_api_call(self, service: str, endpoint: str, 
                     params: dict, response_code: int):
        """记录API调用"""
        # 清理参数中的敏感信息
        safe_params = self._sanitize_dict(params)
        self.log_event(
            event_type='api_call',
            user=os.getenv('USER', 'unknown'),
            resource=f'{service}:{endpoint}',
            action='call',
            status='success' if 200 <= response_code < 300 else 'failed',
            details={
                'params': safe_params,
                'response_code': response_code
            }
        )

# 使用示例
audit = AuditLogger()

try:
    result = process_file(filename)
    audit.log_event(
        'file_operation',
        os.getenv('USER'),
        filename,
        'process',
        'success',
        {'size': len(result)}
    )
except Exception as e:
    audit.log_event(
        'file_operation',
        os.getenv('USER'),
        filename,
        'process',
        'failed',
        {'error': str(e)}
    )