Merge pull request #768 from abigailgold/dev_1.5.0_OneHotAttribute

One-hot attribute inference

Author: beat-buesser

art/attacks/attack.py

@@ -317,7 +317,7 @@ class AttributeInferenceAttack(InferenceAttack):
 
     attack_params = InferenceAttack.attack_params + ["attack_feature"]
 
-    def __init__(self, estimator, attack_feature: int = 0):
+    def __init__(self, estimator, attack_feature: Union[int, slice] = 0):
         """
         :param estimator: A trained estimator targeted for inference attack.
         :type estimator: :class:`.art.estimators.estimator.BaseEstimator`
@@ -346,10 +346,6 @@ def set_params(self, **kwargs) -> None:
         super().set_params(**kwargs)
         self._check_params()
 
-    def _check_params(self) -> None:
-        if self.attack_feature < 0:
-            raise ValueError("Attack feature must be positive.")
-
 
 class ReconstructionAttack(Attack):
     """

art/attacks/inference/attribute_inference/black_box.py

@@ -22,7 +22,7 @@
 from __future__ import absolute_import, division, print_function, unicode_literals
 
 import logging
-from typing import Optional, TYPE_CHECKING
+from typing import Optional, Union, TYPE_CHECKING
 
 import numpy as np
 from sklearn.neural_network import MLPClassifier
@@ -51,16 +51,24 @@ class AttributeInferenceBlackBox(AttributeInferenceAttack):
     _estimator_requirements = (BaseEstimator, ClassifierMixin)
 
     def __init__(
-        self, classifier: "CLASSIFIER_TYPE", attack_model: Optional["CLASSIFIER_TYPE"] = None, attack_feature: int = 0
+        self,
+        classifier: "CLASSIFIER_TYPE",
+        attack_model: Optional["CLASSIFIER_TYPE"] = None,
+        attack_feature: Union[int, slice] = 0,
     ):
         """
         Create an AttributeInferenceBlackBox attack instance.
 
         :param classifier: Target classifier.
         :param attack_model: The attack model to train, optional. If none is provided, a default model will be created.
-        :param attack_feature: The index of the feature to be attacked.
+        :param attack_feature: The index of the feature to be attacked or a slice representing multiple indexes in
+                               case of a one-hot encoded feature.
         """
         super().__init__(estimator=classifier, attack_feature=attack_feature)
+        if isinstance(self.attack_feature, int):
+            self.single_index_feature = True
+        else:
+            self.single_index_feature = False
 
         if attack_model:
             if ClassifierMixin not in type(attack_model).__mro__:
@@ -104,16 +112,18 @@ def fit(self, x: np.ndarray) -> None:
         # Checks:
         if self.estimator.input_shape[0] != x.shape[1]:
             raise ValueError("Shape of x does not match input_shape of classifier")
-        if self.attack_feature >= x.shape[1]:
+        if self.single_index_feature and self.attack_feature >= x.shape[1]:
             raise ValueError("attack_feature must be a valid index to a feature in x")
 
         # get model's predictions for x
         predictions = np.array([np.argmax(arr) for arr in self.estimator.predict(x)]).reshape(-1, 1)
 
         # get vector of attacked feature
         y = x[:, self.attack_feature]
-        y_one_hot = float_to_categorical(y)
-        y_ready = check_and_transform_label_format(y_one_hot, len(np.unique(y)), return_one_hot=True)
+        y_ready = y
+        if self.single_index_feature:
+            y_one_hot = float_to_categorical(y)
+            y_ready = check_and_transform_label_format(y_one_hot, len(np.unique(y)), return_one_hot=True)
 
         # create training set for attack model
         x_train = np.concatenate((np.delete(x, self.attack_feature, 1), predictions), axis=1).astype(np.float32)
@@ -127,18 +137,27 @@ def infer(self, x: np.ndarray, y: np.ndarray, **kwargs) -> np.ndarray:
 
         :param x: Input to attack. Includes all features except the attacked feature.
         :param y: Original model's predictions for x.
