能源行业数据安全合规实战：SM4 加密存储与 HMAC-SM3 完整性校验

前言

2026年7月1日，国家能源局《能源行业数据安全管理办法（试行）》（国能发规划规〔2025〕108号）正式施行。这是能源行业数据安全领域的第一部专项规章，标志着能源数据从此进入"强监管、可考核、可追责"的新时代。

该办法对密码技术提出了明确要求：

• 第十四条：能源行业重要数据处理者应每年至少开展一次数据安全风险评估
• 第十六条：综合运用加密、鉴权、认证、脱敏、校验、审计等技术手段进行安全保护
• 第二十一条：安全事件处置溯源相关日志留存不少于一年，向他人提供、委托处理相关日志留存不少于三年
• 第二十五条：能源行业核心数据的处理者应优先使用商用密码进行保护

法规解读：密码技术在能源数据安全中的定位

数据分级与密码要求

办法第四条将能源行业数据分为三级：

• 存储处理重要数据的信息网络应落实三级及以上等保（第十三条）
• 存储处理核心数据的信息网络，如涉及关基应在等保基础上落实关基保护；不涉及关基的应落实四级等保（第十三条）
• 核心数据处理者跨法人提供/转移/共享核心数据，累计达上年末静态总量 30% 及以上须经国家能源局组织风险评估（第二十四条）

密码技术的合规落地路径

办法第十六条列举了六项技术手段，其中与密码直接相关的是：

• 加密 — 数据机密性保护
• 鉴权 — 身份鉴别（SM2 数字证书）
• 认证 — 消息来源认证（HMAC-SM3）
• 校验 — 数据完整性（SM3 哈希）

实战：SM4 字段级加密存储

环境准备

# 运行环境：Python 3.9+，需安装 cryptography 库
# pip install cryptography
# 注意：cryptography 库支持 SM4，但不支持 SM2 密钥生成
# 如需 SM2 请使用 gmssl 库

代码实现

