特征流水线的工程设计原则：安全视角下的防篡改实践

"""
安全特征流水线的Airflow DAG配置
包含防篡改机制、日志审计和监控
"""

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta
import hashlib
import hmac
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 安全配置
SECRET_KEY = "your-secret-key"  # 实际使用中应从环境变量或密钥管理系统获取
airflow_home = "/opt/airflow"

# 生成签名函数
def generate_signature(data):
    """生成HMAC签名"""
    return hmac.new(
        SECRET_KEY.encode('utf-8'),
        data.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

# 验证签名函数
def verify_signature(data, signature):
    """验证HMAC签名"""
    expected_signature = generate_signature(data)
    return hmac.compare_digest(expected_signature, signature)

# 数据采集任务
def data_collection(**kwargs):
    """数据采集任务，包含完整性验证"""
    logger.info("开始数据采集...")
    # 模拟数据采集
    raw_data = "raw_security_data"
    # 计算数据哈希
    data_hash = hashlib.sha256(raw_data.encode('utf-8')).hexdigest()
    logger.info(f"数据采集完成，哈希值：{data_hash}")
    # 将数据和哈希值传递给下一个任务
    kwargs['ti'].xcom_push(key='raw_data', value=raw_data)
    kwargs['ti'].xcom_push(key='data_hash', value=data_hash)

# 数据清洗任务
def data_cleaning(**kwargs):
    """数据清洗任务，包含防投毒检测"""
    logger.info("开始数据清洗...")
    # 获取上一个任务的数据
    raw_data = kwargs['ti'].xcom_pull(key='raw_data', task_ids='data_collection')
    data_hash = kwargs['ti'].xcom_pull(key='data_hash', task_ids='data_collection')
    
    # 验证数据完整性
    computed_hash = hashlib.sha256(raw_data.encode('utf-8')).hexdigest()
    if computed_hash != data_hash:
        raise ValueError("数据完整性验证失败，数据可能被篡改！")
    
    # 模拟数据清洗
    cleaned_data = raw_data + "_cleaned"
    # 计算清洗后数据的哈希
    cleaned_hash = hashlib.sha256(cleaned_data.encode('utf-8')).hexdigest()
    logger.info(f"数据清洗完成，哈希值：{cleaned_hash}")
    # 将清洗后的数据和哈希值传递给下一个任务
    kwargs['ti'].xcom_push(key='cleaned_data', value=cleaned_data)
    kwargs['ti'].xcom_push(key='cleaned_hash', value=cleaned_hash)

# 特征计算任务
def feature_computation(**kwargs):
    """特征计算任务，包含特征签名"""
    logger.info("开始特征计算...")
    # 获取上一个任务的数据
    cleaned_data = kwargs['ti'].xcom_pull(key='cleaned_data', task_ids='data_cleaning')
    cleaned_hash = kwargs['ti'].xcom_pull(key='cleaned_hash', task_ids='data_cleaning')
    
    # 验证数据完整性
    computed_hash = hashlib.sha256(cleaned_data.encode('utf-8')).hexdigest()
    if computed_hash != cleaned_hash:
        raise ValueError("数据完整性验证失败，数据可能被篡改！")
    
    # 模拟特征计算
    features = {"feature1": 1.0, "feature2": 2.0, "feature3": 3.0}
    features_str = str(features)
    # 生成特征签名
    features_signature = generate_signature(features_str)
    logger.info(f"特征计算完成，签名：{features_signature}")
    # 将特征和签名传递给下一个任务
    kwargs['ti'].xcom_push(key='features', value=features)
    kwargs['ti'].xcom_push(key='features_signature', value=features_signature)

# 特征存储任务
def feature_storage(**kwargs):
    """特征存储任务，包含加密存储"""
    logger.info("开始特征存储...")
    # 获取上一个任务的数据
    features = kwargs['ti'].xcom_pull(key='features', task_ids='feature_computation')
    features_signature = kwargs['ti'].xcom_pull(key='features_signature', task_ids='feature_computation')
    
    # 验证特征签名
    features_str = str(features)
    if not verify_signature(features_str, features_signature):
        raise ValueError("特征签名验证失败，特征可能被篡改！")
    
    # 模拟特征存储（实际应用中应使用加密存储）
    logger.info(f"特征存储完成，特征：{features}")

# 定义DAG默认参数
default_args = {
    'owner': 'security-team',
    'depends_on_past': False,
    'start_date': datetime(2026, 1, 9),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# 创建DAG
dag = DAG(
    'security_feature_pipeline',
    default_args=default_args,
    description='安全特征流水线DAG，包含防篡改机制',
    schedule_interval=timedelta(hours=1),
    catchup=False,
    tags=['security', 'feature-pipeline', 'airflow'],
)

# 定义任务
data_collection_task = PythonOperator(
    task_id='data_collection',
    python_callable=data_collection,
    provide_context=True,
    dag=dag,
)

data_cleaning_task = PythonOperator(
    task_id='data_cleaning',
    python_callable=data_cleaning,
    provide_context=True,
    dag=dag,
)

feature_computation_task = PythonOperator(
    task_id='feature_computation',
    python_callable=feature_computation,
    provide_context=True,
    dag=dag,
)

feature_storage_task = PythonOperator(
    task_id='feature_storage',
    python_callable=feature_storage,
    provide_context=True,
    dag=dag,
)

# 定义任务依赖
data_collection_task >> data_cleaning_task >> feature_computation_task >> feature_storage_task

#!/bin/bash
# Airflow安全配置脚本

# 设置Airflow家目录
export AIRFLOW_HOME="/opt/airflow"

# 1. 生成密码哈希
# 安装passlib库
pip install passlib[bcrypt]
# 生成密码哈希（替换your-password为实际密码）
python -c "from passlib.hash import bcrypt; print(bcrypt.hash('your-password'))"

# 2. 配置airflow.cfg
cat > $AIRFLOW_HOME/airflow.cfg << EOF
[core]
# 禁用示例DAG
dags_are_paused_at_creation = True
exload_examples = False

[webserver]
# 启用RBAC
rbac = True
# 启用SSL
web_server_ssl_cert = /path/to/ssl/cert.pem
web_server_ssl_key = /path/to/ssl/key.pem
# 禁用默认密码
authenticate = True
auth_backend = airflow.contrib.auth.backends.password_auth

[database]
# 配置安全的数据库连接
# 示例：postgresql://user:password@localhost:5432/airflow?sslmode=require
sql_alchemy_conn = your-secure-db-connection

[logging]
# 配置日志级别
logging_level = INFO
# 配置日志文件路径
base_log_folder = /opt/airflow/logs
# 启用日志轮换
log_rotation_age = 14
log_rotation_size = 100MB

[kubernetes]
# 启用Kubernetes执行器（可选）
executor = KubernetesExecutor
EOF

# 3. 初始化Airflow数据库
airflow db init

# 4. 创建管理员用户
# 替换admin、admin@example.com和your-password哈希为实际值
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com \
    --password 2b12xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

# 5. 启动Airflow服务（后台运行）
airflow webserver -p 8080 -D
airflow scheduler -D

"""
特征流水线篡改检测脚本
使用机器学习算法实时检测特征流水线中的异常行为
"""

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import logging
import time

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class FeaturePipelineTamperDetector:
    """特征流水线篡改检测器"""
    
    def __init__(self, contamination=0.1):
        """初始化检测器
        
        Args:
            contamination: 异常样本比例
        """
        self.scaler = StandardScaler()
        self.model = IsolationForest(contamination=contamination, random_state=42)
        self.is_trained = False
        logger.info(f"初始化篡改检测器，异常比例：{contamination}")
    
    def fit(self, normal_data):
        """使用正常数据训练模型
        
        Args:
            normal_data: 正常数据，DataFrame格式
        """
        logger.info(f"开始训练模型，样本数：{len(normal_data)}")
        # 特征缩放
        scaled_data = self.scaler.fit_transform(normal_data)
        # 训练模型
        self.model.fit(scaled_data)
        self.is_trained = True
        logger.info("模型训练完成")
    
    def detect(self, current_data):
        """检测当前数据是否异常
        
        Args:
            current_data: 当前数据，DataFrame格式
            
        Returns:
            检测结果，-1表示异常，1表示正常
        """
        if not self.is_trained:
            raise ValueError("模型未训练，无法进行检测")
        
        logger.info(f"开始检测，样本数：{len(current_data)}")
        # 特征缩放
        scaled_data = self.scaler.transform(current_data)
        # 进行检测
        predictions = self.model.predict(scaled_data)
        # 统计异常数量
        anomaly_count = np.sum(predictions == -1)
        logger.info(f"检测完成，异常样本数：{anomaly_count}/{len(current_data)}")
        
        return predictions
    
    def get_anomaly_score(self, current_data):
        """获取异常分数
        
        Args:
            current_data: 当前数据，DataFrame格式
            
        Returns:
            异常分数，分数越低越异常
        """
        if not self.is_trained:
            raise ValueError("模型未训练，无法计算异常分数")
        
        # 特征缩放
        scaled_data = self.scaler.transform(current_data)
        # 计算异常分数
        scores = self.model.decision_function(scaled_data)
        return scores

# 示例用法
if __name__ == "__main__":
    # 1. 生成正常训练数据
    np.random.seed(42)
    normal_data = pd.DataFrame({
        'feature_1': np.random.normal(0, 1, 1000),
        'feature_2': np.random.normal(0, 1, 1000),
        'processing_time': np.random.normal(100, 10, 1000)  # 处理时间
    })
    
    # 2. 初始化并训练检测器
    detector = FeaturePipelineTamperDetector(contamination=0.05)
    detector.fit(normal_data)
    
    # 3. 生成测试数据（包含正常和异常数据）
    # 正常数据
    normal_test = pd.DataFrame({
        'feature_1': np.random.normal(0, 1, 100),
        'feature_2': np.random.normal(0, 1, 100),
        'processing_time': np.random.normal(100, 10, 100)
    })
    
    # 异常数据（处理时间异常）
    anomaly_test = pd.DataFrame({
        'feature_1': np.random.normal(0, 1, 20),
        'feature_2': np.random.normal(0, 1, 20),
        'processing_time': np.random.normal(500, 50, 20)  # 处理时间异常长
    })
    
    # 合并测试数据
    test_data = pd.concat([normal_test, anomaly_test], ignore_index=True)
    
    # 4. 进行检测
    predictions = detector.detect(test_data)
    scores = detector.get_anomaly_score(test_data)
    
    # 5. 输出检测结果
    results = pd.DataFrame({
        'prediction': predictions,
        'anomaly_score': scores
    })
    print("检测结果统计：")
    print(results['prediction'].value_counts())
    print("\n异常分数统计：")
    print(results['anomaly_score'].describe())

"""
安全特征服务API，包含身份认证和访问控制
"""

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import uvicorn
import hashlib
import hmac
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 安全配置
SECRET_KEY = "your-secret-key"  # 实际使用中应从环境变量或密钥管理系统获取
API_KEYS = {
    "model-1": "api-key-1",
    "model-2": "api-key-2"
}  # 实际使用中应存储在安全的地方，如数据库或密钥管理系统

# 创建FastAPI应用
app = FastAPI(
    title="安全特征服务API",
    description="提供安全的特征查询服务，包含身份认证和访问控制",
    version="1.0.0"
)

# 定义安全方案
security = HTTPBearer()

# 定义请求模型
class FeatureRequest(BaseModel):
    entity_id: str
    feature_names: list[str]

# 定义响应模型
class FeatureResponse(BaseModel):
    entity_id: str
    features: dict
    timestamp: float
    signature: str

# 验证API密钥
async def verify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """验证API密钥"""
    api_key = credentials.credentials
    if api_key not in API_KEYS.values():
        logger.warning(f"无效的API密钥：{api_key}")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的API密钥",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 查找对应的模型ID
    for model_id, key in API_KEYS.items():
        if key == api_key:
            return model_id
    return None

# 生成特征签名
def generate_feature_signature(entity_id: str, features: dict, timestamp: float) -> str:
    """生成特征签名"""
    data = f"{entity_id}{features}{timestamp}"
    return hmac.new(
        SECRET_KEY.encode('utf-8'),
        data.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

# 模拟特征存储
feature_store = {
    "user-1": {
        "feature1": 1.0,
        "feature2": 2.0,
        "feature3": 3.0
    },
    "user-2": {
        "feature1": 4.0,
        "feature2": 5.0,
        "feature3": 6.0
    }
}

# 定义API端点
@app.post("/features", response_model=FeatureResponse)
async def get_features(
    request: FeatureRequest,
    model_id: str = Depends(verify_api_key)
):
    """获取特征，包含身份认证和签名验证"""
    logger.info(f"收到特征请求，实体ID：{request.entity_id}，模型ID：{model_id}")
    
    # 检查实体是否存在
    if request.entity_id not in feature_store:
        logger.warning(f"实体不存在：{request.entity_id}")
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"实体不存在：{request.entity_id}"
        )
    
    # 获取请求的特征
    entity_features = feature_store[request.entity_id]
    requested_features = {}
    for feature_name in request.feature_names:
        if feature_name in entity_features:
            requested_features[feature_name] = entity_features[feature_name]
        else:
            logger.warning(f"特征不存在：{feature_name}")
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"特征不存在：{feature_name}"
            )
    
    # 生成时间戳
    import time
    timestamp = time.time()
    
    # 生成签名
    signature = generate_feature_signature(
        request.entity_id,
        requested_features,
        timestamp
    )
    
    # 记录日志
    logger.info(f"特征请求处理完成，实体ID：{request.entity_id}，特征数量：{len(requested_features)}")
    
    # 返回响应
    return FeatureResponse(
        entity_id=request.entity_id,
        features=requested_features,
        timestamp=timestamp,
        signature=signature
    )

# 健康检查端点
@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "ok", "message": "特征服务运行正常"}

if __name__ == "__main__":
    # 启动服务
    uvicorn.run(
        "feature_service:app",
        host="0.0.0.0",
        port=8000,
        ssl_keyfile="/path/to/ssl/key.pem",  # 实际使用中应配置SSL
        ssl_certfile="/path/to/ssl/cert.pem",
        reload=True
    )

# 安装必要的库
pip install apache-airflow pandas scikit-learn fastapi uvicorn passlib[bcrypt] python-multipart

# 初始化Airflow数据库
airflow db init

# 启动Airflow服务
airflow webserver -p 8080 -D
airflow scheduler -D

# 启动特征服务API
python feature_service.py