-        :param values: Possible values for attacked feature.
+        :param values: Possible values for attacked feature. Only needed in case of categorical feature (not one-hot).
         :type values: `np.ndarray`
         :return: The inferred feature values.
         """
         if y.shape[0] != x.shape[0]:
             raise ValueError("Number of rows in x and y do not match")
-        if self.estimator.input_shape[0] != x.shape[1] + 1:
+        if self.single_index_feature and self.estimator.input_shape[0] != x.shape[1] + 1:
             raise ValueError("Number of features in x + 1 does not match input_shape of classifier")
 
-        if "values" not in kwargs.keys():
-            raise ValueError("Missing parameter `values`.")
-        values: np.ndarray = kwargs.get("values")
-
         x_test = np.concatenate((x, y), axis=1).astype(np.float32)
-        return np.array([values[np.argmax(arr)] for arr in self.attack_model.predict(x_test)])
+
+        if self.single_index_feature:
+            if "values" not in kwargs.keys():
+                raise ValueError("Missing parameter `values`.")
+            values: np.ndarray = kwargs.get("values")
+            return np.array([values[np.argmax(arr)] for arr in self.attack_model.predict(x_test)])
+        else:
+            return np.array(self.attack_model.predict(x_test))
+
+    def _check_params(self) -> None:
+        if not isinstance(self.attack_feature, int) and not isinstance(self.attack_feature, slice):
+            raise ValueError("Attack feature must be either an integer or a slice object.")
+        if isinstance(self.attack_feature, int) and self.attack_feature < 0:
+            raise ValueError("Attack feature index must be positive.")

art/attacks/inference/attribute_inference/white_box_decision_tree.py

@@ -55,6 +55,7 @@ def __init__(self, classifier: ScikitlearnDecisionTreeClassifier, attack_feature
         :param attack_feature: The index of the feature to be attacked.
         """
         super().__init__(estimator=classifier, attack_feature=attack_feature)
+        self._check_params()
 
     def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.ndarray:
         """
@@ -138,3 +139,7 @@ def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.n
                 for index, value in enumerate(predicted_pred)
             ]
         )
+
+    def _check_params(self) -> None:
+        if self.attack_feature < 0:
+            raise ValueError("Attack feature must be positive.")

art/attacks/inference/attribute_inference/white_box_lifestyle_decision_tree.py

@@ -55,6 +55,7 @@ def __init__(self, classifier: "CLASSIFIER_TYPE", attack_feature: int = 0):
         :param attack_feature: The index of the feature to be attacked.
         """
         super().__init__(estimator=classifier, attack_feature=attack_feature)
+        self._check_params()
 
     def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.ndarray:
         """
@@ -130,3 +131,7 @@ def _calculate_phi(self, x, values, n_samples):
             phi.append(num_value)
 
         return phi
+
+    def _check_params(self) -> None:
+        if self.attack_feature < 0:
+            raise ValueError("Attack feature must be positive.")

tests/attacks/inference/attribute_inference/test_black_box.py

@@ -23,11 +23,13 @@
 import numpy as np
 import torch.nn as nn
 import torch.optim as optim
+from sklearn.tree import DecisionTreeClassifier
 
 from art.attacks.inference.attribute_inference.black_box import AttributeInferenceBlackBox
 from art.estimators.classification.pytorch import PyTorchClassifier
 from art.estimators.estimator import BaseEstimator
 from art.estimators.classification import ClassifierMixin
+from art.estimators.classification.scikitlearn import ScikitlearnDecisionTreeClassifier
 
 from tests.attacks.utils import backend_test_classifier_type_check_fail
 from tests.utils import ARTTestException
@@ -143,5 +145,89 @@ def transform_feature(x):
         art_warning(e)
 
 
+@pytest.mark.skipMlFramework("dl_frameworks")
+def test_black_box_one_hot(art_warning, get_iris_dataset):
+    try:
+        attack_feature = 2  # petal length
+
+        # need to transform attacked feature into categorical
+        def transform_feature(x):
+            x[x > 0.5] = 2
+            x[(x > 0.2) & (x <= 0.5)] = 1
+            x[x <= 0.2] = 0
+
+        (x_train_iris, y_train_iris), (x_test_iris, y_test_iris) = get_iris_dataset
+        # training data without attacked feature
+        x_train_for_attack = np.delete(x_train_iris, attack_feature, 1)
+        # only attacked feature
+        x_train_feature = x_train_iris[:, attack_feature].copy().reshape(-1, 1)
+        transform_feature(x_train_feature)
+        # transform to one-hot encoding
+        train_one_hot = np.zeros((x_train_feature.size, int(x_train_feature.max()) + 1))
+        train_one_hot[np.arange(x_train_feature.size), x_train_feature.reshape(1, -1).astype(int)] = 1
+        # training data with attacked feature (after transformation)
+        x_train = np.concatenate((x_train_for_attack[:, :attack_feature], train_one_hot), axis=1)
+        x_train = np.concatenate((x_train, x_train_for_attack[:, attack_feature:]), axis=1)
+
+        y_train = np.array([np.argmax(y) for y in y_train_iris]).reshape(-1, 1)
+
+        # test data without attacked feature
+        x_test_for_attack = np.delete(x_test_iris, attack_feature, 1)
+        # only attacked feature
+        x_test_feature = x_test_iris[:, attack_feature].copy().reshape(-1, 1)
+        transform_feature(x_test_feature)
+        # transform to one-hot encoding
+        test_one_hot = np.zeros((x_test_feature.size, int(x_test_feature.max()) + 1))
+        test_one_hot[np.arange(x_test_feature.size), x_test_feature.reshape(1, -1).astype(int)] = 1
+        # test data with attacked feature (after transformation)
+        x_test = np.concatenate((x_test_for_attack[:, :attack_feature], test_one_hot), axis=1)
+        x_test = np.concatenate((x_test, x_test_for_attack[:, attack_feature:]), axis=1)
+
+        tree = DecisionTreeClassifier()
+        tree.fit(x_train, y_train)
+        classifier = ScikitlearnDecisionTreeClassifier(tree)
+
+        attack = AttributeInferenceBlackBox(classifier, attack_feature=slice(attack_feature, attack_feature + 3))
+        # get original model's predictions
+        x_train_predictions = np.array([np.argmax(arr) for arr in classifier.predict(x_train)]).reshape(-1, 1)
+        x_test_predictions = np.array([np.argmax(arr) for arr in classifier.predict(x_test)]).reshape(-1, 1)
+        # train attack model
+        attack.fit(x_train)
+        # infer attacked feature
+        inferred_train = attack.infer(x_train_for_attack, x_train_predictions)
+        inferred_test = attack.infer(x_test_for_attack, x_test_predictions)
+        # check accuracy
+        train_acc = np.sum(np.all(inferred_train == train_one_hot, axis=1)) / len(inferred_train)
+        test_acc = np.sum(np.all(inferred_test == test_one_hot, axis=1)) / len(inferred_test)
+        assert pytest.approx(0.9145, abs=0.03) == train_acc
+        assert pytest.approx(0.9333, abs=0.03) == test_acc
+
+    except ARTTestException as e:
+        art_warning(e)
+
+
+def test_errors(art_warning, tabular_dl_estimator_for_attack, get_iris_dataset):
+    try:
+        classifier = tabular_dl_estimator_for_attack(AttributeInferenceBlackBox)
+        (x_train, y_train), (x_test, y_test) = get_iris_dataset
+
+        with pytest.raises(ValueError):
+            AttributeInferenceBlackBox(classifier, attack_feature="a")
+        with pytest.raises(ValueError):
+            AttributeInferenceBlackBox(classifier, attack_feature=-3)
+        attack = AttributeInferenceBlackBox(classifier, attack_feature=8)
+        with pytest.raises(ValueError):
+            attack.fit(x_train)
+        attack = AttributeInferenceBlackBox(classifier)
+        with pytest.raises(ValueError):
+            attack.fit(np.delete(x_train, 1, 1))
+        with pytest.raises(ValueError):
+            attack.infer(x_train, y_test)
+        with pytest.raises(ValueError):
+            attack.infer(x_train, y_train)
+    except ARTTestException as e:
+        art_warning(e)
+
+
 def test_classifier_type_check_fail():
     backend_test_classifier_type_check_fail(AttributeInferenceBlackBox, (BaseEstimator, ClassifierMixin))