8137716344

Committed 22 Jan 2024 10:53PM UTC coverage: 98.47%. Remained the same

Build # 8137716344

Build Type

push

github

Committed by

web-flow

Commit Message

Bump pillow from 10.1.0 to 10.2.0 in /tests (#718)

Bumps [pillow](https://github.com/python-pillow/Pillow) from 10.1.0 to 10.2.0.
- [Release notes](https://github.com/python-pillow/Pillow/releases)
- [Changelog](https://github.com/python-pillow/Pillow/blob/main/CHANGES.rst)
- [Commits](https://github.com/python-pillow/Pillow/compare/10.1.0...10.2.0)

---
updated-dependencies:
- dependency-name: pillow
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

3475 of 3529 relevant lines covered (98.47%)

7.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.44

/foolbox/attacks/base.py

from typing import Callable, TypeVar, Any, Union, Optional, Sequence, List, Tuple, Dict
from typing_extensions import final, overload
from abc import ABC, abstractmethod
from collections.abc import Iterable
import eagerpy as ep

from ..models import Model

from ..criteria import Criterion
from ..criteria import Misclassification

from ..devutils import atleast_kd

from ..distances import Distance


T = TypeVar("T")
CriterionType = TypeVar("CriterionType", bound=Criterion)


# TODO: support manually specifying early_stop in __call__


class Attack(ABC):
    @overload
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Sequence[Union[float, None]],
        **kwargs: Any,
    ) -> Tuple[List[T], List[T], T]:
        ...

    @overload  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[float, None],
        **kwargs: Any,
    ) -> Tuple[T, T, T]:
        ...

    @abstractmethod  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[Sequence[Union[float, None]], float, None],
        **kwargs: Any,
    ) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
        # in principle, the type of criterion is Union[Criterion, T]
        # but we want to give subclasses the option to specify the supported
        # criteria explicitly (i.e. specifying a stricter type constraint)
        ...

    @abstractmethod
    def repeat(self, times: int) -> "Attack":
        ...

    def __repr__(self) -> str:
        args = ", ".join(f"{k.strip('_')}={v}" for k, v in vars(self).items())
        return f"{self.__class__.__name__}({args})"


class AttackWithDistance(Attack):
    @property
    @abstractmethod
    def distance(self) -> Distance:
        ...

    def repeat(self, times: int) -> Attack:
        return Repeated(self, times)


class Repeated(AttackWithDistance):
    """Repeats the wrapped attack and returns the best result"""

    def __init__(self, attack: AttackWithDistance, times: int):
        if times < 1:
            raise ValueError(f"expected times >= 1, got {times}")  # pragma: no cover

        self.attack = attack
        self.times = times

    @property
    def distance(self) -> Distance:
        return self.attack.distance

    @overload
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Sequence[Union[float, None]],
        **kwargs: Any,
    ) -> Tuple[List[T], List[T], T]:
        ...

    @overload  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[float, None],
        **kwargs: Any,
    ) -> Tuple[T, T, T]:
        ...

    def __call__(  # noqa: F811
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[Sequence[Union[float, None]], float, None],
        **kwargs: Any,
    ) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
        x, restore_type = ep.astensor_(inputs)
        del inputs

        verify_input_bounds(x, model)

        criterion = get_criterion(criterion)

        was_iterable = True
        if not isinstance(epsilons, Iterable):
            epsilons = [epsilons]
            was_iterable = False

        N = len(x)
        K = len(epsilons)

        for i in range(self.times):
            # run the attack
            xps, xpcs, success = self.attack(
                model, x, criterion, epsilons=epsilons, **kwargs
            )
            assert len(xps) == K
            assert len(xpcs) == K
            for xp in xps:
                assert xp.shape == x.shape
            for xpc in xpcs:
                assert xpc.shape == x.shape
            assert success.shape == (K, N)

            if i == 0:
                best_xps = xps
                best_xpcs = xpcs
                best_success = success
                continue

            # TODO: test if stacking the list to a single tensor and
            # getting rid of the loop is faster

            for k, epsilon in enumerate(epsilons):
                first = best_success[k].logical_not()
                assert first.shape == (N,)
                if epsilon is None:
                    # if epsilon is None, we need the minimum

                    # TODO: maybe cache some of these distances
                    # and then remove the else part
                    closer = self.distance(x, xps[k]) < self.distance(x, best_xps[k])
                    assert closer.shape == (N,)
                    new_best = ep.logical_and(success[k], ep.logical_or(closer, first))
                else:
                    # for concrete epsilon, we just need a successful one
                    new_best = ep.logical_and(success[k], first)
                new_best = atleast_kd(new_best, x.ndim)
                best_xps[k] = ep.where(new_best, xps[k], best_xps[k])
                best_xpcs[k] = ep.where(new_best, xpcs[k], best_xpcs[k])

            best_success = ep.logical_or(success, best_success)

        best_xps_ = [restore_type(xp) for xp in best_xps]
        best_xpcs_ = [restore_type(xpc) for xpc in best_xpcs]
        if was_iterable:
            return best_xps_, best_xpcs_, restore_type(best_success)
        else:
            assert len(best_xps_) == 1
            assert len(best_xpcs_) == 1
            return (
                best_xps_[0],
                best_xpcs_[0],
                restore_type(best_success.squeeze(axis=0)),
            )

    def repeat(self, times: int) -> "Repeated":
        return Repeated(self.attack, self.times * times)


