《网络数据安全风险评估办法》深度解读：密码学视角下的合规落地指南

#!/usr/bin/env python3
"""
企业密码资产清单自动扫描工具
基于《网络数据安全风险评估办法》第11-12条要求
运行依赖: pip install cryptography requests
"""
import json
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional

@dataclass
class CryptoAsset:
    """密码资产记录"""
    asset_type: str          # pki_key | tls_cert | symmetric_key | hash_algorithm
    algorithm: str           # RSA-2048 | ECC-P256 | SM2 | AES-256 | SM4 ...
    key_size: int            # 密钥长度（比特）
    purpose: str             # 用途描述
    location: str            # 存储位置
    owner: str               # 责任人
    created_at: Optional[str] = None
    expires_at: Optional[str] = None
    quantum_vulnerable: bool = False  # 是否量子脆弱
    compliance_status: str = "unknown"  # compliant | non_compliant | needs_review


def scan_tls_certificates(domains: List[str]) -> List[CryptoAsset]:
    """扫描域名的TLS证书信息"""
    assets = []
    for domain in domains:
        try:
            # 使用 openssl 获取证书信息
            cmd = f"echo | openssl s_client -servername {domain} -connect {domain}:443 2>/dev/null"
            result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=15)
            output = result.stdout
            
            # 提取算法和密钥长度
            if "Peer Public Key" in output or "Server Certificate" in output:
                # 简化解析：实际生产环境应使用 ssl.get_server_certificate()
                algo = "unknown"
                key_size = 0
                
                if "ecdsa" in output.lower() or "EC" in output:
                    algo = "ECDSA"
                    key_size = 256
                elif "rsa" in output.lower() or "RSA" in output:
                    algo = "RSA"
                    # 尝试提取密钥长度
                    for line in output.split('\n'):
                        if 'Public-Key' in line:
                            if '2048' in line:
                                key_size = 2048
                            elif '4096' in line:
                                key_size = 4096
                            else:
                                key_size = 2048
                
                # 判断量子脆弱性
                quantum_vuln = algo in ("RSA", "ECDSA", "DSA", "Diffie-Hellman")
                
                asset = CryptoAsset(
                    asset_type="tls_cert",
                    algorithm=algo,
                    key_size=key_size,
                    purpose=f"TLS证书 - {domain}",
                    location=f"DNS:{domain}",
                    owner="security-team",
                    quantum_vulnerable=quantum_vuln,
                    compliance_status="non_compliant" if quantum_vuln else "compliant"
                )
                assets.append(asset)
                print(f"  ✅ {domain}: {algo}-{key_size} (量子脆弱: {quantum_vuln})")
        except Exception as e:
            print(f"  ⚠️ {domain}: 扫描失败 - {e}")
    
    return assets


def assess_key_compliance(assets: List[CryptoAsset]) -> dict:
    """评估密码资产合规状态"""
    total = len(assets)
    quantum_vulnerable = sum(1 for a in assets if a.quantum_vulnerable)
    unknown = sum(1 for a in assets if a.algorithm == "unknown")
    
    # 《办法》第14条评估标准：传输加密强度是否足够
    # 根据《商用密码管理条例》和 GM/T 0054 要求：
    # - 关键信息基础设施应使用国密或AES-256
    # - RSA-2048以下、ECC-256以下视为弱加密
    weak_algorithms = ["DES", "3DES", "RC4", "MD5", "RSA-1024"]
    
    report = {
        "scan_time": datetime.now(timezone.utc).isoformat(),
        "total_assets": total,
        "quantum_vulnerable": quantum_vulnerable,
        "quantum_vulnerable_ratio": f"{quantum_vulnerable/total*100:.1f}%" if total > 0 else "0%",
        "unknown_algorithms": unknown,
        "compliance_score": max(0, 100 - quantum_vulnerable * 20 - unknown * 10),  # 注：此为参考评分模型，非官方评估标准
        "recommendations": []
    }
    
    # 生成建议
    if quantum_vulnerable > 0:
        report["recommendations"].append({
            "priority": "HIGH",
            "category": "后量子迁移",
            "detail": f"发现 {quantum_vulnerable} 个量子脆弱密钥，建议制定CNSA 2.0/国密迁移计划",
            "deadline": "2027-01-01"
        })
    
    if unknown > 0:
        report["recommendations"].append({
            "priority": "MEDIUM",
            "category": "资产清单完善",
            "detail": f"发现 {unknown} 个未知算法资产，建议人工核查",
            "deadline": "2026-08-20"
        })
    
    return report


