Merge pull request #711 from Trusted-AI/development_issue_710

Change order of mask and norm steps in PGD attacks

Author: beat-buesser

art/attacks/evasion/fast_gradient.py

@@ -191,27 +191,6 @@ def _minimal_perturbation(self, x: np.ndarray, y: np.ndarray, mask: np.ndarray)
 
         return adv_x
 
-    @staticmethod
-    def _get_mask(x: np.ndarray, **kwargs) -> np.ndarray:
-        """
-        Get the mask from the kwargs.
-
-        :param x: An array with the original inputs.
-        :param mask: An array with a mask to be applied to the adversarial perturbations. Shape needs to be
-                     broadcastable to the shape of x. Any features for which the mask is zero will not be adversarially
-                     perturbed.
-        :type mask: `np.ndarray`
-        :return: The mask.
-        """
-        mask = kwargs.get("mask")
-
-        if mask is not None:
-            # Ensure the mask is broadcastable
-            if len(mask.shape) > len(x.shape) or mask.shape != x.shape[-len(mask.shape) :]:
-                raise ValueError("Mask shape must be broadcastable to input shape.")
-
-        return mask
-
     def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> np.ndarray:
         """Generate adversarial samples and return them in an array.
 
@@ -226,9 +205,7 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         :type mask: `np.ndarray`
         :return: An array holding the adversarial examples.
         """
-        mask = kwargs.get("mask")
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
@@ -355,13 +332,19 @@ def _check_params(self) -> None:
         if not isinstance(self.minimal, bool):
             raise ValueError("The flag `minimal` has to be of type bool.")
 
-    def _compute_perturbation(self, batch: np.ndarray, batch_labels: np.ndarray, mask: np.ndarray) -> np.ndarray:
+    def _compute_perturbation(
+        self, batch: np.ndarray, batch_labels: np.ndarray, mask: Optional[np.ndarray]
+    ) -> np.ndarray:
         # Pick a small scalar to avoid division by 0
         tol = 10e-8
 
         # Get gradient wrt loss; invert it if attack is targeted
         grad = self.estimator.loss_gradient(batch, batch_labels) * (1 - 2 * int(self.targeted))
 
+        # Apply mask
+        if mask is not None:
+            grad = np.where(mask == 0.0, 0.0, grad)
+
         # Apply norm bound
         def _apply_norm(grad, object_type=False):
             if self.norm in [np.inf, "inf"]:
@@ -389,10 +372,7 @@ def _apply_norm(grad, object_type=False):
 
         assert batch.shape == grad.shape
 