```python """ 能源行业数据安全合规 — SM4 字段级加密存储方案 依据：《能源行业数据安全管理办法（试行）》第十四、十六、二十五条

⚠️ 环境要求：

• Python 3.9+
• cryptography >= 41.0.0（支持 algorithms.SM4）
• 仅支持 SM4 对称加密，不支持 SM2/SM3（SM3 使用 hashlib）

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives import padding, hashes from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hmac as crypto_hmac import os import hmac as _stdlib_hmac import hashlib import json import base64 from datetime import datetime, timezone

class SM4FieldEncryption: """ SM4 字段级加密器 实现方案： - 加密模式：SM4-CBC（符合行业密码模块安全要求） - 密钥派生：HKDF-SM3（从主密钥派生工作密钥，salt 基于上下文保证可重复） - 完整性：HMAC-SM3（数据完整性校验） - 密钥管理：三级密钥体系（主密钥 → 密钥加密密钥 → 工作密钥） """ BLOCK_SIZE = 16 # SM4 分组大小：128 bits = 16 bytes KEY_SIZE = 16 # SM4 密钥大小：128 bits = 16 bytes def __init__(self, master_key: bytes): """ 初始化加密器 Args: master_key: 主密钥（16 字节 SM4 密钥） ⚠️ 生产环境中主密钥应存储在 HSM 或 KM 中， 不应以明文形式出现在代码或配置文件中 """ if len(master_key) != self.KEY_SIZE: raise ValueError(f"主密钥长度必须为 {self.KEY_SIZE} 字节") self.master_key = master_key self.backend = default_backend() def _derive_work_key(self, context: bytes) -> bytes: """ 从主密钥派生工作密钥（密钥加密密钥） 依据 GM/T 0028-2014 三级密钥体系： - 主密钥（MK）：系统根密钥，存储在 HSM 中 - 密钥加密密钥（KEK）：用于加密工作密钥 - 工作密钥（WK）：直接用于数据加密 Args: context: 上下文标识（如表名、字段名），确保不同字段使用不同密钥 Returns: 16 字节工作密钥 """ hkdf = HKDF( algorithm=hashes.SM3(), # 使用 SM3 作为 HKDF 的 PRF，salt 基于上下文派生保证确定性 length=self.KEY_SIZE, salt=hashlib.pbkdf2_hmac("sha256", context.encode("utf-8"), b"energy-salt", 1000)[:16], info=context.encode("utf-8"), backend=self.backend ) return hkdf.derive(self.master_key) def encrypt_field(self, plaintext: str, field_context: str) -> dict: """ 对单个字段进行 SM4-CBC 加密 Args: plaintext: 明文数据 field_context: 字段上下文标识（用于密钥派生） Returns: 加密结果字典，包含 nonce、密文和 HMAC """ # 1. 派生工作密钥 work_key = self._derive_work_key(field_context) # 2. 生成随机 IV（nonce） nonce = os.urandom(self.BLOCK_SIZE) # 3. PKCS7 填充 padder = padding.PKCS7(self.BLOCK_SIZE * 8).padder() padded_data = padder.update(plaintext.encode("utf-8")) + padder.finalize() # 4. SM4-CBC 加密 cipher = Cipher( algorithms.SM4(work_key), modes.CBC(nonce), backend=self.backend ) encryptor = cipher.encryptor() ciphertext = encryptor.update(padded_data) + encryptor.finalize() # 5. HMAC-SM3 完整性校验（加密+IV） mac = crypto_hmac.HMAC(work_key, hashes.SM3(), backend=self.backend) mac.update(nonce) mac.update(ciphertext) mac_digest = mac.finalize() return { "algorithm": "SM4-CBC", "nonce": base64.b64encode(nonce).decode("ascii"), "ciphertext": base64.b64encode(ciphertext).decode("ascii"), "hmac": base64.b64encode(mac_digest).decode("ascii"), "field": field_context, "encrypted_at": datetime.now(timezone.utc).isoformat() } def decrypt_field(self, encrypted_data: dict, field_context: str) -> str: """ 解密 SM4-CBC 加密的字段 先验证 HMAC 再解密，防止密文篡改攻击 Args: encrypted_data: encrypt_field 返回的字典 field_context: 字段上下文标识 Returns: 解密后的明文 """ work_key = self._derive_work_key(field_context) nonce = base64.b64decode(encrypted_data["nonce"]) ciphertext = base64.b64decode(encrypted_data["ciphertext"]) expected_hmac = base64.b64decode(encrypted_data["hmac"]) # 1. 先验证 HMAC（时序安全比较） mac = crypto_hmac.HMAC(work_key, hashes.SM3(), backend=self.backend) mac.update(nonce) mac.update(ciphertext) computed_hmac = mac.finalize() if not _stdlib_hmac.compare_digest(computed_hmac, expected_hmac): raise ValueError("完整性校验失败：密文可能被篡改") # 2. SM4-CBC 解密 cipher = Cipher( algorithms.SM4(work_key), modes.CBC(nonce), backend=self.backend ) decryptor = cipher.decryptor() padded_plaintext = decryptor.update(ciphertext) + decryptor.finalize() # 3. 去除 PKCS7 填充 unpadder = padding.PKCS7(self.BLOCK_SIZE * 8).unpadder() plaintext = unpadder.update(padded_plaintext) + unpadder.finalize() return plaintext.decode("utf-8")

class EnergyDataCompliance: """ 能源行业数据安全合规辅助类 实现法规要求的关键功能： - 数据分级标记 - 日志留存（1年/3年） - 密钥自动轮换 """ DATA_LEVELS = ["一般", "重要", "核心"] def __init__(self, master_key: bytes): self.encryption = SM4FieldEncryption(master_key) self.audit_log = [] self.key_rotation_date = datetime.now(timezone.utc) def classify_and_encrypt(self, record: dict, data_level: str) -> dict: """ 对数据记录进行分类分级并加密敏感字段 依据办法第四条：数据分为一般、重要、核心三级 依据办法第二十五条：核心数据优先使用商用密码保护 Args: record: 原始数据字典 data_level: 数据级别（"一般"/"重要"/"核心"） Returns: 加密后的记录，包含元数据和审计日志 """ if data_level not in self.DATA_LEVELS: raise ValueError(f"数据级别必须是 {self.DATA_LEVELS} 之一") # 定义哪些字段需要加密（根据行业数据分类分级标准） sensitive_fields = { "一般": [], # 一般数据无需加密 "重要": ["phone", "email", "gps_location", "meter_reading"], # 重要数据：手机号、邮箱、GPS、电表读数 "核心": ["phone", "email", "gps_location", "meter_reading", "grid_topology", "generation_capacity", "transformer_status"] # 核心数据：电网拓扑、发电容量、变压器状态 } encrypted_record = {"_metadata": { "data_level": data_level, "encrypted_at": datetime.now(timezone.utc).isoformat(), "encryption_algorithm": "SM4-CBC", "integrity_algorithm": "HMAC-SM3", "fields_encrypted": sensitive_fields[data_level] }} for key, value in record.items(): if key.startswith("_"): encrypted_record[key] = value continue if key in sensitive_fields[data_level]: encrypted_record[key] = self.encryption.encrypt_field(str(value), f"{data_level}:{key}") self._add_audit_entry("encrypt", key, data_level) else: encrypted_record[key] = value return encrypted_record def decrypt_record(self, encrypted_record: dict) -> dict: """ 解密完整的加密记录 Returns: 解密后的原始记录 """ metadata = encrypted_record.get("_metadata", {}) data_level = metadata.get("data_level", "一般") fields_to_decrypt = metadata.get("fields_encrypted", []) decrypted = {} for key, value in encrypted_record.items(): if key.startswith("_"): decrypted[key] = value continue if key in fields_to_decrypt and isinstance(value, dict) and "ciphertext" in value: decrypted[key] = self.encryption.decrypt_field(value, f"{data_level}:{key}") self._add_audit_entry("decrypt", key, data_level) else: decrypted[key] = value return decrypted def _add_audit_entry(self, action: str, field: str, level: str): """ 添加审计日志 依据办法第二十一条： - 安全事件处置、溯源相关日志留存不少于一年 - 向他人提供、委托处理相关日志留存不少于三年 """ entry = { "timestamp": datetime.now(timezone.utc).isoformat(), "action": action, "field": field, "data_level": level, "retention_years": 3 if action == "encrypt" else 1 } self.audit_log.append(entry) def rotate_keys(self, new_master_key: bytes): """ 密钥轮换 依据办法第二十五条：支持按周/月/季度自动更新密钥 Args: new_master_key: 新的主密钥（16 字节） """ old_key = self.encryption.master_key self.encryption = SM4FieldEncryption(new_master_key) self._add_audit_entry("key_rotate", "master_key", "all") print(f"[审计] 密钥已轮换：{datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC")

if __name__ == "__main__": # 主密钥：16 字节（生产环境应从 HSM 或密钥管理系统获取） MASTER_KEY = os.urandom(16) compliance = EnergyDataCompliance(MASTER_KEY) # 模拟能源行业数据：智能电表读数 meter_record = { "device_id": "SMART-METER-001", "station": "北京朝阳变电站", "phone": "13800138000", "email": "operator@example.com", "gps_location": "116.4074,39.9042", "meter_reading": 15234.7, "grid_topology": "220kV-双母线接线", "generation_capacity": "500MW", "timestamp": "2026-07-04T10:00:00Z" } print("=" * 60) print("能源行业数据安全合规实战演示") print("=" * 60) # 场景 1：加密重要数据 print("\n[场景 1] 加密重要数据（智能电表用户信息）") encrypted_important = compliance.classify_and_encrypt(meter_record, "重要") print(json.dumps(encrypted_important["_metadata"], indent=2, ensure_ascii=False)) print(f" phone: {encrypted_important['phone']['ciphertext'][:40]}...") print(f" gps_location: {encrypted_important['gps_location']['ciphertext'][:40]}...") # 场景 2：加密核心数据 print("\n[场景 2] 加密核心数据（电网拓扑与发电容量）") encrypted_core = compliance.classify_and_encrypt(meter_record, "核心") print(json.dumps(encrypted_core["_metadata"], indent=2, ensure_ascii=False)) print(f" grid_topology: {encrypted_core['grid_topology']['ciphertext'][:40]}...") print(f" generation_capacity: {encrypted_core['generation_capacity']['ciphertext'][:40]}...") # 场景 3：解密验证 print("\n[场景 3] 解密验证") decrypted = compliance.decrypt_record(encrypted_important) print(f" phone: {decrypted['phone']}") print(f" gps_location: {decrypted['gps_location']}") # 场景 4：密钥轮换 print("\n[场景 4] 密钥轮换（季度轮换策略）") new_key = os.urandom(16) compliance.rotate_keys(new_key) # 用新密钥解密（需要用新密钥重新加密的记录才能解密） new_encrypted = compliance.classify_and_encrypt(meter_record, "重要") decrypted_new = compliance.decrypt_record(new_encrypted) print(f" 新密钥解密 phone: {decrypted_new['phone']}") # 场景 5：篡改检测 print("\n[场景 5] 密文篡改检测（完整性保护）") tampered = dict(encrypted_important["phone"]) ct_bytes = base64.b64decode(tampered["ciphertext"]) tampered["ciphertext"] = base64.b64encode(ct_bytes[:-1] + bytes([(ct_bytes[-1] + 1) % 256])).decode("ascii") try: compliance.encryption.decrypt_field(tampered, "重要:phone") print(" ⚠️ 未检测到篡改（不应该发生）") except ValueError as e: print(f" ✅ 检测到篡改：{e}") # 审计日志 print(f"\n[审计日志] 共记录 {len(compliance.audit_log)} 条操作") for entry in compliance.audit_log[:5]: print(f" {entry['timestamp']} | {entry['action']:8s} | {entry['field']:20s} | 级别:{entry['data_level']}") print("\n" + "=" * 60) print("合规要点总结：") print(" 1. SM4-CBC 加密：满足办法第十六条加密要求") print(" 2. HMAC-SM3 校验：满足办法第十六条校验要求") print(" 3. 三级密钥体系：满足 GM/T 0028-2014 密码模块安全要求") print(" 4. 审计日志留存：满足办法第二十一条日志留存要求") print(" 5. 核心数据优先商用密码：满足办法第二十五条要求") print("=" * 60)