class FixedEpsilonAttack(AttackWithDistance):
    """Fixed-epsilon attacks try to find adversarials whose perturbation sizes
    are limited by a fixed epsilon"""

    @abstractmethod
    def run(
        self, model: Model, inputs: T, criterion: Any, *, epsilon: float, **kwargs: Any
    ) -> T:
        """Runs the attack and returns perturbed inputs.

        The size of the perturbations should be at most epsilon, but this
        is not guaranteed and the caller should verify this or clip the result.
        """
        ...

    @overload
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Sequence[Union[float, None]],
        **kwargs: Any,
    ) -> Tuple[List[T], List[T], T]:
        ...

    @overload  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[float, None],
        **kwargs: Any,
    ) -> Tuple[T, T, T]:
        ...

    @final  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[Sequence[Union[float, None]], float, None],
        **kwargs: Any,
    ) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:

        x, restore_type = ep.astensor_(inputs)
        del inputs

        verify_input_bounds(x, model)

        criterion = get_criterion(criterion)
        is_adversarial = get_is_adversarial(criterion, model)

        was_iterable = True
        if not isinstance(epsilons, Iterable):
            epsilons = [epsilons]
            was_iterable = False

        N = len(x)
        K = len(epsilons)

        # None means: just minimize, no early stopping, no limit on the perturbation size
        if any(eps is None for eps in epsilons):
            # TODO: implement a binary search
            raise NotImplementedError(
                "FixedEpsilonAttack subclasses do not yet support None in epsilons"
            )
        real_epsilons = [eps for eps in epsilons if eps is not None]
        del epsilons

        xps = []
        xpcs = []
        success = []
        for epsilon in real_epsilons:
            xp = self.run(model, x, criterion, epsilon=epsilon, **kwargs)

            # clip to epsilon because we don't really know what the attack returns;
            # alternatively, we could check if the perturbation is at most epsilon,
            # but then we would need to handle numerical violations;
            xpc = self.distance.clip_perturbation(x, xp, epsilon)
            is_adv = is_adversarial(xpc)

            xps.append(xp)
            xpcs.append(xpc)
            success.append(is_adv)

        # # TODO: the correction we apply here should make sure that the limits
        # # are not violated, but this is a hack and we need a better solution
        # # Alternatively, maybe can just enforce the limits in __call__
        # xps = [
        #     self.run(model, x, criterion, epsilon=epsilon, **kwargs)
        #     for epsilon in real_epsilons
        # ]

        # is_adv = ep.stack([is_adversarial(xp) for xp in xps])
        # assert is_adv.shape == (K, N)

        # in_limits = ep.stack(
        #     [
        #         self.distance(x, xp) <= epsilon
        #         for xp, epsilon in zip(xps, real_epsilons)
        #     ],
        # )
        # assert in_limits.shape == (K, N)

        # if not in_limits.all():
        #     # TODO handle (numerical) violations
        #     # warn user if run() violated the epsilon constraint
        #     import pdb

        #     pdb.set_trace()

        # success = ep.logical_and(in_limits, is_adv)
        # assert success.shape == (K, N)

        success_ = ep.stack(success)
        assert success_.shape == (K, N)

        xps_ = [restore_type(xp) for xp in xps]
        xpcs_ = [restore_type(xpc) for xpc in xpcs]

        if was_iterable:
            return xps_, xpcs_, restore_type(success_)
        else:
            assert len(xps_) == 1
            assert len(xpcs_) == 1
            return xps_[0], xpcs_[0], restore_type(success_.squeeze(axis=0))


