tools/ia_utils/adversarial_vector_attack.py

"""

Gradient-based adversarial stress-testing for enhanced_dense_healing_hybrid's

Phi-Trigger decision -- a targeted, crafted perturbation instead of the

random NaN/Inf corruption this healing pipeline is normally tested

against, adapted from IGME's chained-differentiable-attack idea

(arXiv:2607.27465, "IGME: Efficient Chained Method Ensemble for

Transferable Semantic Segmentation Attacks") applied to vector sequences

instead of image segmentation.



evaluate_phi_trigger (dense_evolution.healing) makes its keep-vs-median-

replace decision by thresholding |v_dinamic| against NON_STATIC_THRESHOLD_A

-- a hard step, not differentiable at the boundary. But v_dinamic itself

(via calculate_phi_ab -> calculate_vettore_dinamico) is built entirely

from norms, dot products, log, and clip -- all JAX-differentiable. This

crafts a minimal perturbation to a single vector in the sequence, via

gradient ascent/descent on |v_dinamic| (projected back into an epsilon

L2-ball each step, the standard PGD pattern), that flips the trigger's

decision at that point:



- "flip_to_dynamic": push an originally-static point (would be replaced

  by the local median) across the threshold so the trigger keeps it

  as-is instead -- the more security-relevant direction, since it

  represents a worst-case corruption crafted to *evade* the healer by

  looking like genuine motion, rather than obvious noise.

- "flip_to_static": push an originally-dynamic point (would be kept)

  across the threshold so the trigger discards it as noise instead --

  the failure mode of genuine signal getting wrongly suppressed.



This does not attack the median-filter fallback itself, only the

Phi-Trigger's keep-vs-replace decision -- see craft_adversarial_healing_

perturbation's own docstring for what "success" means precisely.

"""

from typing import Optional

import numpy as np
import jax
import jax.numpy as jnp

from dense_evolution.mitigation.healing import calculate_phi_ab, calculate_vettore_dinamico, GLOBAL_CONSTANTS

__all__ = ['craft_adversarial_healing_perturbation']


def _trigger_magnitude(state_B, state_A, ipg_vector):
    """|v_dinamic| -- the continuous quantity evaluate_phi_trigger

    thresholds. Differentiable w.r.t. state_B (the vector being attacked);

    state_A/ipg_vector are treated as fixed context, not attacked."""
    phi_ab = calculate_phi_ab(state_A, state_B, ipg_vector)
    E_A = jnp.linalg.norm(state_A)
    E_B = jnp.linalg.norm(state_B)
    v_dinamic = calculate_vettore_dinamico(E_A, E_B, phi_ab)
    return jnp.abs(v_dinamic)


_trigger_magnitude_grad = jax.grad(_trigger_magnitude, argnums=0)


