From cc1263e2c7e2a18796061e7c5b8d2e3c447f71dc Mon Sep 17 00:00:00 2001 From: Pranjali <155157936+Pranjali-Nagpal@users.noreply.github.com> Date: Sun, 19 Jul 2026 04:04:44 +0530 Subject: [PATCH 1/2] Implement ANDA attack with statistics tracking Added ANDA attack implementation with statistics collection and sampling. --- core/transfer_attack_core.py | 138 +++++++++++++++++++++++++++++++++++ 1 file changed, 138 insertions(+) diff --git a/core/transfer_attack_core.py b/core/transfer_attack_core.py index ace34f44..6243eb8d 100644 --- a/core/transfer_attack_core.py +++ b/core/transfer_attack_core.py @@ -1,3 +1,4 @@ +import math import os import random import uuid @@ -40,6 +41,7 @@ 'IDAA', 'DPA_HMA', 'DYNAMIC_MORPH', + 'ANDA', ] ATTACK_COLS = { @@ -62,6 +64,7 @@ 'IDAA': 'idaa_path', 'DYNAMIC_MORPH': 'dynamic_morph_path', 'DPA_HMA': 'dpa_hma_path', + 'ANDA': 'anda_path', } EPSILON = 0.062 @@ -83,6 +86,9 @@ PGN_BETA = 3.0 PGN_GAMMA = 0.5 PGN_NUM_NEIGHBOR = 20 +ANDA_N_ENS = int(os.environ.get('TRANSFER_ATTACK_ANDA_N_ENS', '25')) +ANDA_AUG_MAX = float(os.environ.get('TRANSFER_ATTACK_ANDA_AUG_MAX', '0.3')) +ANDA_NUM_ITER = int(os.environ.get('TRANSFER_ATTACK_ANDA_NUM_ITER', str(NUM_ITER))) DPA_HMA_SEED = int(os.environ.get('TRANSFER_ATTACK_DPA_HMA_SEED', '1')) DPA_HMA_NUM_ITER = int(os.environ.get('TRANSFER_ATTACK_DPA_HMA_NUM_ITER', str(NUM_ITER))) DPA_HMA_ENSEMBLE_NUM_ITER = int(os.environ.get('TRANSFER_ATTACK_DPA_HMA_ENSEMBLE_NUM_ITER', str(DPA_HMA_NUM_ITER))) @@ -1273,6 +1279,136 @@ def pgn_attack(model, x, tgt_emb, attack_type): return adv +# ANDA (Asymptotically Normal Distribution Attack) +# Paper basis: "Strong Transferable Adversarial Attacks via Ensembled +# Asymptotically Normal Distribution Learning" (CVPR 2024), Fang et al. +class ANDAStatistics: + + def __init__(self, shape): + self.shape = tuple(shape) + self._flat_dim = int(np.prod(self.shape)) + self.n_models = 0.0 + self.noise_mean = tf.zeros((1,) + self.shape, dtype=tf.float32) + self.noise_cov_mat_sqrt = tf.zeros([0, self._flat_dim], dtype=tf.float32) + + def collect_stat(self, noise): + """noise: tensor of shape (bs,) + self.shape, one gradient per augmentation.""" + bs = int(noise.shape[0]) + bs_f = float(bs) + # Running mean update, identical in form to the official: + # mean_(t+1) = mean_t * n/(n+bs) + sum(noise)/(n+bs) + new_mean = ( + self.noise_mean * (self.n_models / (self.n_models + bs_f)) + + tf.reduce_sum(noise, axis=0, keepdims=True) / (self.n_models + bs_f) + ) + # Deviation of each augmented sample from the *updated* mean, stacked + # as additional rows/columns of the covariance square-root matrix D. + dev = tf.reshape(noise - new_mean, [bs, self._flat_dim]) + self.noise_cov_mat_sqrt = tf.concat([self.noise_cov_mat_sqrt, dev], axis=0) + self.noise_mean = new_mean + self.n_models += bs_f + + def sample(self, n_sample=1, scale=1.0, seed=None): + """Draw perturbation(s) from the learned Gaussian posterior N(mean, cov).""" + if seed is not None: + tf.random.set_seed(seed) + mean = self.noise_mean + if scale == 0.0: + assert n_sample == 1 + return mean + cov_sqrt = self.noise_cov_mat_sqrt + k = int(cov_sqrt.shape[0]) + rand = tf.random.normal([n_sample, k]) + cov_sample = tf.matmul(rand, cov_sqrt) / tf.sqrt(tf.cast(k - 1, tf.float32)) + rand_sample = tf.reshape(cov_sample, (n_sample,) + self.shape) + return mean + scale * rand_sample + + +def _anda_get_thetas(n_ens, min_r, max_r): + + n = int(round(math.sqrt(n_ens))) + if n * n != n_ens: + raise ValueError('ANDA requires n_ens to be a perfect square, as in the official implementation.') + coords = np.linspace(min_r, max_r, n) + thetas_ij = [(float(i), float(j)) for i in coords for j in coords] + return thetas_ij + + +def _anda_translate_batch(batch, thetas_ij, width, height): + + transforms = [] + for (i, j) in thetas_ij: + tx = i * (float(width) / 2.0) + ty = j * (float(height) / 2.0) + transforms.append([1.0, 0.0, tx, 0.0, 1.0, ty, 0.0, 0.0]) + transforms = tf.constant(transforms, dtype=tf.float32) + out_shape = tf.constant([height, width], dtype=tf.int32) + return tf.raw_ops.ImageProjectiveTransformV3( + images=tf.cast(batch, tf.float32), + transforms=transforms, + output_shape=out_shape, + interpolation='BILINEAR', + fill_mode='CONSTANT', + fill_value=0.0, + ) + + +def anda_attack( + model, + x, + tgt_emb, + attack_type, + input_size, + n_ens: int = ANDA_N_ENS, + aug_max: float = ANDA_AUG_MAX, + num_iter: int = ANDA_NUM_ITER, + sample: bool = False, +): + """ANDA adapted to embedding-based face verification. + The only intentional algorithmic substitution is the loss: the original + cross-entropy classification loss is replaced with this repository's + cosine-similarity `attack_loss(...)`, per assignment requirements. + """ + thetas_ij = _anda_get_thetas(n_ens, -aug_max, aug_max) + width, height = int(input_size[0]), int(input_size[1]) + alpha = EPSILON / num_iter + tgt_emb = tf.nn.l2_normalize(tgt_emb, axis=1) + is_impersonation = str(attack_type).strip().lower() == 'impersonation_attack' + + xt = tf.identity(x) + stat = ANDAStatistics(shape=tuple(int(d) for d in x.shape[1:])) + sample_xt = None + + for it in range(num_iter): + xt_batch = tf.repeat(xt, n_ens, axis=0) + with tf.GradientTape() as tape: + tape.watch(xt_batch) + aug_batch = _anda_translate_batch(xt_batch, thetas_ij, width, height) + emb = compute_embedding(model, aug_batch) + tgt_rep = tf.repeat(tgt_emb, n_ens, axis=0) + cos = tf.reduce_sum(emb * tgt_rep, axis=1) + loss = attack_loss(cos, attack_type) + grad = tape.gradient(loss, xt_batch) + grad = tf.where(tf.math.is_finite(grad), grad, tf.zeros_like(grad)) + + stat.collect_stat(grad) + sample_noise = stat.noise_mean + + if sample and it == num_iter - 1: + sample_noise_draw = stat.sample(n_sample=1, scale=1.0)[0] + sample_xt = xt + alpha * tf.sign(sample_noise_draw) + sample_xt = tf.clip_by_value(sample_xt, x - EPSILON, x + EPSILON) + sample_xt = tf.clip_by_value(sample_xt, -1.0, 1.0) + + xt = xt + alpha * tf.sign(sample_noise) + xt = tf.clip_by_value(xt, x - EPSILON, x + EPSILON) + xt = tf.clip_by_value(xt, -1.0, 1.0) + + if sample: + return sample_xt + return xt + + def dpa_hma(model, x, tgt_emb, attack_type, num_copies: int = 8, num_iter: int = DPA_HMA_NUM_ITER): _ensure_dpa_hma_seed() tgt_emb = tf.nn.l2_normalize(tgt_emb, axis=1) @@ -1540,6 +1676,8 @@ def run_attack(attack_name: str, model, src, tgt, attack_type: str, input_size): return dpa_hma(model, src, tgt_emb, attack_type) if attack_name == 'DYNAMIC_MORPH': return dynamic_morph_mi_fgsm(model, src, tgt, attack_type, input_size) + if attack_name == 'ANDA': + return anda_attack(model, src, tgt_emb, attack_type, input_size) if attack_name == 'DPA_HMA_ENSEMBLE': raise ValueError( 'DPA_HMA_ENSEMBLE requires dpa_hma_ensemble(...) with victim-specific ' From 295c5d66a70b09414f409b282ba08c9dec45e498 Mon Sep 17 00:00:00 2001 From: Pranjali <155157936+Pranjali-Nagpal@users.noreply.github.com> Date: Sun, 19 Jul 2026 04:53:03 +0530 Subject: [PATCH 2/2] Fix indentation for ANDA attack handling --- core/transfer_attack_core.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/transfer_attack_core.py b/core/transfer_attack_core.py index 6243eb8d..0d3e989c 100644 --- a/core/transfer_attack_core.py +++ b/core/transfer_attack_core.py @@ -1676,8 +1676,8 @@ def run_attack(attack_name: str, model, src, tgt, attack_type: str, input_size): return dpa_hma(model, src, tgt_emb, attack_type) if attack_name == 'DYNAMIC_MORPH': return dynamic_morph_mi_fgsm(model, src, tgt, attack_type, input_size) - if attack_name == 'ANDA': - return anda_attack(model, src, tgt_emb, attack_type, input_size) + if attack_name == 'ANDA': + return anda_attack(model, src, tgt_emb, attack_type, input_size) if attack_name == 'DPA_HMA_ENSEMBLE': raise ValueError( 'DPA_HMA_ENSEMBLE requires dpa_hma_ensemble(...) with victim-specific '