class MinimizationAttack(AttackWithDistance):
    """Minimization attacks try to find adversarials with minimal perturbation sizes"""

    @abstractmethod
    def run(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        early_stop: Optional[float] = None,
        **kwargs: Any,
    ) -> T:
        """Runs the attack and returns perturbed inputs.

        The size of the perturbations should be as small as possible such that
        the perturbed inputs are still adversarial. In general, this is not
        guaranteed and the caller has to verify this.
        """
        ...

    @overload
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Sequence[Union[float, None]],
        **kwargs: Any,
    ) -> Tuple[List[T], List[T], T]:
        ...

    @overload  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[float, None],
        **kwargs: Any,
    ) -> Tuple[T, T, T]:
        ...

    @final  # noqa: F811
    def __call__(
        self,
        model: Model,
        inputs: T,
        criterion: Any,
        *,
        epsilons: Union[Sequence[Union[float, None]], float, None],
        **kwargs: Any,
    ) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
        x, restore_type = ep.astensor_(inputs)
        del inputs

        verify_input_bounds(x, model)

        criterion = get_criterion(criterion)
        is_adversarial = get_is_adversarial(criterion, model)

        was_iterable = True
        if not isinstance(epsilons, Iterable):
            epsilons = [epsilons]
            was_iterable = False

        N = len(x)
        K = len(epsilons)

        # None means: just minimize, no early stopping, no limit on the perturbation size
        if any(eps is None for eps in epsilons):
            early_stop = None
        else:
            early_stop = min(epsilons)  # type: ignore

        # run the actual attack
        xp = self.run(model, x, criterion, early_stop=early_stop, **kwargs)

        xpcs = []
        success = []
        for epsilon in epsilons:
            if epsilon is None:
                xpc = xp
            else:
                xpc = self.distance.clip_perturbation(x, xp, epsilon)
            is_adv = is_adversarial(xpc)

            xpcs.append(xpc)
            success.append(is_adv)

        success_ = ep.stack(success)
        assert success_.shape == (K, N)

        xp_ = restore_type(xp)
        xpcs_ = [restore_type(xpc) for xpc in xpcs]

        if was_iterable:
            return [xp_] * K, xpcs_, restore_type(success_)
        else:
            assert len(xpcs_) == 1
            return xp_, xpcs_[0], restore_type(success_.squeeze(axis=0))


class FlexibleDistanceMinimizationAttack(MinimizationAttack):
    def __init__(self, *, distance: Optional[Distance] = None):
        self._distance = distance

    @property
    def distance(self) -> Distance:
        if self._distance is None:
            # we delay the error until the distance is needed,
            # e.g. when __call__ is executed (that way, run
            # can be used without specifying a distance)
            raise ValueError(
                "unknown distance, please pass `distance` to the attack initializer"
            )
        return self._distance


def get_is_adversarial(
    criterion: Criterion, model: Model
) -> Callable[[ep.Tensor], ep.Tensor]:
    def is_adversarial(perturbed: ep.Tensor) -> ep.Tensor:
        outputs = model(perturbed)
        return criterion(perturbed, outputs)

    return is_adversarial


def get_criterion(criterion: Union[Criterion, Any]) -> Criterion:
    if isinstance(criterion, Criterion):
        return criterion
    else:
        return Misclassification(criterion)


def get_channel_axis(model: Model, ndim: int) -> Optional[int]:
    data_format = getattr(model, "data_format", None)
    if data_format is None:
        return None
    if data_format == "channels_first":
        return 1
    if data_format == "channels_last":
        return ndim - 1
    raise ValueError(
        f"unknown data_format, expected 'channels_first' or 'channels_last', got {data_format}"
    )


def raise_if_kwargs(kwargs: Dict[str, Any]) -> None:
    if kwargs:
        raise TypeError(
            f"attack got an unexpected keyword argument '{next(iter(kwargs.keys()))}'"
        )