def craft_adversarial_healing_perturbation(

    vettori: np.ndarray,

    target_idx: int,

    radius_baseline: Optional[int] = None,

    epsilon: float = 0.1,

    n_steps: int = 50,

    step_size: Optional[float] = None,

    direction: str = "flip_to_dynamic",

) -> dict:
    """Crafts a minimal adversarial perturbation to vettori[target_idx],

    within an L2 epsilon-ball, that flips enhanced_dense_healing_hybrid's

    Phi-Trigger decision at that index -- a targeted stress test, not

    random noise.



    Reproduces enhanced_dense_healing_hybrid's own per-step computation

    at target_idx exactly (same baseline_mean window, same adaptive

    radius default, same inter-point-gradient vector) so the crafted

    perturbation is faithful to what the real healing pipeline would

    actually see, not a simplified stand-in.



    BUG FIX: step_size used to default to 2*epsilon/n_steps -- tying the

    per-step move size to the epsilon budget. Verified directly this

    makes LARGER epsilon give WORSE (higher final |v_dinamic|) results,

    not better: a bigger budget means a bigger step, which overshoots

    and oscillates around the minimum instead of converging to it,

    which is the opposite of what a bigger budget should ever do for a

    correct optimizer. step_size is now independent of epsilon (a small

    fixed default, tuned to v_dinamic's typical local scale) -- epsilon

    only bounds where the iterate is allowed to end up (via projection

    after each step), not how big each step is.



    Args:

        vettori: array-like, shape (n_steps, dim), the (unperturbed)

            vector sequence. Not modified in place.

        target_idx: index to attack; must be >= 2 (the healing loop's

            own starting point) and < len(vettori).

        radius_baseline: same meaning as enhanced_dense_healing_hybrid's

            own parameter; None uses the same adaptive default.

        epsilon: L2-norm budget for the perturbation (the attack is

            projected back into this ball after every gradient step).

        n_steps: number of PGD-style gradient steps.

        step_size: per-step move size along the normalized gradient;

            None uses a small fixed default (0.02) independent of

            epsilon -- see the bug-fix note above for why.

        direction: "flip_to_dynamic" (evade -- make static-looking

            input pass through unhealed) or "flip_to_static" (suppress

            -- make dynamic-looking input get median-replaced instead).



    Returns:

        dict with:

            perturbed_vettori: copy of vettori with vettori[target_idx]

                replaced by the crafted perturbation.

            success: bool, True only if the trigger decision actually

                flipped in the requested direction (a small epsilon or

                too few steps can fail to cross the threshold).

            original_trigger_active / final_trigger_active: bool, the

                Phi-Trigger's decision (True = dynamic/kept) before and

                after the perturbation.

            original_magnitude / final_magnitude: float, |v_dinamic|

                before and after.

            perturbation_norm: float, actual L2 norm of the applied

                perturbation (<= epsilon).

    """
    vettori = np.asarray(vettori, dtype=np.float64)
    n, hidden_dim = vettori.shape
    if target_idx < 2 or target_idx >= n:
        raise ValueError(f"target_idx must be in [2, {n - 1}], got {target_idx} (n={n})")
    if direction not in ("flip_to_dynamic", "flip_to_static"):
        raise ValueError(f"direction must be 'flip_to_dynamic' or 'flip_to_static', got {direction!r}")
    if epsilon <= 0:
        raise ValueError(f"epsilon must be positive, got {epsilon}")

    if radius_baseline is None:
        radius_baseline = min(20, max(3, n // 3))
    lo = max(0, target_idx - radius_baseline)

    state_A = jnp.array(np.mean(vettori[lo:target_idx], axis=0))
    ipg_raw = vettori[target_idx - 1] - vettori[target_idx - 2]
    norm_ipg_raw = np.linalg.norm(ipg_raw)
    ipg_vector = jnp.array(ipg_raw / norm_ipg_raw) if norm_ipg_raw > 1e-9 else jnp.array(ipg_raw)

    original_state_B = jnp.array(vettori[target_idx])
    threshold = GLOBAL_CONSTANTS['NON_STATIC_THRESHOLD_A']
    original_magnitude = float(_trigger_magnitude(original_state_B, state_A, ipg_vector))
    original_trigger_active = original_magnitude > threshold

    sign = 1.0 if direction == "flip_to_dynamic" else -1.0
    if step_size is None:
        step_size = 0.02

    # Track the best iterate seen (most favorable to `direction`), not
    # just the last one -- with a fixed small step size the trajectory
    # can overshoot past the optimum and drift back worse on later
    # steps, especially once it's pinned against the epsilon-ball
    # boundary; keeping the best-so-far makes the result robust to
    # exactly how many steps were requested.
    state_B = original_state_B
    best_state_B = original_state_B
    best_magnitude = original_magnitude
    for _ in range(n_steps):
        grad = _trigger_magnitude_grad(state_B, state_A, ipg_vector)
        grad_norm = jnp.linalg.norm(grad)
        step = jnp.where(grad_norm > 1e-12, grad / grad_norm, jnp.zeros_like(grad))
        state_B = state_B + sign * step_size * step
        # Project back into the epsilon L2-ball around the original vector.
        delta = state_B - original_state_B
        delta_norm = jnp.linalg.norm(delta)
        state_B = jnp.where(delta_norm > epsilon, original_state_B + delta / delta_norm * epsilon, state_B)

        current_magnitude = float(_trigger_magnitude(state_B, state_A, ipg_vector))
        is_better = (current_magnitude > best_magnitude) if direction == "flip_to_dynamic" else (current_magnitude < best_magnitude)
        if is_better:
            best_magnitude = current_magnitude
            best_state_B = state_B

    state_B = best_state_B
    final_magnitude = float(_trigger_magnitude(state_B, state_A, ipg_vector))
    final_trigger_active = final_magnitude > threshold
    perturbation_norm = float(jnp.linalg.norm(state_B - original_state_B))

    perturbed_vettori = vettori.copy()
    perturbed_vettori[target_idx] = np.asarray(state_B)

    flipped = final_trigger_active != original_trigger_active
    success = flipped and (
        (direction == "flip_to_dynamic" and final_trigger_active)
        or (direction == "flip_to_static" and not final_trigger_active)
    )

    return {
        "perturbed_vettori": perturbed_vettori,
        "success": bool(success),
        "original_trigger_active": bool(original_trigger_active),
        "final_trigger_active": bool(final_trigger_active),
        "original_magnitude": original_magnitude,
        "final_magnitude": final_magnitude,
        "perturbation_norm": perturbation_norm,
    }