if __name__ == "__main__":
    print("=" * 60)
    print("企业密码资产清单扫描器 v1.0")
    print("依据：《网络数据安全风险评估办法》第11-12条")
    print("=" * 60)
    
    # 配置待扫描域名列表
    domains = sys.argv[1:] if len(sys.argv) > 1 else ["localhost"]
    
    print(f"\n[1/3] 扫描 {len(domains)} 个域名的TLS证书...")
    assets = scan_tls_certificates(domains)
    
    print(f"\n[2/3] 评估密码算法合规性...")
    report = assess_key_compliance(assets)
    
    print(f"\n[3/3] 生成评估报告:")
    print(json.dumps(report, indent=2, ensure_ascii=False))
    
    # 保存报告
    report_path = Path(f"crypto_scan_report_{datetime.now().strftime('%Y%m%d')}.json")
    report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding='utf-8')
    print(f"\n📄 报告已保存: {report_path}")

#!/bin/bash
# PKI合规检查脚本 - 检查证书算法和密钥长度
echo "=== PKI 密码合规检查 ==="
echo "检查时间: $(date -u +%Y-%m-%dT%H:%M:%SZ)"
echo ""

# 检查项1: 证书公钥算法
echo "--- 检查1: 证书公钥算法 ---"
for cert in /etc/ssl/certs/*.pem; do
    if [ -f "$cert" ]; then
        algo=$(openssl x509 -in "$cert" -noout -pubkey 2>/dev/null | openssl pkey -pubin -text 2>/dev/null | head -1)
        key_size=$(openssl x509 -in "$cert" -noout -text 2>/dev/null | grep "Public-Key" | awk '{print $NF}')
        subject=$(openssl x509 -in "$cert" -noout -subject 2>/dev/null | sed 's/subject=//')
        
        status="✅"
        if echo "$algo" | grep -qi "rsa"; then
            size_num=$(echo "$key_size" | grep -oP '\d+')
            if [ "$size_num" -lt 2048 ] 2>/dev/null; then
                status="❌ 密钥过短"
            fi
        fi
        
        if echo "$algo" | grep -qi "ecdsa"; then
            size_num=$(echo "$key_size" | grep -oP '\d+')
            if [ "$size_num" -lt 256 ] 2>/dev/null; then
                status="⚠️ 建议升级至256位以上"
            fi
        fi
        
        echo "  $status $subject | $algo $key_size"
    fi
done

# 检查项2: 即将过期的证书
echo ""
echo "--- 检查2: 证书有效期（90天内过期）---"
for cert in /etc/ssl/certs/*.pem; do
    if [ -f "$cert" ]; then
        expiry=$(openssl x509 -in "$cert" -noout -enddate 2>/dev/null | cut -d= -f2)
        expiry_epoch=$(date -d "$expiry" +%s 2>/dev/null)
        now_epoch=$(date +%s)
        days_left=$(( (expiry_epoch - now_epoch) / 86400 ))
        
        if [ "$days_left" -lt 90 ] 2>/dev/null; then
            subject=$(openssl x509 -in "$cert" -noout -subject 2>/dev/null | sed 's/subject=//')
            echo "  ⚠️ $subject | 剩余 $days_left 天"
        fi
    fi
done

# 检查项3: 证书链完整性
echo ""
echo "--- 检查3: 证书链完整性 ---"
for cert in /etc/ssl/certs/*.pem; do
    if [ -f "$cert" ]; then
        verify_result=$(openssl verify -CAfile /etc/ssl/certs/ca-certificates.crt "$cert" 2>&1)
        if echo "$verify_result" | grep -q "OK"; then
            echo "  ✅ $(basename $cert)"
        else
            echo "  ❌ $(basename $cert) | $verify_result"
        fi
    fi
done

echo ""
echo "=== 检查完成 ==="
echo "📋 建议: 将以上输出作为《风险评估报告》附件B（密码技术评估）"

# 符合《网络数据安全风险评估办法》要求的Nginx TLS配置
server {
    listen 443 ssl http2;
    
    # 1. 仅允许TLS 1.2+（禁用SSLv3/TLS 1.0/TLS 1.1）
    ssl_protocols TLSv1.2 TLSv1.3;
    
    # 2. 密码套件要求：优先国密/前向安全套件
    # 禁用已知弱密码套件（RC4, DES, 3DES, MD5签名）
    ssl_ciphers "ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-CHACHA20-POLY1305:TLS_AES_256_GCM_SHA384";
    ssl_prefer_server_ciphers on;
    
    # 3. 证书配置
    ssl_certificate     /etc/ssl/certs/server.pem;
    ssl_certificate_key /etc/ssl/private/server.key;
    ssl_trusted_certificate /etc/ssl/certs/chain.pem;
    
    # 4. 会话安全（减少握手开销，降低密钥暴露面）
    ssl_session_timeout 1d;
    ssl_session_cache shared:SSL:50m;
    ssl_session_tickets off;  # 开启后需定期轮换ticket密钥
    
    # 5. OCSP Stapling（证书状态实时验证）
    ssl_stapling on;
    ssl_stapling_verify on;
    resolver 114.114.114.114 valid=300s;
    
    # 6. 安全头（数据泄露防护）
    add_header Strict-Transport-Security "max-age=63072000; includeSubDomains" always;
    add_header X-Content-Type-Options "nosniff" always;
}

# 国密双证书方案：同时支持SM2和国际证书
# ⚠️ 环境要求: Nginx需编译国密模块（Tongsuo/GmSSL/BabaSSL），标准OpenSSL不支持以下密码套件
# 密码套件名称遵循 GM/T 0024 标准，需国密SSL库支持
server {
    listen 443 ssl;
    ssl_protocols TLSv1.2 TLSv1.3;
    
    # 国密密码套件（GM/T 0024）— 仅在国密SSL环境下可用
    ssl_ciphers "ECDHE-SM2-SM4-GCM-SM3:ECDHE-SM2-SM4-CBC-SM3";
    
    # SM2证书（优先协商）
    ssl_certificate /etc/ssl/certs/sm2_server.pem;
    ssl_certificate_key /etc/ssl/private/sm2_server.key;
    
    # 国际证书（fallback）
    ssl_certificate /etc/ssl/certs/rsa_server.pem;
    ssl_certificate_key /etc/ssl/private/rsa_server.key;
}

#!/usr/bin/env python3
"""
密钥生命周期管理自动化工具
符合《办法》第12条"数据安全管理制度建设情况"评估要求
运行依赖: pip install cryptography
"""
import os
import json
import hashlib
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Tuple

from cryptography.hazmat.primitives.asymmetric import ec, rsa, padding
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.backends import default_backend


class KeyLifecycleManager:
    """密钥生命周期管理器"""
    
    # 按数据级别定义轮换周期（天）
    ROTATION_POLICY = {
        "core": 180,      # 核心数据：6个月
        "important": 365, # 重要数据：1年
        "general": 730,   # 一般数据：2年
    }
    
    # 密钥状态
    KEY_STATES = ["active", "rotating", "deprecated", "destroyed"]
    
    def __init__(self, key_store_path: str):
        self.key_store = Path(key_store_path)
        self.key_store.mkdir(parents=True, exist_ok=True)
        self.metadata_file = self.key_store / "key_metadata.json"
        self.metadata = self._load_metadata()
    
    def _load_metadata(self) -> dict:
        if self.metadata_file.exists():
            return json.loads(self.metadata_file.read_text(encoding='utf-8'))
        return {"keys": {}, "rotation_log": []}
    
    def _save_metadata(self):
        self.metadata_file.write_text(
            json.dumps(self.metadata, indent=2, ensure_ascii=False),
            encoding='utf-8'
        )
    
    def generate_key_pair(self, key_id: str, algorithm: str = "ecdsa",
                         data_level: str = "important") -> dict:
        """生成密钥对并记录元数据"""
        
        if algorithm == "sm2":
            # ⚠️ SM2国密椭圆曲线与NIST P-256参数不同
            # 原生SM2需使用gmssl库：from gmssl import sm2, func; private_key = sm2.CryptSM2(None)
            # 此处使用SECP256R1作为兼容性近似，仅在非国密合规场景使用
            private_key = ec.generate_private_key(
                ec.SECP256R1(), default_backend()
            )
            algo_name = "SM2-256(SECP256R1近似)"
        elif algorithm == "ecdsa":
            private_key = ec.generate_private_key(
                ec.SECP384R1(), default_backend()
            )
            algo_name = "ECDSA-P384"
        elif algorithm == "rsa":
            private_key = rsa.generate_private_key(
                public_exponent=65537,
                key_size=4096,
                backend=default_backend()
            )
            algo_name = "RSA-4096"
        else:
            raise ValueError(f"不支持的算法: {algorithm}")
        
        now = datetime.now(timezone.utc)
        rotation_days = self.ROTATION_POLICY.get(data_level, 365)
        
        # 序列化密钥
        private_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.BestAvailableEncryption(
                b"change-this-password"  # 生产环境应从KMS获取
            )
        )
        public_pem = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo
        )
        
        # 保存密钥文件
        priv_path = self.key_store / f"{key_id}_private.pem"
        pub_path = self.key_store / f"{key_id}_public.pem"
        priv_path.write_bytes(private_pem)
        pub_path.write_bytes(public_pem)
        
        # 记录元数据
        key_record = {
            "key_id": key_id,
            "algorithm": algo_name,
            "data_level": data_level,
            "created_at": now.isoformat(),
            "next_rotation": (now + timedelta(days=rotation_days)).isoformat(),
            "rotation_period_days": rotation_days,
            "state": "active",
            "public_key_fingerprint": hashlib.sha256(public_pem).hexdigest()[:16],
            "quantum_vulnerable": algo_name in ("RSA-4096", "ECDSA-P384"),
        }
        
        self.metadata["keys"][key_id] = key_record
        self._save_metadata()
        
        print(f"✅ 密钥已生成: {key_id}")
        print(f"   算法: {algo_name}")
        print(f"   数据级别: {data_level}")
        print(f"   下次轮换: {key_record['next_rotation']}")
        print(f"   量子脆弱: {key_record['quantum_vulnerable']}")
        print(f"   指纹: {key_record['public_key_fingerprint']}")
        
        return key_record
    
    def check_rotation_needed(self) -> list:
        """检查需要轮换的密钥"""
        now = datetime.now(timezone.utc)
        needs_rotation = []
        
        for key_id, record in self.metadata["keys"].items():
            if record["state"] == "destroyed":
                continue
            
            next_rotation = datetime.fromisoformat(record["next_rotation"])
            days_until = (next_rotation - now).days
            
            if days_until <= 0:
                status = "⚠️ 已过期，需立即轮换"
                needs_rotation.append(key_id)
            elif days_until <= 30:
                status = f"⏰ {days_until}天后到期，建议提前轮换"
                needs_rotation.append(key_id)
            else:
                status = f"✅ 剩余{days_until}天"
            
            print(f"  {record['algorithm']} [{key_id}] {status}")
        
        return needs_rotation
    
    def rotate_key(self, old_key_id: str) -> dict:
        """执行密钥轮换"""
        if old_key_id not in self.metadata["keys"]:
            raise KeyError(f"密钥不存在: {old_key_id}")
        
        old_record = self.metadata["keys"][old_key_id]
        
        # 1. 标记旧密钥为轮换中
        old_record["state"] = "rotating"
        
        # 2. 生成新密钥
        new_key_id = f"{old_key_id}_v{datetime.now().strftime('%Y%m%d')}"
        new_record = self.generate_key_pair(
            new_key_id,
            algorithm="sm2" if "SM2" in old_record["algorithm"] else "ecdsa",  # 继承原算法类型
            data_level=old_record["data_level"]
        )
        
        # 3. 记录轮换日志
        rotation_entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "old_key_id": old_key_id,
            "new_key_id": new_key_id,
            "reason": "scheduled_rotation"
        }
        self.metadata.setdefault("rotation_log", []).append(rotation_entry)
        
        # 4. 标记旧密钥为已弃用（保留一段时间用于解密历史数据）
        old_record["state"] = "deprecated"
        old_record["deprecated_at"] = datetime.now(timezone.utc).isoformat()
        
        self._save_metadata()
        
        print(f"\n🔄 密钥轮换完成:")
        print(f"   旧密钥: {old_key_id} → deprecated")
        print(f"   新密钥: {new_key_id} → active")
        
        return new_record


if __name__ == "__main__":
    print("=" * 50)
    print("密钥生命周期管理工具 v1.0")
    print("=" * 50)
    
    manager = KeyLifecycleManager("/tmp/key_store")
    
    # 生成一个测试密钥
    print("\n[步骤1] 生成密钥对...")
    manager.generate_key_pair("web-server-tls", "sm2", "important")
    
    # 检查轮换状态
    print("\n[步骤2] 检查轮换状态...")
    needs_rotation = manager.check_rotation_needed()
    
    print(f"\n📊 需要轮换的密钥: {len(needs_rotation)} 个")
    print("\n💡 生产环境建议: 将轮换通知接入企业IM，提前30天预警")

#!/usr/bin/env python3
"""
数据库加密合规检查工具
依据：《网络数据安全风险评估办法》第14条（一）款
检查项：存储加密 + 传输加密 + 密钥管理
"""
import subprocess
import json
from dataclasses import dataclass
from typing import List
from datetime import datetime, timezone


@dataclass
class CheckResult:
    check_name: str
    status: str          # PASS | FAIL | WARN
    detail: str
    recommendation: str


def check_mysql_encryption(host: str = "localhost", user: str = "root") -> List[CheckResult]:
    """检查MySQL/MariaDB加密配置"""
    results = []
    
    # 检查1: SSL连接状态
    try:
        cmd = f"mysql -h {host} -u {user} -e \"SHOW STATUS LIKE 'Ssl_cipher';\" 2>/dev/null"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
        if "Ssl_cipher" in result.stdout and "Value" not in result.stdout:
            ssl_cipher = result.stdout.strip().split("\n")[-1].strip()
            if ssl_cipher:
                results.append(CheckResult(
                    "MySQL SSL连接", "PASS",
                    f"当前SSL密码套件: {ssl_cipher}",
                    "确保密码套件为AES256-GCM-SHA384或更高强度"
                ))
            else:
                results.append(CheckResult(
                    "MySQL SSL连接", "FAIL",
                    "SSL未启用",
                    "在my.cnf中配置: require_secure_transport=ON, ssl_ca/ssl_cert/ssl_key"
                ))
        else:
            results.append(CheckResult(
                "MySQL SSL连接", "WARN",
                "无法获取SSL状态或SSL未配置",
                "配置SSL连接: ALTER USER 'user'@'%' REQUIRE SSL;"
            ))
    except Exception as e:
        results.append(CheckResult("MySQL SSL连接", "WARN", f"检查失败: {e}", "手动检查SSL配置"))
    
    # 检查2: InnoDB表空间加密
    try:
        cmd = f"mysql -h {host} -u {user} -e \"SHOW VARIABLES LIKE 'innodb_encrypt%';\" 2>/dev/null"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
        if "ON" in result.stdout:
            results.append(CheckResult(
                "InnoDB表空间加密", "PASS",
                "InnoDB加密已启用",
                "确保使用keyring_file或keyring_encrypted_file插件"
            ))
        else:
            results.append(CheckResult(
                "InnoDB表空间加密", "WARN",
                "InnoDB加密未启用或未检测到",
                "安装keyring插件: INSTALL PLUGIN keyring_file SONAME 'keyring_file.so'"
            ))
    except Exception as e:
        results.append(CheckResult("InnoDB表空间加密", "WARN", f"检查失败: {e}", "手动检查"))
    
    return results


def check_postgres_encryption(host: str = "localhost") -> List[CheckResult]:
    """检查PostgreSQL加密配置"""
    results = []
    
    # 检查1: SSL状态
    try:
        cmd = f"psql -h {host} -c \"SHOW ssl;\" 2>/dev/null"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
        if "on" in result.stdout.lower():
            results.append(CheckResult(
                "PostgreSQL SSL", "PASS", "SSL已启用",
                "在postgresql.conf中确保ssl=on并配置强密码套件"
            ))
        else:
            results.append(CheckResult(
                "PostgreSQL SSL", "FAIL", "SSL未启用",
                "配置: ssl=on, ssl_min_protocol_version=TLSv1.2"
            ))
    except Exception as e:
        results.append(CheckResult("PostgreSQL SSL", "WARN", f"检查失败: {e}", "手动检查pg_hba.conf"))
    
    return results


def generate_compliance_report(results: List[CheckResult]) -> dict:
    """生成合规评估报告"""
    total = len(results)
    passed = sum(1 for r in results if r.status == "PASS")
    failed = sum(1 for r in results if r.status == "FAIL")
    warnings = sum(1 for r in results if r.status == "WARN")
    
    report = {
        "report_title": "数据库加密合规评估报告",
        "regulation": "《网络数据安全风险评估办法》",
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "summary": {
            "total_checks": total,
            "passed": passed,
            "failed": failed,
            "warnings": warnings,
            "compliance_rate": f"{passed/total*100:.0f}%" if total > 0 else "0%"
        },
        "findings": [
            {
                "check": r.check_name,
                "status": r.status,
                "detail": r.detail,
                "recommendation": r.recommendation
            }
            for r in results
        ]
    }
    
    return report


if __name__ == "__main__":
    print("=" * 50)
    print("数据库加密合规检查工具 v1.0")
    print("=" * 50)
    
    all_results = []
    
    print("\n[检查] MySQL加密配置...")
    all_results.extend(check_mysql_encryption())
    
    print("[检查] PostgreSQL加密配置...")
    all_results.extend(check_postgres_encryption())
    
    print("\n[报告] 生成合规评估...")
    report = generate_compliance_report(all_results)
    
    # 输出报告
    print(f"\n{'='*50}")
    print(f"合规率: {report['summary']['compliance_rate']}")
    print(f"通过: {report['summary']['passed']} | 失败: {report['summary']['failed']} | 警告: {report['summary']['warnings']}")
    print(f"{'='*50}")
    
    for finding in report["findings"]:
        icon = {"PASS": "✅", "FAIL": "❌", "WARN": "⚠️"}[finding["status"]]
        print(f"\n{icon} {finding['check']}: {finding['status']}")
        print(f"   详情: {finding['detail']}")
        print(f"   建议: {finding['recommendation']}")
    
    # 保存JSON报告
    report_path = Path(f"db_crypto_compliance_{datetime.now().strftime('%Y%m%d')}.json")
    report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False), encoding='utf-8')
    print(f"\n📄 JSON报告: {report_path}")