面向供应链安全的Lakehouse架构：数据隔离与可信计算实践

supply-chain-lakehouse-sec/
│
├── config/
│   ├── __init__.py
│   └── settings.py           # 项目配置：路径、分区键、供应商列表等
│
├── data_generator/
│   ├── __init__.py
│   └── simulate.py           # 模拟生成供应链交易数据
│
├── data_isolation/
│   ├── __init__.py
│   ├── writer.py             # 将数据按分区写入Delta Lake
│   └── reader.py             # 提供安全的视图读取接口
│
├── tee/
│   ├── __init__.py
│   ├── enclave_simulator.py  # SGX Enclave模拟核心
│   └── crypto_utils.py       # 非对称加密辅助函数（模拟）
│
├── compute/
│   ├── __init__.py
│   └── trusted_aggregator.py # 可信计算协调器与主逻辑
│
├── main.py                   # 主程序入口：运行完整流程
├── requirements.txt          # 项目依赖
└── run.sh                    # 一键运行脚本（可选）

# 项目核心配置
import os
from datetime import datetime

# 路径配置
BASE_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DATA_RAW_PATH = os.path.join(BASE_PATH, "data/raw")  # 模拟原始数据输出路径
DELTA_TABLE_PATH = os.path.join(BASE_PATH, "data/delta/supply_chain_facts")  # Delta表路径
RESULTS_PATH = os.path.join(BASE_PATH, "data/results")

# Lakehouse 表配置
TABLE_NAME = "supply_chain_facts"
PARTITION_COLS = ["supplier_id", "date"]  # 分区列
ZORDER_COL = "part_id"                     # Z-Ordering优化列

# 供应链模拟配置
SUPPLIERS = ["Supplier_A", "Supplier_B", "Supplier_C"]
PARTS = ["P-1001", "P-1002", "P-1003", "P-1004"]
DATE_RANGE = ("2024-01-01", "2024-03-31")  # Q1 数据

# 可信计算配置 (模拟)
# 在实际SGX中，密钥对在enclave内生成，此处为演示生成两对RSA密钥。
# key_pair_1: 用于加密输入数据 (公钥公开，私钥仅enclave知晓)
# key_pair_2: 用于加密输出结果 (公钥在enclave内，私钥由请求方持有)
TEE_KEY_INPUT_PUBLIC_PATH = os.path.join(BASE_PATH, "tee/keys/input_public.pem")
TEE_KEY_INPUT_PRIVATE_PATH = os.path.join(BASE_PATH, "tee/keys/input_private.pem") # 保密，仅模拟器加载
TEE_KEY_OUTPUT_PUBLIC_PATH = os.path.join(BASE_PATH, "tee/keys/output_public.pem") # 在enclave内
TEE_KEY_OUTPUT_PRIVATE_PATH = os.path.join(BASE_PATH, "tee/keys/output_private.pem")

# Spark配置 (本地模式)
SPARK_CONFIG = {
    "spark.master": "local[*]",
    "spark.databricks.delta.retentionDurationCheck.enabled": "false",
    "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    "spark.driver.memory": "4g"
}

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import os
from config.settings import SUPPLIERS, PARTS, DATE_RANGE, DATA_RAW_PATH

def generate_transactions_for_supplier(supplier_id, num_records=1000):
    """为单个供应商生成模拟交易数据"""
    np.random.seed(hash(supplier_id) % 10000)  # 确保可重现性

    base_date = datetime.strptime(DATE_RANGE[0], "%Y-%m-%d")
    date_list = [base_date + timedelta(days=np.random.randint(0, 90)) for _ in range(num_records)]

    data = {
        "transaction_id": [f"TX-{supplier_id}-{i:06d}" for i in range(num_records)],
        "date": [d.strftime("%Y-%m-%d") for d in date_list],
        "supplier_id": [supplier_id] * num_records,
        "part_id": np.random.choice(PARTS, num_records, p=[0.4, 0.3, 0.2, 0.1]),
        "quantity": np.random.randint(10, 500, num_records),
        "unit_price": np.round(np.random.uniform(50.0, 500.0, num_records), 2),  # 敏感字段
        "total_amount": None,  # 由quantity*unit_price计算
        "warehouse_id": np.random.choice(["WH-1", "WH-2", "WH-3"], num_records),
        "po_number": [f"PO-{supplier_id}-{np.random.randint(1000,9999)}" for _ in range(num_records)]
    }
    df = pd.DataFrame(data)
    df["total_amount"] = df["quantity"] * df["unit_price"]
    # 添加细微的供应商特定模式
    if "A" in supplier_id:
        df["unit_price"] = df["unit_price"] * np.random.uniform(0.95, 1.05, num_records)
    elif "B" in supplier_id:
        df["quantity"] = df["quantity"] * np.random.randint(1, 3, num_records)
    return df