def verify_input_bounds(input: ep.Tensor, model: Model) -> None:
    # verify that input to the attack lies within model's input bounds
    assert input.min().item() >= model.bounds.lower
    assert input.max().item() <= model.bounds.upper

1	from typing import Callable, TypeVar, Any, Union, Optional, Sequence, List, Tuple, Dict	10✔
2	from typing_extensions import final, overload	10✔
3	from abc import ABC, abstractmethod	10✔
4	from collections.abc import Iterable	10✔
5	import eagerpy as ep	10✔
6
7	from ..models import Model	10✔
8
9	from ..criteria import Criterion	10✔
10	from ..criteria import Misclassification	10✔
11
12	from ..devutils import atleast_kd	10✔
13
14	from ..distances import Distance	10✔
15
16
17	T = TypeVar("T")	10✔
18	CriterionType = TypeVar("CriterionType", bound=Criterion)	10✔
19
20
21	# TODO: support manually specifying early_stop in __call__
22
23
24	class Attack(ABC):	10✔
25	@overload
26	def __call__(
27	self,
28	model: Model,
29	inputs: T,
30	criterion: Any,
31	*,
32	epsilons: Sequence[Union[float, None]],
33	**kwargs: Any,
34	) -> Tuple[List[T], List[T], T]:
35	...
36
37	@overload # noqa: F811
38	def __call__(
39	self,
40	model: Model,
41	inputs: T,
42	criterion: Any,
43	*,
44	epsilons: Union[float, None],
45	**kwargs: Any,
46	) -> Tuple[T, T, T]:
47	...
48
49	@abstractmethod # noqa: F811
50	def __call__(
51	self,
52	model: Model,
53	inputs: T,
54	criterion: Any,
55	*,
56	epsilons: Union[Sequence[Union[float, None]], float, None],
57	**kwargs: Any,
58	) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
59	# in principle, the type of criterion is Union[Criterion, T]
60	# but we want to give subclasses the option to specify the supported
61	# criteria explicitly (i.e. specifying a stricter type constraint)
62	...
63
64	@abstractmethod
65	def repeat(self, times: int) -> "Attack":
66	...
67
68	def __repr__(self) -> str:	10✔
69	args = ", ".join(f"{k.strip('_')}={v}" for k, v in vars(self).items())	10✔
70	return f"{self.__class__.__name__}({args})"	10✔
71
72
73	class AttackWithDistance(Attack):	10✔
74	@property	10✔
75	@abstractmethod
76	def distance(self) -> Distance:
77	...
78
79	def repeat(self, times: int) -> Attack:	10✔
80	return Repeated(self, times)	10✔
81
82
83	class Repeated(AttackWithDistance):	10✔
84	"""Repeats the wrapped attack and returns the best result"""
85
86	def __init__(self, attack: AttackWithDistance, times: int):	10✔
87	if times < 1:	10✔
88	raise ValueError(f"expected times >= 1, got {times}") # pragma: no cover
89
90	self.attack = attack	10✔
91	self.times = times	10✔
92
93	@property	10✔
94	def distance(self) -> Distance:	10✔
95	return self.attack.distance	8✔
96
97	@overload
98	def __call__(
99	self,
100	model: Model,
101	inputs: T,
102	criterion: Any,
103	*,
104	epsilons: Sequence[Union[float, None]],
105	**kwargs: Any,
106	) -> Tuple[List[T], List[T], T]:
107	...
108
109	@overload # noqa: F811
110	def __call__(
111	self,
112	model: Model,
113	inputs: T,
114	criterion: Any,
115	*,
116	epsilons: Union[float, None],
117	**kwargs: Any,
118	) -> Tuple[T, T, T]:
119	...
120
121	def __call__( # noqa: F811	10✔
122	self,
123	model: Model,
124	inputs: T,
125	criterion: Any,
126	*,
127	epsilons: Union[Sequence[Union[float, None]], float, None],
128	**kwargs: Any,
129	) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
130	x, restore_type = ep.astensor_(inputs)	8✔
131	del inputs	8✔
132
133	verify_input_bounds(x, model)	8✔
134
135	criterion = get_criterion(criterion)	8✔
136
137	was_iterable = True	8✔
138	if not isinstance(epsilons, Iterable):	8✔
139	epsilons = [epsilons]	8✔
140	was_iterable = False	8✔
141
142	N = len(x)	8✔
143	K = len(epsilons)	8✔
144
145	for i in range(self.times):	8✔
146	# run the attack
147	xps, xpcs, success = self.attack(	8✔
148	model, x, criterion, epsilons=epsilons, **kwargs
149	)
150	assert len(xps) == K	8✔
151	assert len(xpcs) == K	8✔
152	for xp in xps:	8✔
153	assert xp.shape == x.shape	8✔
154	for xpc in xpcs:	8✔
155	assert xpc.shape == x.shape	8✔
156	assert success.shape == (K, N)	8✔
157
158	if i == 0:	8✔
159	best_xps = xps	8✔
160	best_xpcs = xpcs	8✔
161	best_success = success	8✔
162	continue	8✔
163
164	# TODO: test if stacking the list to a single tensor and
165	# getting rid of the loop is faster
166
167	for k, epsilon in enumerate(epsilons):	8✔
168	first = best_success[k].logical_not()	8✔
169	assert first.shape == (N,)	8✔
170	if epsilon is None:	8✔
171	# if epsilon is None, we need the minimum
172
173	# TODO: maybe cache some of these distances
174	# and then remove the else part
175	closer = self.distance(x, xps[k]) < self.distance(x, best_xps[k])	8✔
176	assert closer.shape == (N,)	8✔
177	new_best = ep.logical_and(success[k], ep.logical_or(closer, first))	8✔
178	else:
179	# for concrete epsilon, we just need a successful one
180	new_best = ep.logical_and(success[k], first)	8✔
181	new_best = atleast_kd(new_best, x.ndim)	8✔
182	best_xps[k] = ep.where(new_best, xps[k], best_xps[k])	8✔
183	best_xpcs[k] = ep.where(new_best, xpcs[k], best_xpcs[k])	8✔
184
185	best_success = ep.logical_or(success, best_success)	8✔
186
187	best_xps_ = [restore_type(xp) for xp in best_xps]	8✔
188	best_xpcs_ = [restore_type(xpc) for xpc in best_xpcs]	8✔
189	if was_iterable:	8✔
190	return best_xps_, best_xpcs_, restore_type(best_success)	×
191	else:
192	assert len(best_xps_) == 1	8✔
193	assert len(best_xpcs_) == 1	8✔
194	return (	8✔
195	best_xps_[0],
196	best_xpcs_[0],
197	restore_type(best_success.squeeze(axis=0)),
198	)
199
200	def repeat(self, times: int) -> "Repeated":	10✔
201	return Repeated(self.attack, self.times * times)	10✔
202
203
204	class FixedEpsilonAttack(AttackWithDistance):	10✔
205	"""Fixed-epsilon attacks try to find adversarials whose perturbation sizes
206	are limited by a fixed epsilon"""
207
208	@abstractmethod
209	def run(
210	self, model: Model, inputs: T, criterion: Any, , epsilon: float, *kwargs: Any
211	) -> T:
212	"""Runs the attack and returns perturbed inputs.
213
214	The size of the perturbations should be at most epsilon, but this
215	is not guaranteed and the caller should verify this or clip the result.
216	"""
217	...
218
219	@overload
220	def __call__(
221	self,
222	model: Model,
223	inputs: T,
224	criterion: Any,
225	*,
226	epsilons: Sequence[Union[float, None]],
227	**kwargs: Any,
228	) -> Tuple[List[T], List[T], T]:
229	...
230
231	@overload # noqa: F811
232	def __call__(
233	self,
234	model: Model,
235	inputs: T,
236	criterion: Any,
237	*,
238	epsilons: Union[float, None],
239	**kwargs: Any,
240	) -> Tuple[T, T, T]:
241	...
242
243	@final # noqa: F811	10✔
244	def __call__(	10✔
245	self,
246	model: Model,
247	inputs: T,
248	criterion: Any,
249	*,
250	epsilons: Union[Sequence[Union[float, None]], float, None],
251	**kwargs: Any,
252	) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
253
254	x, restore_type = ep.astensor_(inputs)	8✔
255	del inputs	8✔
256
257	verify_input_bounds(x, model)	8✔
258
259	criterion = get_criterion(criterion)	8✔
260	is_adversarial = get_is_adversarial(criterion, model)	8✔
261
262	was_iterable = True	8✔
263	if not isinstance(epsilons, Iterable):	8✔
264	epsilons = [epsilons]	8✔
265	was_iterable = False	8✔
266
267	N = len(x)	8✔
268	K = len(epsilons)	8✔
269
270	# None means: just minimize, no early stopping, no limit on the perturbation size
271	if any(eps is None for eps in epsilons):	8✔
272	# TODO: implement a binary search
273	raise NotImplementedError(
274	"FixedEpsilonAttack subclasses do not yet support None in epsilons"
275	)
276	real_epsilons = [eps for eps in epsilons if eps is not None]	8✔
277	del epsilons	8✔
278
279	xps = []	8✔
280	xpcs = []	8✔
281	success = []	8✔
282	for epsilon in real_epsilons:	8✔
283	xp = self.run(model, x, criterion, epsilon=epsilon, **kwargs)	8✔
284
285	# clip to epsilon because we don't really know what the attack returns;
286	# alternatively, we could check if the perturbation is at most epsilon,
287	# but then we would need to handle numerical violations;
288	xpc = self.distance.clip_perturbation(x, xp, epsilon)	8✔
289	is_adv = is_adversarial(xpc)	8✔
290
291	xps.append(xp)	8✔
292	xpcs.append(xpc)	8✔
293	success.append(is_adv)	8✔
294
295	# # TODO: the correction we apply here should make sure that the limits
296	# # are not violated, but this is a hack and we need a better solution
297	# # Alternatively, maybe can just enforce the limits in __call__
298	# xps = [
299	# self.run(model, x, criterion, epsilon=epsilon, **kwargs)
300	# for epsilon in real_epsilons
301	# ]
302
303	# is_adv = ep.stack([is_adversarial(xp) for xp in xps])
304	# assert is_adv.shape == (K, N)
305
306	# in_limits = ep.stack(
307	# [
308	# self.distance(x, xp) <= epsilon
309	# for xp, epsilon in zip(xps, real_epsilons)
310	# ],
311	# )
312	# assert in_limits.shape == (K, N)
313
314	# if not in_limits.all():
315	# # TODO handle (numerical) violations
316	# # warn user if run() violated the epsilon constraint
317	# import pdb
318
319	# pdb.set_trace()
320
321	# success = ep.logical_and(in_limits, is_adv)
322	# assert success.shape == (K, N)
323
324	success_ = ep.stack(success)	8✔
325	assert success_.shape == (K, N)	8✔
326
327	xps_ = [restore_type(xp) for xp in xps]	8✔
328	xpcs_ = [restore_type(xpc) for xpc in xpcs]	8✔
329
330	if was_iterable:	8✔
331	return xps_, xpcs_, restore_type(success_)	8✔
332	else:
333	assert len(xps_) == 1	8✔
334	assert len(xpcs_) == 1	8✔
335	return xps_[0], xpcs_[0], restore_type(success_.squeeze(axis=0))	8✔
336
337
338	class MinimizationAttack(AttackWithDistance):	10✔
339	"""Minimization attacks try to find adversarials with minimal perturbation sizes"""
340
341	@abstractmethod
342	def run(
343	self,
344	model: Model,
345	inputs: T,
346	criterion: Any,
347	*,
348	early_stop: Optional[float] = None,
349	**kwargs: Any,
350	) -> T:
351	"""Runs the attack and returns perturbed inputs.
352
353	The size of the perturbations should be as small as possible such that
354	the perturbed inputs are still adversarial. In general, this is not
355	guaranteed and the caller has to verify this.
356	"""
357	...
358
359	@overload
360	def __call__(
361	self,
362	model: Model,
363	inputs: T,
364	criterion: Any,
365	*,
366	epsilons: Sequence[Union[float, None]],
367	**kwargs: Any,
368	) -> Tuple[List[T], List[T], T]:
369	...
370
371	@overload # noqa: F811
372	def __call__(
373	self,
374	model: Model,
375	inputs: T,
376	criterion: Any,
377	*,
378	epsilons: Union[float, None],
379	**kwargs: Any,
380	) -> Tuple[T, T, T]:
381	...
382
383	@final # noqa: F811	10✔
384	def __call__(	10✔
385	self,
386	model: Model,
387	inputs: T,
388	criterion: Any,
389	*,
390	epsilons: Union[Sequence[Union[float, None]], float, None],
391	**kwargs: Any,
392	) -> Union[Tuple[List[T], List[T], T], Tuple[T, T, T]]:
393	x, restore_type = ep.astensor_(inputs)	10✔
394	del inputs	10✔
395
396	verify_input_bounds(x, model)	10✔
397
398	criterion = get_criterion(criterion)	10✔
399	is_adversarial = get_is_adversarial(criterion, model)	10✔
400
401	was_iterable = True	10✔
402	if not isinstance(epsilons, Iterable):	10✔
403	epsilons = [epsilons]	10✔
404	was_iterable = False	10✔
405
406	N = len(x)	10✔
407	K = len(epsilons)	10✔
408
409	# None means: just minimize, no early stopping, no limit on the perturbation size
410	if any(eps is None for eps in epsilons):	10✔
411	early_stop = None	10✔
412	else:
413	early_stop = min(epsilons) # type: ignore	8✔
414
415	# run the actual attack
416	xp = self.run(model, x, criterion, early_stop=early_stop, **kwargs)	10✔
417
418	xpcs = []	8✔
419	success = []	8✔
420	for epsilon in epsilons:	8✔
421	if epsilon is None:	8✔
422	xpc = xp	8✔
423	else:
424	xpc = self.distance.clip_perturbation(x, xp, epsilon)	8✔
425	is_adv = is_adversarial(xpc)	8✔
426
427	xpcs.append(xpc)	8✔
428	success.append(is_adv)	8✔
429
430	success_ = ep.stack(success)	8✔
431	assert success_.shape == (K, N)	8✔
432
433	xp_ = restore_type(xp)	8✔
434	xpcs_ = [restore_type(xpc) for xpc in xpcs]	8✔
435
436	if was_iterable:	8✔
437	return [xp_] * K, xpcs_, restore_type(success_)	8✔
438	else:
439	assert len(xpcs_) == 1	8✔
440	return xp_, xpcs_[0], restore_type(success_.squeeze(axis=0))	8✔
441
442
443	class FlexibleDistanceMinimizationAttack(MinimizationAttack):	10✔
444	def __init__(self, *, distance: Optional[Distance] = None):	10✔
445	self._distance = distance	10✔
446
447	@property	10✔
448	def distance(self) -> Distance:	10✔
449	if self._distance is None:	8✔
450	# we delay the error until the distance is needed,
451	# e.g. when __call__ is executed (that way, run
452	# can be used without specifying a distance)
453	raise ValueError(	8✔
454	"unknown distance, please pass `distance` to the attack initializer"
455	)
456	return self._distance	8✔
457
458
459	def get_is_adversarial(	10✔
460	criterion: Criterion, model: Model
461	) -> Callable[[ep.Tensor], ep.Tensor]:
462	def is_adversarial(perturbed: ep.Tensor) -> ep.Tensor:	10✔
463	outputs = model(perturbed)	8✔
464	return criterion(perturbed, outputs)	8✔
465
466	return is_adversarial	10✔
467
468
469	def get_criterion(criterion: Union[Criterion, Any]) -> Criterion:	10✔
470	if isinstance(criterion, Criterion):	10✔
471	return criterion	10✔
472	else:
473	return Misclassification(criterion)	10✔
474
475
476	def get_channel_axis(model: Model, ndim: int) -> Optional[int]:	10✔
477	data_format = getattr(model, "data_format", None)	10✔
478	if data_format is None:	10✔
479	return None	10✔
480	if data_format == "channels_first":	10✔
481	return 1	10✔
482	if data_format == "channels_last":	10✔
483	return ndim - 1	10✔
484	raise ValueError(	10✔
485	f"unknown data_format, expected 'channels_first' or 'channels_last', got {data_format}"
486	)
487
488
489	def raise_if_kwargs(kwargs: Dict[str, Any]) -> None:	10✔
490	if kwargs:	10✔
491	raise TypeError(	8✔
492	f"attack got an unexpected keyword argument '{next(iter(kwargs.keys()))}'"
493	)
494
495
496	def verify_input_bounds(input: ep.Tensor, model: Model) -> None:	10✔
497	# verify that input to the attack lies within model's input bounds
498	assert input.min().item() >= model.bounds.lower	10✔
499	assert input.max().item() <= model.bounds.upper	10✔

bethgelab / foolbox / 8137716344

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous