Merge pull request #1588 from abigailgold/dev_1.10.0_att_fix

Support slices of size 1 in attribute attacks

Author: beat-buesser

art/attacks/inference/attribute_inference/baseline.py

@@ -29,7 +29,13 @@
 
 from art.estimators.classification.classifier import ClassifierMixin
 from art.attacks.attack import AttributeInferenceAttack
-from art.utils import check_and_transform_label_format, float_to_categorical, floats_to_one_hot, get_feature_values
+from art.utils import (
+    check_and_transform_label_format,
+    float_to_categorical,
+    floats_to_one_hot,
+    get_feature_values,
+    get_feature_index,
+)
 
 if TYPE_CHECKING:
     from art.utils import CLASSIFIER_TYPE
@@ -65,11 +71,6 @@ def __init__(
         """
         super().__init__(estimator=None, attack_feature=attack_feature)
 
-        if isinstance(self.attack_feature, int):
-            self.single_index_feature = True
-        else:
-            self.single_index_feature = False
-
         self._values: Optional[list] = None
 
         if attack_model:
@@ -108,6 +109,7 @@ def __init__(
             raise ValueError("Illegal value for parameter `attack_model_type`.")
 
         self._check_params()
+        self.attack_feature = get_feature_index(self.attack_feature)
 
     def fit(self, x: np.ndarray) -> None:
         """
@@ -117,13 +119,13 @@ def fit(self, x: np.ndarray) -> None:
         """
 
         # Checks:
-        if self.single_index_feature and isinstance(self.attack_feature, int) and self.attack_feature >= x.shape[1]:
+        if isinstance(self.attack_feature, int) and self.attack_feature >= x.shape[1]:
             raise ValueError("attack_feature must be a valid index to a feature in x")
 
         # get vector of attacked feature
         y = x[:, self.attack_feature]
-        self._values = get_feature_values(y, self.single_index_feature)
-        if self.single_index_feature:
+        self._values = get_feature_values(y, isinstance(self.attack_feature, int))
+        if isinstance(self.attack_feature, int):
             y_one_hot = float_to_categorical(y)
         else:
             y_one_hot = floats_to_one_hot(y)
@@ -161,7 +163,7 @@ def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.n
         predictions = self.attack_model.predict(x_test).astype(np.float32)
 
         if self._values is not None:
-            if self.single_index_feature:
+            if isinstance(self.attack_feature, int):
                 predictions = np.array([self._values[np.argmax(arr)] for arr in predictions])
             else:
                 i = 0

art/attacks/inference/attribute_inference/black_box.py

@@ -32,7 +32,13 @@
 from art.estimators.classification.classifier import ClassifierMixin
 from art.attacks.attack import AttributeInferenceAttack
 from art.estimators.regression import RegressorMixin
-from art.utils import check_and_transform_label_format, float_to_categorical, floats_to_one_hot, get_feature_values
+from art.utils import (
+    check_and_transform_label_format,
+    float_to_categorical,
+    floats_to_one_hot,
+    get_feature_values,
+    get_feature_index,
+)
 
 if TYPE_CHECKING:
     from art.utils import CLASSIFIER_TYPE, REGRESSOR_TYPE
@@ -83,10 +89,6 @@ def __init__(
                                          `estimator` is a regressor and if `scale_range` is not supplied.
         """
         super().__init__(estimator=estimator, attack_feature=attack_feature)
-        if isinstance(self.attack_feature, int):
-            self.single_index_feature = True
-        else:
-            self.single_index_feature = False
 
         self._values: Optional[list] = None
         self._attack_model_type = attack_model_type
@@ -131,6 +133,7 @@ def __init__(
         self.scale_range = scale_range
 
         self._check_params()
+        self.attack_feature = get_feature_index(self.attack_feature)
 
     def fit(self, x: np.ndarray, y: Optional[np.ndarray] = None) -> None:
         """
@@ -144,7 +147,7 @@ def fit(self, x: np.ndarray, y: Optional[np.ndarray] = None) -> None:
         if self.estimator.input_shape is not None:
             if self.estimator.input_shape[0] != x.shape[1]:
                 raise ValueError("Shape of x does not match input_shape of model")
-        if self.single_index_feature and isinstance(self.attack_feature, int) and self.attack_feature >= x.shape[1]:
+        if isinstance(self.attack_feature, int) 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
@@ -162,8 +165,8 @@ def fit(self, x: np.ndarray, y: Optional[np.ndarray] = None) -> None:
 
         # get vector of attacked feature
         y_attack = x[:, self.attack_feature]
-        self._values = get_feature_values(y_attack, self.single_index_feature)
-        if self.single_index_feature:
+        self._values = get_feature_values(y_attack, isinstance(self.attack_feature, int))
+        if isinstance(self.attack_feature, int):
             y_one_hot = float_to_categorical(y_attack)
         else:
             y_one_hot = floats_to_one_hot(y_attack)
@@ -210,7 +213,7 @@ def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.n
         if pred.shape[0] != x.shape[0]:
             raise ValueError("Number of rows in x and y do not match")
         if self.estimator.input_shape is not None:
-            if self.single_index_feature and self.estimator.input_shape[0] != x.shape[1] + 1:
+            if isinstance(self.attack_feature, int) and self.estimator.input_shape[0] != x.shape[1] + 1:
                 raise ValueError("Number of features in x + 1 does not match input_shape of model")
 
         if RegressorMixin in type(self.estimator).__mro__:
@@ -234,7 +237,7 @@ def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.n
         predictions = self.attack_model.predict(x_test).astype(np.float32)
 
         if self._values is not None:
-            if self.single_index_feature:
+            if isinstance(self.attack_feature, int):
                 predictions = np.array([self._values[np.argmax(arr)] for arr in predictions])
             else:
                 i = 0

art/attacks/inference/attribute_inference/meminf_based.py

@@ -30,6 +30,7 @@
 from art.attacks.attack import AttributeInferenceAttack, MembershipInferenceAttack
 from art.estimators.regression import RegressorMixin
 from art.exceptions import EstimatorError
+from art.utils import get_feature_index
 
 if TYPE_CHECKING:
     from art.utils import CLASSIFIER_TYPE, REGRESSOR_TYPE
@@ -68,6 +69,7 @@ def __init__(
 
         self.membership_attack = membership_attack
         self._check_params()
+        self.attack_feature = get_feature_index(self.attack_feature)
 
     def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.ndarray:
         """
@@ -104,7 +106,6 @@ def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.n
                 v_full = np.full((x.shape[0], 1), value).astype(x.dtype)
                 x_value = np.concatenate((x[:, : self.attack_feature], v_full), axis=1)
                 x_value = np.concatenate((x_value, x[:, self.attack_feature :]), axis=1)
-
                 predicted = self.membership_attack.infer(x_value, y, probabilities=True)
                 if first:
                     probabilities = predicted

art/attacks/inference/attribute_inference/true_label_baseline.py

@@ -30,7 +30,13 @@
 
 from art.estimators.classification.classifier import ClassifierMixin
 from art.attacks.attack import AttributeInferenceAttack
-from art.utils import check_and_transform_label_format, float_to_categorical, floats_to_one_hot, get_feature_values
+from art.utils import (
+    check_and_transform_label_format,
+    float_to_categorical,
+    floats_to_one_hot,
+    get_feature_values,
+    get_feature_index,
+)
 
 if TYPE_CHECKING:
     from art.utils import CLASSIFIER_TYPE
@@ -74,11 +80,6 @@ def __init__(
         """
         super().__init__(estimator=None, attack_feature=attack_feature)
 
-        if isinstance(self.attack_feature, int):
-            self.single_index_feature = True
-        else:
-            self.single_index_feature = False
-
         self._values: Optional[list] = None
 
         if attack_model:
@@ -119,6 +120,7 @@ def __init__(
         self.prediction_normal_factor = prediction_normal_factor
         self.scale_range = scale_range
         self._check_params()
+        self.attack_feature = get_feature_index(self.attack_feature)
 
     def fit(self, x: np.ndarray, y: np.ndarray) -> None:
         """
@@ -129,13 +131,13 @@ def fit(self, x: np.ndarray, y: np.ndarray) -> None:
         """
 
         # Checks:
-        if self.single_index_feature and isinstance(self.attack_feature, int) and self.attack_feature >= x.shape[1]:
+        if isinstance(self.attack_feature, int) and self.attack_feature >= x.shape[1]:
             raise ValueError("attack_feature must be a valid index to a feature in x")
 
         # get vector of attacked feature
         attacked_feature = x[:, self.attack_feature]
-        self._values = get_feature_values(attacked_feature, self.single_index_feature)
-        if self.single_index_feature:
+        self._values = get_feature_values(attacked_feature, isinstance(self.attack_feature, int))
+        if isinstance(self.attack_feature, int):
             y_one_hot = float_to_categorical(attacked_feature)
         else:
             y_one_hot = floats_to_one_hot(attacked_feature)
@@ -187,7 +189,7 @@ def infer(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.n
         predictions = self.attack_model.predict(x_test).astype(np.float32)
 
         if self._values is not None:
-            if self.single_index_feature:
+            if isinstance(self.attack_feature, int):
                 predictions = np.array([self._values[np.argmax(arr)] for arr in predictions])
             else:
                 i = 0

art/utils.py

@@ -675,6 +675,30 @@ def get_feature_values(x: np.ndarray, single_index_feature: bool) -> list:
     return values
 
 
+def get_feature_index(feature: Union[int, slice]) -> Union[int, slice]:
+    """
+    Returns a modified feature index: in case of a slice of size 1, returns the corresponding integer. Otherwise,
+    returns the same value (integer or slice) as passed.
+
+    :param feature: The index or slice representing a feature to attack
+    :return: An integer representing a single column index or a slice representing a multi-column index
+    """
+    if isinstance(feature, int):
+        return feature
+
+    start = feature.start
+    stop = feature.stop
+    step = feature.step
+    if start is None:
+        start = 0
+    if step is None:
+        step = 1
+    if feature.stop is not None and ((stop - start) // step) == 1:
+        return start
+
+    return feature
+
+
 def compute_success_array(
     classifier: "CLASSIFIER_TYPE",
     x_clean: np.ndarray,

tests/attacks/inference/attribute_inference/test_baseline.py

@@ -81,6 +81,60 @@ def transform_feature(x):
         art_warning(e)
 
 
+@pytest.mark.skip_framework("dl_frameworks")
+@pytest.mark.parametrize("model_type", ["nn", "rf"])
+def test_black_box_baseline_slice(art_warning, get_iris_dataset, model_type):
+    try:
+        attack_feature = 2  # petal length
+
+        # need to transform attacked feature into categorical
+        def transform_feature(x):
+            x[x > 0.5] = 2.0
+            x[(x > 0.2) & (x <= 0.5)] = 1.0
+            x[x <= 0.2] = 0.0
+
+        values = [0.0, 1.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)
+        # training data with attacked feature (after transformation)
+        x_train = np.concatenate((x_train_for_attack[:, :attack_feature], x_train_feature), axis=1)
+        x_train = np.concatenate((x_train, x_train_for_attack[:, attack_feature:]), axis=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)
+
+        baseline_attack = AttributeInferenceBaseline(
+            attack_feature=slice(attack_feature, attack_feature + 1), attack_model_type=model_type
+        )
+        # train attack model
+        baseline_attack.fit(x_train)
+        # infer attacked feature
+        baseline_inferred_train = baseline_attack.infer(x_train_for_attack, values=values)
+        baseline_inferred_test = baseline_attack.infer(x_test_for_attack, values=values)
+        # check accuracy
+        baseline_train_acc = np.sum(baseline_inferred_train == x_train_feature.reshape(1, -1)) / len(
+            baseline_inferred_train
+        )
+        baseline_test_acc = np.sum(baseline_inferred_test == x_test_feature.reshape(1, -1)) / len(
+            baseline_inferred_test
+        )
+
+        assert 0.8 <= baseline_train_acc
+        assert 0.7 <= baseline_test_acc
+
+    except ARTTestException as e:
+        art_warning(e)
+
+
 @pytest.mark.skip_framework("dl_frameworks")
 @pytest.mark.parametrize("model_type", ["nn", "rf"])
 def test_black_box_baseline_no_values(art_warning, get_iris_dataset, model_type):

tests/attacks/inference/attribute_inference/test_black_box.py

@@ -83,6 +83,59 @@ def transform_feature(x):
         # check accuracy
         train_acc = np.sum(inferred_train == x_train_feature.reshape(1, -1)) / len(inferred_train)
         test_acc = np.sum(inferred_test == x_test_feature.reshape(1, -1)) / len(inferred_test)
+        assert pytest.approx(0.8285, abs=0.2) == train_acc
+        assert pytest.approx(0.8888, abs=0.18) == test_acc
+
+    except ARTTestException as e:
+        art_warning(e)
+
+
+@pytest.mark.skip_framework("dl_frameworks")
+@pytest.mark.parametrize("model_type", ["nn", "rf"])
+def test_black_box_slice(art_warning, decision_tree_estimator, get_iris_dataset, model_type):
+    try:
+        attack_feature = 2  # petal length
+
+        # need to transform attacked feature into categorical
+        def transform_feature(x):
+            x[x > 0.5] = 2.0
+            x[(x > 0.2) & (x <= 0.5)] = 1.0
+            x[x <= 0.2] = 0.0
+
+        values = [0.0, 1.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)
+        # training data with attacked feature (after transformation)
+        x_train = np.concatenate((x_train_for_attack[:, :attack_feature], x_train_feature), axis=1)
+        x_train = np.concatenate((x_train, x_train_for_attack[:, attack_feature:]), axis=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)
+
+        classifier = decision_tree_estimator()
+
+        attack = AttributeInferenceBlackBox(
+            classifier, attack_feature=slice(attack_feature, attack_feature + 1), attack_model_type=model_type
+        )
+        # get original model's predictions
+        x_train_predictions = np.array([np.argmax(arr) for arr in classifier.predict(x_train_iris)]).reshape(-1, 1)
+        x_test_predictions = np.array([np.argmax(arr) for arr in classifier.predict(x_test_iris)]).reshape(-1, 1)
+        # train attack model
+        attack.fit(x_train)
+        # infer attacked feature
+        inferred_train = attack.infer(x_train_for_attack, pred=x_train_predictions, values=values)
+        inferred_test = attack.infer(x_test_for_attack, pred=x_test_predictions, values=values)
+        # check accuracy
+        train_acc = np.sum(inferred_train == x_train_feature.reshape(1, -1)) / len(inferred_train)
+        test_acc = np.sum(inferred_test == x_test_feature.reshape(1, -1)) / len(inferred_test)
         assert pytest.approx(0.8285, abs=0.12) == train_acc
         assert pytest.approx(0.8888, abs=0.18) == test_acc
 
@@ -135,7 +188,7 @@ def transform_feature(x):
         train_acc = np.sum(inferred_train == x_train_feature.reshape(1, -1)) / len(inferred_train)
         test_acc = np.sum(inferred_test == x_test_feature.reshape(1, -1)) / len(inferred_test)
         assert pytest.approx(0.8285, abs=0.12) == train_acc
-        assert pytest.approx(0.8888, abs=0.16) == test_acc
+        assert pytest.approx(0.8888, abs=0.18) == test_acc
 
     except ARTTestException as e:
         art_warning(e)