def generate_all_data():
    """为所有供应商生成数据并保存为Parquet格式（模拟数据源）"""
    os.makedirs(DATA_RAW_PATH, exist_ok=True)
    all_dfs = []
    for supplier in SUPPLIERS:
        print(f"Generating data for {supplier}...")
        df = generate_transactions_for_supplier(supplier, 1200)
        file_path = os.path.join(DATA_RAW_PATH, f"{supplier}_transactions.parquet")
        df.to_parquet(file_path, index=False)
        all_dfs.append(df)
        print(f"  Saved to {file_path}")

    # 也可合并一个全局视图（仅用于验证）
    full_df = pd.concat(all_dfs, ignore_index=True)
    full_path = os.path.join(DATA_RAW_PATH, "all_transactions.parquet")
    full_df.to_parquet(full_path, index=False)
    print(f"\nFull dataset saved to {full_path}")
    return full_df

if __name__ == "__main__":
    generate_all_data()

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
import os
from config.settings import SPARK_CONFIG, DELTA_TABLE_PATH, PARTITION_COLS, ZORDER_COL, DATA_RAW_PATH

def get_spark_session(app_name="SupplyChainLakehouse"):
    """创建并返回配置好的Spark会话"""
    builder = SparkSession.builder.appName(app_name)
    for key, value in SPARK_CONFIG.items():
        builder = builder.config(key, value)
    spark = builder.getOrCreate()
    return spark

def write_to_delta_with_isolation():
    """
    核心写入逻辑：读取各供应商的Parquet文件，按分区写入Delta表。
    模拟对‘unit_price'进行加密（使用TEE公钥），实际生产应与TEE模块集成。
    """
    spark = get_spark_session("DataIngestion")
    print("开始数据写入与隔离...")

    # 读取所有供应商的原始数据（模拟从不同源头摄入）
    supplier_files = [f for f in os.listdir(DATA_RAW_PATH) if f.endswith('.parquet') and 'all' not in f]
    df_list = []
    for file in supplier_files:
        supplier = file.replace('_transactions.parquet', '')
        file_path = os.path.join(DATA_RAW_PATH, file)
        df = spark.read.parquet(file_path)
        # 此处为模拟：在实际中，应在数据进入Lakehouse前，由数据所有者用TEE公钥加密敏感字段。
        # 我们这里添加一个标记列，表示该字段‘应被加密'。
        df = df.withColumn("_is_sensitive", col("unit_price").isNotNull())
        df_list.append(df)

    # 合并并写入Delta表，自动按PARTITION_COLS分区
    if df_list:
        full_df = df_list[0]
        for other_df in df_list[1:]:
            full_df = full_df.union(other_df)

        print(f"写入Delta表路径: {DELTA_TABLE_PATH}")
        print(f"分区列: {PARTITION_COLS}")
        full_df.write \
            .format("delta") \
            .mode("overwrite") \
            .partitionBy(*PARTITION_COLS) \
            .save(DELTA_TABLE_PATH)

        # 对常用查询列进行Z-Ordering优化
        print(f"对列 '{ZORDER_COL}' 进行Z-Ordering优化...")
        spark.sql(f"""
            OPTIMIZE delta.`{DELTA_TABLE_PATH}`
            ZORDER BY ({ZORDER_COL})
        """).show()

        # 创建表引用（方便后续SQL查询）
        spark.sql(f"""
            CREATE TABLE IF NOT EXISTS {TABLE_NAME}
            USING DELTA
            LOCATION '{DELTA_TABLE_PATH}'
        """)
        print("数据写入与优化完成。")
        spark.stop()
    else:
        print("未找到源数据文件。")
        spark.stop()

if __name__ == "__main__":
    write_to_delta_with_isolation()

# 模拟加密工具。在生产SGX环境中，应使用enclave内的加密库（如Mbed TLS）。
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes
import base64

def generate_rsa_key_pair():
    """生成RSA密钥对（模拟）"""
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    public_key = private_key.public_key()
    return private_key, public_key

def serialize_public_key(public_key):
    """序列化公钥为PEM格式字符串"""
    pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    return pem.decode('utf-8')

