020_风险评估框架：系统化评估与防范Web3.0环境中的数字资产安全风险

风险识别 → 风险分析 → 风险评估 → 风险应对 → 持续监控

风险矩阵示例：
          影响程度
发生概率  低  中  高
低        低  中  中
中        中  高  高
高        中  高  极高

风险评估框架层次结构：
┌─────────────────────────────────────────┐
│              监控与改进层               │
├─────────────────────────────────────────┤
│              风险应对层                 │
├─────────────────────────────────────────┤
│              风险评价层                 │
├─────────────────────────────────────────┤
│              风险分析层                 │
├─────────────────────────────────────────┤
│              风险识别层                 │
└─────────────────────────────────────────┘

// Web3风险评分工具 v1.0 (2025)
class Web3RiskAssessment {
  constructor() {
    this.riskFactors = {
      walletType: {
        hot: 80,    // 热钱包风险分
        cold: 30,   // 冷钱包风险分
        multi: 40   // 多签钱包风险分
      },
      backupMethod: {
        paper: 70,
        metal: 30,
        digital: 90,
        none: 100
      },
      twoFactorAuth: {
        enabled: 40,
        disabled: 95
      },
      permissionReview: {
        regular: 50,
        rare: 85,
        never: 100
      },
      deviceSecurity: {
        upToDate: 45,
        outdated: 85,
        jailbroken: 95
      },
      networkUsage: {
        home: 40,
        public: 85,
        vpn: 50
      }
    };
  }

  // 计算风险总分
  calculateRiskScore(userData) {
    let totalScore = 0;
    let factorCount = 0;

    // 计算每个因素的风险分
    Object.keys(userData).forEach(factor => {
      if (this.riskFactors[factor] && this.riskFactors[factor][userData[factor]] !== undefined) {
        totalScore += this.riskFactors[factor][userData[factor]];
        factorCount++;
      }
    });

    // 返回平均分
    return factorCount > 0 ? Math.round(totalScore / factorCount) : 0;
  }

  // 评估风险等级
  assessRiskLevel(score) {
    if (score >= 80) return { level: '高风险', color: 'red', recommendation: '立即采取安全措施' };
    if (score >= 60) return { level: '中高风险', color: 'orange', recommendation: '需要加强安全措施' };
    if (score >= 40) return { level: '中等风险', color: 'yellow', recommendation: '适度提高安全意识' };
    if (score >= 20) return { level: '低风险', color: 'green', recommendation: '保持当前安全习惯' };
    return { level: '极低风险', color: 'blue', recommendation: '安全措施优秀' };
  }

  // 生成风险报告
  generateRiskReport(userData) {
    const score = this.calculateRiskScore(userData);
    const riskLevel = this.assessRiskLevel(score);
    
    return {
      overallScore: score,
      riskLevel: riskLevel.level,
      color: riskLevel.color,
      recommendation: riskLevel.recommendation,
      detailedAnalysis: this.generateDetailedAnalysis(userData),
      improvementSuggestions: this.generateImprovementSuggestions(userData)
    };
  }

  // 生成详细分析
  generateDetailedAnalysis(userData) {
    const analysis = [];
    
    // 分析钱包类型风险
    if (userData.walletType) {
      analysis.push(this.analyzeWalletRisk(userData.walletType));
    }
    
    // 分析备份方法风险
    if (userData.backupMethod) {
      analysis.push(this.analyzeBackupRisk(userData.backupMethod));
    }
    
    // 可添加更多分析...
    
    return analysis;
  }

  // 生成改进建议
  generateImprovementSuggestions(userData) {
    const suggestions = [];
    
    // 钱包安全建议
    if (userData.walletType === 'hot') {
      suggestions.push('考虑使用硬件钱包存储大额资产');
      suggestions.push('热钱包仅保留日常使用资金');
    }
    
    // 备份安全建议
    if (userData.backupMethod === 'digital') {
      suggestions.push('使用加密存储备份助记词');
      suggestions.push('考虑使用金属备份方案提高耐久性');
    }
    
    if (userData.backupMethod === 'none') {
      suggestions.push('立即创建助记词备份');
      suggestions.push('使用防火防水材料存储');
    }
    
    // 可添加更多建议...
    
    return suggestions;
  }

