AI辅助的DevSecOps：在生成阶段注入安全策略的左移实践

# 安全增强的代码生成Prompt
SECURITY_PROMPT = """
生成Python代码，遵循以下安全规范：
1. 所有用户输入必须验证和转义
2. 数据库操作使用参数化查询（防SQL注入）
3. 敏感数据必须加密存储
4. 不使用eval()等危险函数
5. 遵循OWASP Top 10安全实践

代码要求：
{user_requirement}
"""

class AISecurityReviewer:
    def review_code(self, code):
        """
        AI实时审查代码安全性
        """
        issues = []
        
        # 检查SQL注入
        if self.detect_sql_injection_risk(code):
            issues.append({
                'type': 'SQL_INJECTION',
                'severity': 'HIGH',
                'fix': '使用参数化查询'
            })
        
        # 检查XSS风险
        if self.detect_xss_risk(code):
            issues.append({
                'type': 'XSS',
                'severity': 'HIGH', 
                'fix': '对用户输入进行HTML转义'
            })
        
        return issues

# 不安全的代码
def get_user(name):
    query = f"SELECT * FROM users WHERE name = '{name}'"
    return db.execute(query)

# AI自动修复
def get_user(name):
    query = "SELECT * FROM users WHERE name = %s"
    return db.execute(query, (name,))  # 参数化查询

def register_user(username, password):
    # 风险1：密码明文存储
    # 风险2：SQL注入
    # 风险3：无输入验证
    query = f"INSERT INTO users VALUES ('{username}', '{password}')"
    db.execute(query)

import bcrypt
import re
from typing import Optional

class UserService:
    def register_user(self, username: str, password: str) -> Optional[dict]:
        """
        用户注册 - 安全实现
        
        安全措施：
        1. 输入验证：防止注入攻击
        2. 密码加密：使用bcrypt哈希
        3. 参数化查询：防止SQL注入
        4. 错误处理：不泄露敏感信息
        """
        # 1. 输入验证
        if not self._validate_username(username):
            raise ValueError("用户名格式无效")
        
        if not self._validate_password(password):
            raise ValueError("密码强度不足")
        
        # 2. 检查用户是否存在
        if self._user_exists(username):
            raise ValueError("用户已存在")
        
        # 3. 密码哈希
        password_hash = bcrypt.hashpw(
            password.encode('utf-8'), 
            bcrypt.gensalt()
        )
        
        # 4. 安全的数据库操作
        try:
            query = """
                INSERT INTO users (username, password_hash, created_at)
                VALUES (%s, %s, NOW())
            """
            db.execute(query, (username, password_hash))
            return {'username': username, 'status': 'created'}
        except Exception:
            # 不泄露数据库错误详情
            raise ValueError("注册失败，请稍后重试")
    
    def _validate_username(self, username: str) -> bool:
        """用户名验证：字母数字，长度3-20"""
        return bool(re.match(r'^[a-zA-Z0-9]{3,20}$', username))
    
    def _validate_password(self, password: str) -> bool:
        """密码强度验证：至少8位，包含大小写和数字"""
        return (
            len(password) >= 8 and
            any(c.isupper() for c in password) and
            any(c.islower() for c in password) and
            any(c.isdigit() for c in password)
        )

class SecurityPolicyConverter:
    def convert_to_prompt(self, policy_file, context):
        """
        将安全策略转换为AI Prompt
        """
        policies = yaml.load(policy_file)
        
        prompt = "生成代码时必须遵循以下安全策略：\n\n"
        
        for policy in policies:
            prompt += f"【{policy['id']}】{policy['name']}\n"
            prompt += f"级别：{policy['severity']}\n"
            prompt += "要求：\n"
            for rule in policy['rules']:
                prompt += f"  - {rule}\n"
            prompt += "\n"
        
        prompt += f"\n上下文：{context}\n"
        prompt += "请生成符合以上所有安全策略的代码。"
        
        return prompt