def encrypt_with_public_key(plaintext_bytes, public_key_pem):
    """使用公钥PEM字符串加密数据（模拟数据所有者加密敏感字段）"""
    public_key = serialization.load_pem_public_key(public_key_pem.encode('utf-8'))
    ciphertext = public_key.encrypt(
        plaintext_bytes,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return base64.b64encode(ciphertext).decode('utf-8')

def decrypt_with_private_key(ciphertext_b64, private_key):
    """使用私钥解密数据（模拟在enclave内解密）"""
    ciphertext = base64.b64decode(ciphertext_b64)
    plaintext = private_key.decrypt(
        ciphertext,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )
    return plaintext

# 注意：此模拟代码未在真正隔离环境中运行。真实SGX应确保私钥永不离开enclave。

"""
SGX Enclave 模拟器。
在生产环境中，此部分代码应使用Intel SGX SDK编译为受信任的Enclave代码。
此处我们用一个Python类模拟其安全边界和行为。
"""
import pandas as pd
import json
from tee.crypto_utils import decrypt_with_private_key, encrypt_with_public_key, serialize_public_key

class SGXEnclaveSimulator:
    """
    模拟一个SGX Enclave。
    初始化时加载‘密封'在内部的私钥（用于解密输入数据）。
    内部拥有另一对密钥，用于加密输出结果。
    """
    def __init__(self, input_private_key, output_public_key_pem):
        # 这些"秘密"在真实Enclave中是在其初始化时生成的或密封进来的，外部不可读。
        self._input_private_key = input_private_key  # 用于解密传入的敏感数据
        self._output_public_key_pem = output_public_key_pem  # 用于加密传出结果

    def _decrypt_sensitive_column(self, encrypted_series):
        """模拟在Enclave内解密一个Pandas Series（假设每个值都是base64加密字符串）"""
        # 注意：真实场景中，数据是逐个记录或批量解密，且格式更复杂。
        decrypted_values = []
        for enc_str in encrypted_series:
            if pd.isna(enc_str):
                decrypted_values.append(None)
            else:
                # 假设加密的是float的字节表示
                plain_bytes = decrypt_with_private_key(enc_str, self._input_private_key)
                value = float(plain_bytes.decode('utf-8'))
                decrypted_values.append(value)
        return pd.Series(decrypted_values, index=encrypted_series.index)

    def compute_secure_aggregation(self, encrypted_data_json, computation_spec):
        """
        在模拟的Enclave内执行安全聚合。
        :param encrypted_data_json: JSON字符串，包含加密的DataFrame（列‘encrypted_unit_price'）
        :param computation_spec: 计算规格，如 {"operation": "mean", "group_by": ["part_id", "quarter"]}
        :return: 加密后的结果（JSON字符串的base64加密形式）
        """
        # 1. 在Enclave边界内解密数据
        df_dict = json.loads(encrypted_data_json)
        df = pd.DataFrame(df_dict)
        print(f"[Enclave Sim] 收到 {len(df)} 条加密记录。")

        if 'encrypted_unit_price' not in df.columns:
            raise ValueError("输入数据缺失加密列 'encrypted_unit_price'")

        df['unit_price'] = self._decrypt_sensitive_column(df['encrypted_unit_price'])
        df['date'] = pd.to_datetime(df['date'])
        df['quarter'] = df['date'].dt.quarter

        # 2. 执行指定的安全计算（此处逻辑是预定义的，不可被外部篡改）
        if computation_spec.get('operation') == 'mean' and computation_spec.get('group_by') == ['part_id', 'quarter']:
            # 计算每个零件在每个季度的平均价格
            result_df = df.groupby(['part_id', 'quarter'])['unit_price'].mean().reset_index()
            result_df.rename(columns={'unit_price': 'avg_unit_price'}, inplace=True)
            result_json = result_df.to_json(orient='records', date_format='iso')
        else:
            raise NotImplementedError(f"不支持的计算规格: {computation_spec}")

        print(f"[Enclave Sim] 计算完成，明文结果片段: {result_df.head().to_string()}")

        # 3. 使用Enclave内部的输出公钥加密结果（确保只有持有对应私钥的请求方能解密）
        encrypted_result = encrypt_with_public_key(result_json.encode('utf-8'), self._output_public_key_pem)
        print("[Enclave Sim] 结果已加密并传出Enclave。")
        return encrypted_result

"""
可信计算协调器。
负责从Delta Lake准备数据，与模拟的SGX Enclave交互，并处理最终结果。
"""
from pyspark.sql import SparkSession, functions as F
from config.settings import (
    get_spark_session, DELTA_TABLE_PATH, SUPPLIERS,
    TEE_KEY_INPUT_PUBLIC_PATH, TEE_KEY_INPUT_PRIVATE_PATH,
    TEE_KEY_OUTPUT_PUBLIC_PATH, TEE_KEY_OUTPUT_PRIVATE_PATH
)
from tee.enclave_simulator import SGXEnclaveSimulator
from tee.crypto_utils import (
    generate_rsa_key_pair, serialize_public_key,
    encrypt_with_public_key, decrypt_with_private_key
)
import json
import pandas as pd
import os

class TrustedAggregator:
    def __init__(self):
        self.spark = get_spark_session("TrustedAggregation")
        # 加载或生成密钥（模拟）
        # 密钥管理在实际系统中非常关键，此处极度简化。
        self._load_or_generate_keys()
        # 初始化模拟Enclave，传入其所需的私钥和公钥
        self.enclave = SGXEnclaveSimulator(self.input_private_key, self.output_public_key_pem)

    def _load_or_generate_keys(self):
        """模拟密钥的生成与加载过程"""
        keys_dir = os.path.dirname(TEE_KEY_INPUT_PUBLIC_PATH)
        os.makedirs(keys_dir, exist_ok=True)

        # 为演示，如果密钥文件不存在则生成
        if not os.path.exists(TEE_KEY_INPUT_PUBLIC_PATH):
            print("生成模拟RSA密钥对...")
            self.input_private_key, input_public_key = generate_rsa_key_pair()
            self.output_private_key, output_public_key = generate_rsa_key_pair()

            # 保存密钥（模拟：input_private应安全分发给enclave，output_private由请求方保管）
            with open(TEE_KEY_INPUT_PUBLIC_PATH, 'w') as f:
                f.write(serialize_public_key(input_public_key))
            with open(TEE_KEY_INPUT_PRIVATE_PATH, 'wb') as f:
                f.write(self.input_private_key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.TraditionalOpenSSL,
                    encryption_algorithm=serialization.NoEncryption()
                ))
            with open(TEE_KEY_OUTPUT_PUBLIC_PATH, 'w') as f:
                f.write(serialize_public_key(output_public_key))
            self.output_public_key_pem = serialize_public_key(output_public_key)
            with open(TEE_KEY_OUTPUT_PRIVATE_PATH, 'wb') as f:
                f.write(self.output_private_key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.TraditionalOpenSSL,
                    encryption_algorithm=serialization.NoEncryption()
                ))
        else:
            # 加载已存在的密钥
            from cryptography.hazmat.primitives import serialization
            with open(TEE_KEY_INPUT_PRIVATE_PATH, 'rb') as f:
                self.input_private_key = serialization.load_pem_private_key(f.read(), password=None)
            with open(TEE_KEY_OUTPUT_PUBLIC_PATH, 'r') as f:
                self.output_public_key_pem = f.read()
            with open(TEE_KEY_OUTPUT_PRIVATE_PATH, 'rb') as f:
                self.output_private_key = serialization.load_pem_private_key(f.read(), password=None)

    def prepare_and_encrypt_data(self, suppliers, start_date, end_date):
        """
        从Delta Lake读取指定供应商和日期范围的数据，
        并使用TEE输入公钥加密敏感字段（模拟，实际应在数据源头加密）。
        """
        print(f"准备数据: suppliers={suppliers}, date_range=[{start_date}, {end_date}]")
        df = self.spark.read.format("delta").load(DELTA_TABLE_PATH)
        df = df.filter(
            (F.col("supplier_id").isin(suppliers)) &
            (F.col("date").between(start_date, end_date))
        ).select("date", "supplier_id", "part_id", "unit_price") # 只选择必要列

        # 转换为Pandas DataFrame以便与模拟Enclave交互（生产环境可能使用专用序列化）
        pd_df = df.toPandas()
        print(f"从Delta Lake读取了 {len(pd_df)} 条记录。")

        # **模拟加密过程**：在实际中，‘unit_price'在进入Lakehouse前已由数据提供方加密。
        # 这里我们读取后，用input公钥"现场"加密来模拟。
        with open(TEE_KEY_INPUT_PUBLIC_PATH, 'r') as f:
            input_pub_key_pem = f.read()

        def encrypt_price(x):
            # 将浮点数转为字节并加密
            return encrypt_with_public_key(str(x).encode('utf-8'), input_pub_key_pem)

        pd_df['encrypted_unit_price'] = pd_df['unit_price'].apply(encrypt_price)
        # 丢弃明文列，模拟只有密文进入计算流程
        pd_df_for_tee = pd_df.drop(columns=['unit_price'])

        return pd_df_for_tee

    def run_trusted_computation(self, encrypted_pd_df):
        """协调可信计算流程"""
        # 1. 将数据序列化为JSON传入模拟Enclave
        input_json = encrypted_pd_df.to_json(orient='records', date_format='iso')
        # 2. 定义计算任务
        computation_spec = {
            "operation": "mean",
            "group_by": ["part_id", "quarter"]  # quarter将在enclave内从date派生
        }
        print("启动模拟SGX Enclave执行安全聚合计算...")
        # 3. 调用Enclave
        encrypted_result = self.enclave.compute_secure_aggregation(input_json, computation_spec)
        # 4. 在协调器端（外部）使用自己的私钥解密结果
        result_json_bytes = decrypt_with_private_key(encrypted_result, self.output_private_key)
        final_result = json.loads(result_json_bytes.decode('utf-8'))
        print("协调器解密最终结果成功。")
        return final_result

    def execute_example_query(self):
        """执行一个示例查询：计算所有供应商在2024年第一季度各零件的平均价格"""
        suppliers = SUPPLIERS  # 所有供应商
        start_date, end_date = "2024-01-01", "2024-03-31"

        encrypted_data = self.prepare_and_encrypt_data(suppliers, start_date, end_date)
        if encrypted_data.empty:
            print("没有找到符合条件的数据。")
            return

        result = self.run_trusted_computation(encrypted_data)
        result_df = pd.DataFrame(result)
        print("\n" + "="*60)
        print("可信计算最终结果 (2024年Q1 各零件平均单价):")
        print("="*60)
        print(result_df.to_string(index=False))
        print("="*60)
        # 保存结果
        os.makedirs(os.path.dirname(RESULTS_PATH), exist_ok=True)
        result_path = os.path.join(RESULTS_PATH, "trusted_aggregation_q1.json")
        result_df.to_json(result_path, orient='records', date_format='iso')
        print(f"结果已保存至: {result_path}")
        return result_df