  // 分析钱包风险
  analyzeWalletRisk(walletType) {
    switch(walletType) {
      case 'hot':
        return '热钱包风险较高，容易受到网络攻击和恶意软件威胁';
      case 'cold':
        return '冷钱包提供良好的安全保障，但使用不够便捷';
      case 'multi':
        return '多签钱包显著降低单点故障风险，但需要协调多方';
      default:
        return '未知钱包类型，无法评估风险';
    }
  }

  // 分析备份风险
  analyzeBackupRisk(backupMethod) {
    switch(backupMethod) {
      case 'paper':
        return '纸质备份易受损坏，需要防火防水保护';
      case 'metal':
        return '金属备份耐久性好，但成本较高';
      case 'digital':
        return '数字备份容易受到黑客攻击和勒索软件威胁';
      case 'none':
        return '没有备份意味着资产永久丢失的高风险';
      default:
        return '未知备份方法，无法评估风险';
    }
  }
}

// 使用示例
function exampleUsage() {
  const riskTool = new Web3RiskAssessment();
  
  // 用户数据示例
  const userData = {
    walletType: 'hot',
    backupMethod: 'digital',
    twoFactorAuth: 'disabled',
    permissionReview: 'rare',
    deviceSecurity: 'outdated',
    networkUsage: 'public'
  };
  
  // 生成风险报告
  const report = riskTool.generateRiskReport(userData);
  console.log('风险评估报告:', JSON.stringify(report, null, 2));
  
  return report;
}

# 智能合约风险评分模型 v2.0 (2025)
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier

class SmartContractRiskAssessment:
    def __init__(self):
        # 初始化风险因素权重
        self.risk_weights = {
            'code_complexity': 0.15,
            'external_calls': 0.20,
            'access_controls': 0.20,
            'gas_optimization': 0.10,
            'upgradeability': 0.15,
            'testing_coverage': 0.15,
            'documentation_quality': 0.05
        }
        
        # 初始化风险评分阈值
        self.risk_thresholds = {
            'critical': 80,
            'high': 60,
            'medium': 40,
            'low': 20
        }
    
    def calculate_risk_score(self, contract_metrics):
        """计算智能合约风险总分"""
        total_score = 0
        
        # 计算加权风险分数
        for factor, weight in self.risk_weights.items():
            if factor in contract_metrics:
                # 归一化分数到0-100范围
                normalized_score = min(max(contract_metrics[factor], 0), 100)
                total_score += normalized_score * weight
        
        return round(total_score, 2)
    
    def assess_risk_level(self, score):
        """根据风险分数评估风险等级"""
        if score >= self.risk_thresholds['critical']:
            return 'critical', '合约存在严重安全风险，需要立即修复'
        elif score >= self.risk_thresholds['high']:
            return 'high', '合约存在高风险漏洞，强烈建议修复'
        elif score >= self.risk_thresholds['medium']:
            return 'medium', '合约存在中等风险问题，建议在部署前修复'
        elif score >= self.risk_thresholds['low']:
            return 'low', '合约存在低风险问题，可以考虑优化'
        else:
            return 'minimal', '合约风险较低，安全性良好'
    
    def analyze_code_complexity(self, ast_nodes, function_count, cyclomatic_complexity):
        """分析代码复杂度风险"""
        # 基础复杂度分数
        complexity_score = min(ast_nodes / 1000 * 100, 100)
        
        # 函数数量风险加成
        if function_count > 50:
            complexity_score += 10
        elif function_count > 100:
            complexity_score += 20
        
        # 圈复杂度风险加成
        if cyclomatic_complexity > 10:
            complexity_score += 15
        elif cyclomatic_complexity > 20:
            complexity_score += 30
        
        return min(complexity_score, 100)
    
    def analyze_external_calls(self, external_call_count, untrusted_call_count, reentrancy_guards):
        """分析外部调用风险"""
        # 基础外部调用分数
        call_score = external_call_count * 5
        
        # 不受信任调用风险加成
        call_score += untrusted_call_count * 10
        
        # 缺少重入保护惩罚
        if not reentrancy_guards:
            call_score += 30
        
        return min(call_score, 100)
    
    def analyze_access_controls(self, role_based_access, access_control_flaws, admin_functions):
        """分析访问控制风险"""
        # 基础访问控制分数
        if not role_based_access:
            access_score = 70
        else:
            access_score = 20
        
        # 访问控制缺陷惩罚
        access_score += access_control_flaws * 15
        
        # 管理员功能风险调整
        if admin_functions > 5:
            access_score += 10
        
        return min(access_score, 100)
    
    def generate_risk_report(self, contract_data):
        """生成完整的风险评估报告"""
        # 计算各维度风险分数
        contract_metrics = {
            'code_complexity': self.analyze_code_complexity(
                contract_data['ast_nodes'], 
                contract_data['function_count'], 
                contract_data['cyclomatic_complexity']
            ),
            'external_calls': self.analyze_external_calls(
                contract_data['external_call_count'],
                contract_data['untrusted_call_count'],
                contract_data['reentrancy_guards']
            ),
            'access_controls': self.analyze_access_controls(
                contract_data['role_based_access'],
                contract_data['access_control_flaws'],
                contract_data['admin_functions']
            ),
            'gas_optimization': contract_data['gas_inefficiency_score'],
            'upgradeability': contract_data['upgrade_mechanism_risk'],
            'testing_coverage': 100 - contract_data['test_coverage'] * 100,
            'documentation_quality': 100 - contract_data['documentation_score'] * 100
        }
        
        # 计算总风险分数
        total_score = self.calculate_risk_score(contract_metrics)
        risk_level, recommendation = self.assess_risk_level(total_score)
        
        # 生成风险报告
        report = {
            'contract_name': contract_data['contract_name'],
            'total_risk_score': total_score,
            'risk_level': risk_level,
            'recommendation': recommendation,
            'detailed_metrics': contract_metrics,
            'security_issues': self.identify_security_issues(contract_metrics),
            'remediation_steps': self.generate_remediation_steps(contract_metrics)
        }
        
        return report
    
    def identify_security_issues(self, metrics):
        """识别主要安全问题"""
        issues = []
        
        if metrics['code_complexity'] > 70:
            issues.append('代码过于复杂，增加了审计难度和出错概率')
        
        if metrics['external_calls'] > 60:
            issues.append('外部调用风险高，存在潜在的重入攻击风险')
        
        if metrics['access_controls'] > 50:
            issues.append('访问控制机制存在缺陷，可能导致未授权操作')
        
        if metrics['gas_optimization'] > 60:
            issues.append('燃料优化不足，可能导致交易失败或高成本')
        
        if metrics['upgradeability'] > 70:
            issues.append('合约升级机制存在安全风险')
        
        if metrics['testing_coverage'] > 60:
            issues.append('测试覆盖率不足，可能存在未发现的漏洞')
        
        return issues
    
    def generate_remediation_steps(self, metrics):
        """生成修复建议"""
        steps = []
        
        if metrics['code_complexity'] > 70:
            steps.append('重构复杂函数，遵循单一职责原则')
            steps.append('减少嵌套层级，提高代码可读性')
        
        if metrics['external_calls'] > 60:
            steps.append('实施检查-效果-交互模式防止重入攻击')
            steps.append('使用ReentrancyGuard库保护关键函数')
            steps.append('限制对不受信任合约的调用')
        
        if metrics['access_controls'] > 50:
            steps.append('实施严格的基于角色的访问控制')
            steps.append('审查所有管理功能的权限设置')
            steps.append('添加多重签名保护关键操作')
        
        return steps

# 使用示例
def example_contract_assessment():
    # 初始化评估工具
    risk_assessor = SmartContractRiskAssessment()
    
    # 示例合约数据
    contract_data = {
        'contract_name': 'DeFiLendingPlatform',
        'ast_nodes': 3500,
        'function_count': 45,
        'cyclomatic_complexity': 18,
        'external_call_count': 12,
        'untrusted_call_count': 5,
        'reentrancy_guards': False,
        'role_based_access': True,
        'access_control_flaws': 2,
        'admin_functions': 8,
        'gas_inefficiency_score': 55,
        'upgrade_mechanism_risk': 65,
        'test_coverage': 0.75,
        'documentation_score': 0.6
    }
    
    # 生成风险报告
    report = risk_assessor.generate_risk_report(contract_data)
    
    # 打印报告
    print(f"合约名称: {report['contract_name']}")
    print(f"风险总分: {report['total_risk_score']}")
    print(f"风险等级: {report['risk_level']}")
    print(f"建议: {report['recommendation']}")
    print("\n详细指标:")
    for metric, score in report['detailed_metrics'].items():
        print(f"  - {metric}: {score}")
    print("\n安全问题:")
    for issue in report['security_issues']:
        print(f"  - {issue}")
    print("\n修复步骤:")
    for step in report['remediation_steps']:
        print(f"  - {step}")
    
    return report