Black box
This commit is contained in:
63
Tune_new.py
63
Tune_new.py
@@ -165,65 +165,6 @@ def log_metrics(evaluation_domains, reloaded, strategy_in_use):
|
|||||||
strategy=strategy_in_use
|
strategy=strategy_in_use
|
||||||
)
|
)
|
||||||
|
|
||||||
# performs MIA and ZRF attack on models and logs the results
|
|
||||||
def run_unlearning_and_attack_eval(forget_train_loader, retain_test_loader, reloaded, strategy_in_use, suite_runner, device, forget_class):
|
|
||||||
"""
|
|
||||||
Performs adversarial vulnerability stress tests (MIA and ZRF) in-memory
|
|
||||||
on the freshly unlearned model instance without saving it to disk first.
|
|
||||||
"""
|
|
||||||
if suite_runner is None:
|
|
||||||
raise ValueError("An active initialized UnlearningAttackSuite instance must be supplied.")
|
|
||||||
|
|
||||||
print(f"\n>>> Initializing Threat Model Stress Testing Suite for: {strategy_in_use}")
|
|
||||||
|
|
||||||
# 1. Dynamically map the white-box feature extraction hook to the active inner model
|
|
||||||
suite_runner.register_model_hook(reloaded.model)
|
|
||||||
|
|
||||||
# 2. Fire the complete evaluation suite using the isolated data split subsets
|
|
||||||
results = suite_runner.run_complete_evaluation(
|
|
||||||
target_class=forget_class,
|
|
||||||
framework_name=strategy_in_use,
|
|
||||||
forget_loader=forget_train_loader, # Members split from the train data partition
|
|
||||||
retain_test_loader=retain_test_loader, # Clean non-members split from validation data
|
|
||||||
device=device
|
|
||||||
)
|
|
||||||
|
|
||||||
print(f" [Attack Complete] Logit MIA AUC: {results['logit_mia_auc']:.4f} | "
|
|
||||||
f"Internal MIA AUC: {results['internal_mia_auc']:.4f} | "
|
|
||||||
f"ZRF Score: {results['zrf_score']:.4f}")
|
|
||||||
|
|
||||||
|
|
||||||
# performs MIA and ZRF attack on models and logs the results
|
|
||||||
def run_shaddow_attack_eval(forget_train_loader, retain_test_loader, reloaded, strategy_in_use, suite_runner, device, forget_class):
|
|
||||||
"""
|
|
||||||
Performs adversarial vulnerability stress tests matching the localized
|
|
||||||
shadow architecture specifications laid out in thesis Section 5.5.
|
|
||||||
"""
|
|
||||||
if suite_runner is None:
|
|
||||||
raise ValueError("An active initialized UnlearningAttackSuite instance must be supplied.")
|
|
||||||
|
|
||||||
print(f"\n>>> Initializing Threat Model Stress Testing Suite for: {strategy_in_use}")
|
|
||||||
|
|
||||||
# Instantiate a clean copy of the baseline trained model to serve as the Shadow reference proxy
|
|
||||||
# (Since finetuning is done once, we read its parameters cleanly from disk)
|
|
||||||
base_shadow = Model.create(arch=ARCH, device=device, size=CLASS_SIZE)
|
|
||||||
base_shadow.load(arch=ARCH)
|
|
||||||
|
|
||||||
# Execute the updated conditional attack framework
|
|
||||||
results = suite_runner.run_complete_evaluation(
|
|
||||||
framework_name=strategy_in_use,
|
|
||||||
target_class=forget_class,
|
|
||||||
forget_loader=forget_train_loader,
|
|
||||||
retain_test_loader=retain_test_loader,
|
|
||||||
unlearned_instance=reloaded, # The unlearned candidate model
|
|
||||||
base_shadow_instance=base_shadow, # The shadow proxy architecture
|
|
||||||
device=device
|
|
||||||
)
|
|
||||||
|
|
||||||
print(f" [Attack Complete] Adversary Binary Classification Accuracy: {results['mia_accuracy']:.4f}")
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
# Unlearning and strategy eval
|
# Unlearning and strategy eval
|
||||||
def run_unlearning_and_strategy_eval(env_dict, forget_class_idx, strategy, evaluate = False, suite_runner=None):
|
def run_unlearning_and_strategy_eval(env_dict, forget_class_idx, strategy, evaluate = False, suite_runner=None):
|
||||||
"""
|
"""
|
||||||
@@ -393,8 +334,8 @@ if __name__ == "__main__":
|
|||||||
#dist_attacker.run_adversarial_evaluation()
|
#dist_attacker.run_adversarial_evaluation()
|
||||||
#dist_attacker.run_incremental_evaluation(current_class_step=i)
|
#dist_attacker.run_incremental_evaluation(current_class_step=i)
|
||||||
|
|
||||||
if suite_runner is not None:
|
#if suite_runner is not None:
|
||||||
suite_runner.shutdown_hook()
|
#suite_runner.shutdown_hook()
|
||||||
|
|
||||||
except KeyboardInterrupt:
|
except KeyboardInterrupt:
|
||||||
print("\nprogram interrupted. Exit!")
|
print("\nprogram interrupted. Exit!")
|
||||||
|
|||||||
@@ -18,9 +18,9 @@ class UnlearningAttack:
|
|||||||
self.arch = arch
|
self.arch = arch
|
||||||
self.class_size = class_size
|
self.class_size = class_size
|
||||||
|
|
||||||
self.hook = None
|
#self.hook = None
|
||||||
self.model = None
|
#self.model = None
|
||||||
self._hook_features = []
|
#self._hook_features = []
|
||||||
self.criterion = nn.CrossEntropyLoss(reduction='none')
|
self.criterion = nn.CrossEntropyLoss(reduction='none')
|
||||||
self.collecting = False
|
self.collecting = False
|
||||||
|
|
||||||
@@ -46,7 +46,7 @@ class UnlearningAttack:
|
|||||||
|
|
||||||
def calculate_a_dist(self, latent1, latent2):
|
def calculate_a_dist(self, latent1, latent2):
|
||||||
"""Calculates formal A-Distance: 2 * (1 - 2 * epsilon)."""
|
"""Calculates formal A-Distance: 2 * (1 - 2 * epsilon)."""
|
||||||
combined = np.vstack([latent1, latent2])
|
'''combined = np.vstack([latent1, latent2])
|
||||||
mean = np.mean(combined, axis=0)
|
mean = np.mean(combined, axis=0)
|
||||||
std = np.std(combined, axis=0) + 1e-8
|
std = np.std(combined, axis=0) + 1e-8
|
||||||
l1 = (latent1 - mean) / std
|
l1 = (latent1 - mean) / std
|
||||||
@@ -63,13 +63,16 @@ class UnlearningAttack:
|
|||||||
split = int(len(X) * 0.7)
|
split = int(len(X) * 0.7)
|
||||||
|
|
||||||
clf = LogisticRegression(solver='liblinear').fit(X[:split], y[:split])
|
clf = LogisticRegression(solver='liblinear').fit(X[:split], y[:split])
|
||||||
epsilon = 1.0 - accuracy_score(y[split:], clf.predict(X[split:]))
|
epsilon = 1.0 - accuracy_score(y[split:], clf.predict(X[split:]))'''
|
||||||
|
|
||||||
|
accuracy_score = self._comput_adversarial_accuracy(latent1, latent2, axis=0)
|
||||||
|
epsilon = 1 - accuracy_score
|
||||||
|
|
||||||
return 2.0 * np.abs(0.5 - epsilon)
|
return 2.0 * np.abs(0.5 - epsilon)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def _hook_fn(self, module, input, output):
|
'''def _hook_fn(self, module, input, output):
|
||||||
if not self.collecting:
|
if not self.collecting:
|
||||||
return
|
return
|
||||||
flattened_embeddings = torch.flatten(output, 1)
|
flattened_embeddings = torch.flatten(output, 1)
|
||||||
@@ -99,9 +102,11 @@ class UnlearningAttack:
|
|||||||
if hasattr(self, 'hook') and self.hook:
|
if hasattr(self, 'hook') and self.hook:
|
||||||
self.hook.remove()
|
self.hook.remove()
|
||||||
self.hook = None
|
self.hook = None
|
||||||
|
'''
|
||||||
def _extract_attack_features(self, target_model, loader, device, target_class):
|
def _extract_attack_features(self, target_model, loader, device, target_class):
|
||||||
'''target_model.eval()
|
'''
|
||||||
|
# cnahgng to black box.
|
||||||
|
target_model.eval()
|
||||||
all_probs = []
|
all_probs = []
|
||||||
all_entropies = []
|
all_entropies = []
|
||||||
all_losses = []
|
all_losses = []
|
||||||
@@ -223,11 +228,11 @@ class UnlearningAttack:
|
|||||||
# Note: Latent distance is removed as it's not a black-box metric
|
# Note: Latent distance is removed as it's not a black-box metric
|
||||||
return mia_accuracy, 0.0
|
return mia_accuracy, 0.0
|
||||||
|
|
||||||
def _comput_adversarial_accuracy(self, filtered, naive, axis=-1):
|
def _comput_adversarial_accuracy(self, filtered, naive, axis=0):
|
||||||
|
|
||||||
# Z-Score Normalisation
|
# Z-Score Normalisation
|
||||||
filtered = (filtered - np.mean(filtered, axis=-1, keepdims=True)) / (np.std(filtered, axis=-1, keepdims=True) + 1e-8)
|
filtered = (filtered - np.mean(filtered, axis = axis, keepdims=True)) / (np.std(filtered, axis = axis, keepdims = True) + 1e-8)
|
||||||
naive = (naive - np.mean(naive, axis=-1, keepdims=True)) / (np.std(naive, axis=-1, keepdims=True) + 1e-8)
|
naive = (naive - np.mean(naive, axis = axis, keepdims = True)) / (np.std(naive, axis = axis, keepdims = True) + 1e-8)
|
||||||
|
|
||||||
# shuffle indices
|
# shuffle indices
|
||||||
num_images = len(filtered)
|
num_images = len(filtered)
|
||||||
@@ -282,7 +287,7 @@ class UnlearningAttack:
|
|||||||
naive = torch.cat(naive_logits, dim=0).cpu().numpy()
|
naive = torch.cat(naive_logits, dim=0).cpu().numpy()
|
||||||
|
|
||||||
# evaluate similarity of outputs
|
# evaluate similarity of outputs
|
||||||
lookalike_accuracy = self._comput_adversarial_accuracy(filtered=filtered, naive=naive)
|
lookalike_accuracy = self._comput_adversarial_accuracy(filtered=filtered, naive=naive, axis = -1)
|
||||||
# so that the metric is between 0 and 1.
|
# so that the metric is between 0 and 1.
|
||||||
return 2.0 * np.abs(lookalike_accuracy - 0.5)
|
return 2.0 * np.abs(lookalike_accuracy - 0.5)
|
||||||
|
|
||||||
@@ -296,7 +301,7 @@ class UnlearningAttack:
|
|||||||
with open(current_log_file, "w") as f:
|
with open(current_log_file, "w") as f:
|
||||||
f.write("target_class, parameter_mia_accuracy, latent_distance_tell, lookalike_accuracy, A-Dist, JS-Dist\n")
|
f.write("target_class, parameter_mia_accuracy, latent_distance_tell, lookalike_accuracy, A-Dist, JS-Dist\n")
|
||||||
|
|
||||||
self.register_model_hook(unlearned_instance.model)
|
#self.register_model_hook(unlearned_instance.model)
|
||||||
|
|
||||||
# 1. Parameter-Space MIA and Latent Distance
|
# 1. Parameter-Space MIA and Latent Distance
|
||||||
parameter_mia_acc, latent_dist = self.run_parameter_space_mia(
|
parameter_mia_acc, latent_dist = self.run_parameter_space_mia(
|
||||||
|
|||||||
Reference in New Issue
Block a user