"""
主程序：编排整个供应链安全Lakehouse数据流水线。
"""
from data_generator.simulate import generate_all_data
from data_isolation.writer import write_to_delta_with_isolation
from compute.trusted_aggregator import TrustedAggregator
import time

def main():
    print("="*70)
    print("面向供应链安全的Lakehouse架构：数据隔离与可信计算实践")
    print("="*70)

    # 步骤1：生成模拟数据
    print("\n[阶段 1] 生成供应链模拟数据...")
    start = time.time()
    generate_all_data()
    print(f"  耗时: {time.time()-start:.2f} 秒")

    # 步骤2：将数据按隔离策略写入Delta Lake
    print("\n[阶段 2] 数据分区隔离与Lakehouse入库...")
    start = time.time()
    write_to_delta_with_isolation()
    print(f"  耗时: {time.time()-start:.2f} 秒")

    # 步骤3：执行可信聚合计算
    print("\n[阶段 3] 启动可信执行环境进行安全聚合计算...")
    start = time.time()
    aggregator = TrustedAggregator()
    aggregator.execute_example_query()
    print(f"  总耗时: {time.time()-start:.2f} 秒")

    print("\n" + "="*70)
    print("演示流程结束。")
    print("="*70)

if __name__ == "__main__":
    main()

# 创建虚拟环境 (可选)
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

