2026 技术观察：DevSecOps 进入安全发布阶段，代码扫描、依赖风险和灰度拦截成为研发新底座

import json
from datetime import datetime
from collections import defaultdict


CODE_CHANGES = [
    {
        "file": "app/payment.py",
        "content": "query = 'select * from orders where id=' + user_input",
        "author": "dev_A"
    },
    {
        "file": "config/settings.py",
        "content": "API_KEY = 'sk_test_123456'",
        "author": "dev_B"
    },
    {
        "file": "app/user.py",
        "content": "def get_user_profile(user_id): return db.get(user_id)",
        "author": "dev_C"
    }
]


DEPENDENCIES = [
    {
        "name": "requests",
        "version": "2.31.0",
        "risk_level": "normal"
    },
    {
        "name": "old-json-parser",
        "version": "0.8.1",
        "risk_level": "high"
    },
    {
        "name": "web-framework",
        "version": "3.2.0",
        "risk_level": "medium"
    }
]


RELEASE_CONFIG = {
    "service_name": "payment-service",
    "version": "1.8.0",
    "env": "prod",
    "gray_ratio": 0.1
}

SECURITY_RULES = [
    {
        "rule_id": "SQL_INJECTION",
        "keywords": ["select *", "+ user_input"],
        "level": "high",
        "message": "疑似 SQL 拼接，存在注入风险。"
    },
    {
        "rule_id": "SECRET_LEAK",
        "keywords": ["API_KEY", "sk_"],
        "level": "high",
        "message": "疑似硬编码密钥。"
    },
    {
        "rule_id": "DEBUG_CONFIG",
        "keywords": ["debug=True"],
        "level": "medium",
        "message": "生产环境不应开启调试模式。"
    }
]


def scan_code_security(code_changes):
    findings = []

    for change in code_changes:
        content = change["content"]

        for rule in SECURITY_RULES:
            hit = all(
                keyword in content
                for keyword in rule["keywords"]
            )

            if hit:
                findings.append({
                    "file": change["file"],
                    "author": change["author"],
                    "rule_id": rule["rule_id"],
                    "level": rule["level"],
                    "message": rule["message"]
                })

    return findings

def check_dependency_risk(dependencies):
    results = []

    for dep in dependencies:
        if dep["risk_level"] == "high":
            action = "block"
            message = "依赖存在高风险，建议升级或替换。"
        elif dep["risk_level"] == "medium":
            action = "review"
            message = "依赖存在中等风险，建议评估影响范围。"
        else:
            action = "pass"
            message = "依赖风险正常。"

        results.append({
            "name": dep["name"],
            "version": dep["version"],
            "risk_level": dep["risk_level"],
            "action": 31222.t.kuaisou.com 
            "message": message
        })

    return results

def calculate_release_risk(code_findings, dependency_results):
    score = 0
    issues = []

    for finding in code_findings:
        if finding["level"] == "high":
            score += 5
        elif finding["level"] == "medium":
            score += 3

        issues.append(finding["message"])

    for dep in dependency_results:
        if dep["risk_level"] == "high":
            score += 5
            issues.append(f"{dep['name']} 存在高风险依赖问题。")
        elif dep["risk_level"] == "medium":
            score += 2
            issues.append(f"{dep['name']} 存在中等风险依赖问题。")

    if score >= 10:
        level = "critical"
    elif score >= 6:
        level = "high"
    elif score >= 3:
        level = "medium"
    else:
        level = "low"

    return {
        "risk_score": score,
        "risk_level": level,
        "issues": issues
    }

def decide_release_gate(release_config, risk_result):
    level = risk_result["risk_level"]

    if release_config["env"] == "prod" and level in ["critical", "high"]:
        return {
            "decision": "block",
            "message": "生产发布存在高风险，已自动拦截。"
        }

    if level == "medium":
        return {
            "decision": "manual_approval",
            "message": "发布存在中等风险，需要安全负责人审批。"
        }

    return {
        "decision": "allow_gray",
        "message": "安全检查通过，可进入灰度发布。"
    }

def monitor_gray_release():
    metrics = {
        "error_rate": 0.03,
        "avg_latency_ms": 180,
        "security_alert_count": 0
    }

    if metrics["security_alert_count"] > 0:
        status = "security_alert"
        action = "rollback"
    elif metrics["error_rate"] > 0.05:
        status = "unstable"
        action = "pause"
    else:
        status = "stable"
        action = "continue"

    return {
        "metrics": metrics,
        "status": status,
        "action": action
    }

def run_devsecops_release_pipeline():
    code_findings = scan_code_security(
        CODE_CHANGES
    )

    dependency_results = check_dependency_risk(
        DEPENDENCIES
    )

    release_risk = calculate_release_risk(
        code_findings,
        dependency_results
    )

    gate_decision = decide_release_gate(
        RELEASE_CONFIG,
        release_risk
    )

    gray_result = None

    if gate_decision["decision"] == "allow_gray":
        gray_result = monitor_gray_release()

    report = {
        "report_name": "DevSecOps 安全发布报告",
        "release_config": RELEASE_CONFIG,
        "code_findings": code_findings,
        "dependency_results": dependency_results,
        "release_risk": release_risk,
        "gate_decision": gate_decision,
        "gray_result": 31221.t.kuaisou.com 
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_devsecops_release_pipeline()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))