-        if mask is None:
-            return grad
-        else:
-            return grad * (mask.astype(ART_NUMPY_DTYPE))
+        return grad
 
     def _apply_perturbation(
         self, batch: np.ndarray, perturbation: np.ndarray, eps_step: Union[int, float, np.ndarray]
@@ -487,3 +467,35 @@ def _compute(
                     x_adv[batch_index_1:batch_index_2] = x_init[batch_index_1:batch_index_2] + perturbation
 
         return x_adv
+
+    @staticmethod
+    def _get_mask(x: np.ndarray, **kwargs) -> np.ndarray:
+        """
+        Get the mask from the kwargs.
+
+        :param x: An array with the original inputs.
+        :param mask: An array with a mask to be applied to the adversarial perturbations. Shape needs to be
+                     broadcastable to the shape of x. Any features for which the mask is zero will not be adversarially
+                     perturbed.
+        :type mask: `np.ndarray`
+        :return: The mask.
+        """
+        mask = kwargs.get("mask")
+
+        if mask is not None:
+            if mask.ndim > x.ndim:
+                raise ValueError("Mask shape must be broadcastable to input shape.")
+
+            if not (np.issubdtype(mask.dtype, np.floating) or mask.dtype == np.bool):
+                raise ValueError(
+                    "The `mask` has to be either of type np.float32, np.float64 or np.bool. The provided"
+                    "`mask` is of type {}.".format(mask.dtype)
+                )
+
+            if np.issubdtype(mask.dtype, np.floating) and np.amin(mask) < 0.0:
+                raise ValueError(
+                    "The `mask` of type np.float32 or np.float64 requires all elements to be either zero"
+                    "or positive values."
+                )
+
+        return mask

art/attacks/evasion/projected_gradient_descent/projected_gradient_descent_numpy.py

@@ -246,21 +246,14 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         :type mask: `np.ndarray`
         :return: An array holding the adversarial examples.
         """
-        mask = kwargs.get("mask")
-
-        # Check the mask
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
 
         # Check whether random eps is enabled
         self._random_eps()
 
-        # Get the mask
-        mask = self._get_mask(x, **kwargs)
-
         if isinstance(self.estimator, ClassifierMixin):
             # Set up targets
             targets = self._set_targets(x, y)

art/attacks/evasion/projected_gradient_descent/projected_gradient_descent_pytorch.py

@@ -120,9 +120,7 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         """
         import torch  # lgtm [py/repeated-import]
 
-        mask = kwargs.get("mask")
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
@@ -249,7 +247,9 @@ def _generate_batch(
 
         return adv_x.cpu().detach().numpy()
 
-    def _compute_perturbation(self, x: "torch.Tensor", y: "torch.Tensor", mask: "torch.Tensor") -> "torch.Tensor":
+    def _compute_perturbation(
+        self, x: "torch.Tensor", y: "torch.Tensor", mask: Optional["torch.Tensor"]
+    ) -> "torch.Tensor":
         """
         Compute perturbations.
 
@@ -271,6 +271,10 @@ def _compute_perturbation(self, x: "torch.Tensor", y: "torch.Tensor", mask: "tor
         # Get gradient wrt loss; invert it if attack is targeted
         grad = self.estimator.loss_gradient(x=x, y=y) * (1 - 2 * int(self.targeted))
 
+        # Apply mask
+        if mask is not None:
+            grad = torch.where(mask == 0.0, torch.tensor(0.0), grad)
+
         # Apply norm bound
         if self.norm in ["inf", np.inf]:
             grad = grad.sign()
@@ -285,10 +289,7 @@ def _compute_perturbation(self, x: "torch.Tensor", y: "torch.Tensor", mask: "tor
 
         assert x.shape == grad.shape
 
-        if mask is None:
-            return grad
-        else:
-            return grad * mask
+        return grad
 
     def _apply_perturbation(
         self, x: "torch.Tensor", perturbation: "torch.Tensor", eps_step: Union[int, float, np.ndarray]

art/attacks/evasion/projected_gradient_descent/projected_gradient_descent_tensorflow_v2.py

@@ -119,9 +119,7 @@ def generate(self, x: np.ndarray, y: Optional[np.ndarray] = None, **kwargs) -> n
         """
         import tensorflow as tf  # lgtm [py/repeated-import]
 
-        mask = kwargs.get("mask")
-        if mask is not None and mask.ndim > x.ndim:
-            raise ValueError("Mask shape must be broadcastable to input shape.")
+        mask = self._get_mask(x, **kwargs)
 
         # Ensure eps is broadcastable
         self._check_compatibility_input_and_eps(x=x)
@@ -224,17 +222,11 @@ def _generate_batch(
 
         :param x: An array with the original inputs.
         :param targets: Target values (class labels) one-hot-encoded of shape `(nb_samples, nb_classes)`.
-<<<<<<< HEAD
-        :param mask: An array with a mask to be applied to the adversarial perturbations. Shape needs to be
-                     broadcastable to the shape of x. Any features for which the mask is zero will not be adversarially
-                     perturbed.
-        :param eps: Maximum perturbation that the attacker can introduce.
-        :param eps_step: Attack step size (input variation) at each iteration.
-=======
         :param mask: An array with a mask broadcastable to input `x` defining where to apply adversarial perturbations.
                      Shape needs to be broadcastable to the shape of x and can also be of the same shape as `x`. Any
                      features for which the mask is zero will not be adversarially perturbed.
->>>>>>> origin/dev_1.5.0
+        :param eps: Maximum perturbation that the attacker can introduce.
+        :param eps_step: Attack step size (input variation) at each iteration.
         :return: Adversarial examples.
         """
         adv_x = x
@@ -269,6 +261,10 @@ def _compute_perturbation(self, x: "tf.Tensor", y: "tf.Tensor", mask: Optional["
             1 - 2 * int(self.targeted), dtype=ART_NUMPY_DTYPE
         )
 
+        # Apply mask
+        if mask is not None:
+            grad = tf.where(mask == 0.0, 0.0, grad)
+
         # Apply norm bound
         if self.norm == np.inf:
             grad = tf.sign(grad)
@@ -285,10 +281,7 @@ def _compute_perturbation(self, x: "tf.Tensor", y: "tf.Tensor", mask: Optional["
 
         assert x.shape == grad.shape
 
-        if mask is None:
-            return grad
-        else:
-            return grad * mask
+        return grad
 
     def _apply_perturbation(
         self, x: "tf.Tensor", perturbation: "tf.Tensor", eps_step: Union[int, float, np.ndarray]

tests/attacks/test_projected_gradient_descent.py

@@ -158,7 +158,7 @@ def _test_backend_mnist(self, classifier, x_train, y_train, x_test, y_test):
         # Test the masking
         attack = ProjectedGradientDescent(classifier, num_random_init=1)
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(x_test.shape))
-        mask = mask.reshape(x_test.shape)
+        mask = mask.reshape(x_test.shape).astype(np.float32)
 
         x_test_adv = attack.generate(x_test, mask=mask)
         mask_diff = (1 - mask) * (x_test_adv - x_test)
@@ -629,11 +629,11 @@ def _test_framework_vs_numpy(self, classifier):
         )
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_train_mnist.shape))
-        mask = mask.reshape(self.x_train_mnist.shape)
+        mask = mask.reshape(self.x_train_mnist.shape).astype(np.float32)
         x_train_adv_np = attack_np.generate(self.x_train_mnist, mask=mask)
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_test_mnist.shape))
-        mask = mask.reshape(self.x_test_mnist.shape)
+        mask = mask.reshape(self.x_test_mnist.shape).astype(np.float32)
         x_test_adv_np = attack_np.generate(self.x_test_mnist, mask=mask)
 
         master_seed(1234)
@@ -650,11 +650,11 @@ def _test_framework_vs_numpy(self, classifier):
         )
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_train_mnist.shape))
-        mask = mask.reshape(self.x_train_mnist.shape)
+        mask = mask.reshape(self.x_train_mnist.shape).astype(np.float32)
         x_train_adv_fw = attack_fw.generate(self.x_train_mnist, mask=mask)
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_test_mnist.shape))
-        mask = mask.reshape(self.x_test_mnist.shape)
+        mask = mask.reshape(self.x_test_mnist.shape).astype(np.float32)
         x_test_adv_fw = attack_fw.generate(self.x_test_mnist, mask=mask)
 
         # Test
@@ -680,11 +680,11 @@ def _test_framework_vs_numpy(self, classifier):
         )
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_train_mnist.shape[1:]))
-        mask = mask.reshape(self.x_train_mnist.shape[1:])
+        mask = mask.reshape(self.x_train_mnist.shape[1:]).astype(np.float32)
         x_train_adv_np = attack_np.generate(self.x_train_mnist, mask=mask)
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_test_mnist.shape[1:]))
-        mask = mask.reshape(self.x_test_mnist.shape[1:])
+        mask = mask.reshape(self.x_test_mnist.shape[1:]).astype(np.float32)
         x_test_adv_np = attack_np.generate(self.x_test_mnist, mask=mask)
 
         master_seed(1234)
@@ -701,11 +701,11 @@ def _test_framework_vs_numpy(self, classifier):
         )
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_train_mnist.shape[1:]))
-        mask = mask.reshape(self.x_train_mnist.shape[1:])
+        mask = mask.reshape(self.x_train_mnist.shape[1:]).astype(np.float32)
         x_train_adv_fw = attack_fw.generate(self.x_train_mnist, mask=mask)
 
         mask = np.random.binomial(n=1, p=0.5, size=np.prod(self.x_test_mnist.shape[1:]))
-        mask = mask.reshape(self.x_test_mnist.shape[1:])
+        mask = mask.reshape(self.x_test_mnist.shape[1:]).astype(np.float32)
         x_test_adv_fw = attack_fw.generate(self.x_test_mnist, mask=mask)
 
         # Test

tests/attacks/utils.py

@@ -207,7 +207,7 @@ def backend_masked_images(attack, fix_get_mnist_subset):
 
     # generate a random mask:
     mask = np.random.binomial(n=1, p=0.5, size=np.prod(x_test_mnist.shape))
-    mask = mask.reshape(x_test_mnist.shape)
+    mask = mask.reshape(x_test_mnist.shape).astype(np.float32)
 
     x_test_adv = attack.generate(x_test_mnist, mask=mask)
     mask_diff = (1 - mask) * (x_test_adv - x_test_mnist)