# 安装依赖
pip install -r requirements.txt

pyspark>=3.3.0
delta-spark>=2.3.0
pandas>=1.5.0
cryptography>=39.0.0
pyarrow>=10.0.0

python main.py

import sys
sys.path.insert(0, '.')
from data_generator.simulate import generate_transactions_for_supplier
from tee.crypto_utils import generate_rsa_key_pair, encrypt_with_public_key, decrypt_with_private_key, serialize_public_key

def test_data_generation():
    df = generate_transactions_for_supplier("Supplier_Test", 10)
    assert len(df) == 10
    assert 'unit_price' in df.columns
    assert df['supplier_id'].unique()[0] == 'Supplier_Test'
    print("✅ 数据生成测试通过。")
    return df

def test_crypto_utils():
    priv_key, pub_key = generate_rsa_key_pair()
    pub_key_pem = serialize_public_key(pub_key)
    original_text = b"Secret Supply Chain Price: 123.45"
    encrypted = encrypt_with_public_key(original_text, pub_key_pem)
    decrypted = decrypt_with_private_key(encrypted, priv_key)
    assert decrypted == original_text
    print("✅ 加密/解密测试通过。")

if __name__ == "__main__":
    test_data_generation()
    test_crypto_utils()
    print("\n所有基础测试通过。")

python test_basic.py