Skip to content

Documentation for Labelflipping Module

This module provides classes for label flipping attacks in datasets, allowing for the simulation of label noise as a form of data poisoning. It implements both targeted and non-targeted label flipping attacks.

Classes: - LabelFlippingAttack: Main attack class that implements the DatasetAttack interface - LabelFlippingStrategy: Abstract base class for label flipping strategies - TargetedLabelFlippingStrategy: Implementation for targeted label flipping - NonTargetedLabelFlippingStrategy: Implementation for non-targeted label flipping

LabelFlippingAttack

Bases: DatasetAttack

Implements a label flipping attack that can be either targeted or non-targeted.

Source code in nebula/addons/attacks/dataset/labelflipping.py
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
class LabelFlippingAttack(DatasetAttack):
    """
    Implements a label flipping attack that can be either targeted or non-targeted.
    """

    def __init__(self, engine, attack_params: Dict):
        """
        Initialize the label flipping attack.

        Args:
            engine: The engine managing the attack context
            attack_params: Dictionary containing attack parameters
        """
        try:
            round_start = int(attack_params["round_start_attack"])
            round_stop = int(attack_params["round_stop_attack"])
            attack_interval = int(attack_params["attack_interval"])
        except KeyError as e:
            raise ValueError(f"Missing required attack parameter: {e}")
        except ValueError:
            raise ValueError("Invalid value in attack_params. Ensure all values are integers.")

        super().__init__(engine, round_start, round_stop, attack_interval)
        self.datamodule = engine._trainer.datamodule
        self.poisoned_percent = float(attack_params["poisoned_sample_percent"])

        # Create the appropriate strategy based on whether the attack is targeted
        if attack_params.get("targeted", False):
            target_label = int(attack_params.get("target_label") or attack_params.get("targetLabel", 4))
            target_changed_label = int(
                attack_params.get("target_changed_label") or attack_params.get("targetChangedLabel", 7)
            )
            self.strategy = TargetedLabelFlippingStrategy(target_label, target_changed_label)
        else:
            self.strategy = NonTargetedLabelFlippingStrategy()

    def get_malicious_dataset(self):
        """
        Creates a malicious dataset by flipping the labels of selected data points.

        Returns:
            Dataset: The modified dataset with flipped labels
        """
        return self.strategy.flip_labels(
            self.datamodule.train_set,
            self.datamodule.train_set_indices,
            self.poisoned_percent,
        )

__init__(engine, attack_params)

Initialize the label flipping attack.

Parameters:

Name Type Description Default
engine

The engine managing the attack context

required
attack_params Dict

Dictionary containing attack parameters

required
Source code in nebula/addons/attacks/dataset/labelflipping.py
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def __init__(self, engine, attack_params: Dict):
    """
    Initialize the label flipping attack.

    Args:
        engine: The engine managing the attack context
        attack_params: Dictionary containing attack parameters
    """
    try:
        round_start = int(attack_params["round_start_attack"])
        round_stop = int(attack_params["round_stop_attack"])
        attack_interval = int(attack_params["attack_interval"])
    except KeyError as e:
        raise ValueError(f"Missing required attack parameter: {e}")
    except ValueError:
        raise ValueError("Invalid value in attack_params. Ensure all values are integers.")

    super().__init__(engine, round_start, round_stop, attack_interval)
    self.datamodule = engine._trainer.datamodule
    self.poisoned_percent = float(attack_params["poisoned_sample_percent"])

    # Create the appropriate strategy based on whether the attack is targeted
    if attack_params.get("targeted", False):
        target_label = int(attack_params.get("target_label") or attack_params.get("targetLabel", 4))
        target_changed_label = int(
            attack_params.get("target_changed_label") or attack_params.get("targetChangedLabel", 7)
        )
        self.strategy = TargetedLabelFlippingStrategy(target_label, target_changed_label)
    else:
        self.strategy = NonTargetedLabelFlippingStrategy()

get_malicious_dataset()

Creates a malicious dataset by flipping the labels of selected data points.

Returns:

Name Type Description
Dataset

The modified dataset with flipped labels

Source code in nebula/addons/attacks/dataset/labelflipping.py
182
183
184
185
186
187
188
189
190
191
192
193
def get_malicious_dataset(self):
    """
    Creates a malicious dataset by flipping the labels of selected data points.

    Returns:
        Dataset: The modified dataset with flipped labels
    """
    return self.strategy.flip_labels(
        self.datamodule.train_set,
        self.datamodule.train_set_indices,
        self.poisoned_percent,
    )

LabelFlippingStrategy

Bases: ABC

Abstract base class for label flipping strategies.

Source code in nebula/addons/attacks/dataset/labelflipping.py
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class LabelFlippingStrategy(ABC):
    """Abstract base class for label flipping strategies."""

    @abstractmethod
    def flip_labels(
        self,
        dataset,
        indices: list[int],
        poisoned_percent: float,
    ) -> "Dataset":
        """
        Abstract method to flip labels in the dataset.

        Args:
            dataset: The dataset to modify
            indices: List of indices to consider for flipping
            poisoned_percent: Percentage of labels to change (0-100)

        Returns:
            Modified dataset with flipped labels
        """
        pass

flip_labels(dataset, indices, poisoned_percent) abstractmethod

Abstract method to flip labels in the dataset.

Parameters:

Name Type Description Default
dataset

The dataset to modify

required
indices list[int]

List of indices to consider for flipping

required
poisoned_percent float

Percentage of labels to change (0-100)

required

Returns:

Type Description
Dataset

Modified dataset with flipped labels

Source code in nebula/addons/attacks/dataset/labelflipping.py
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@abstractmethod
def flip_labels(
    self,
    dataset,
    indices: list[int],
    poisoned_percent: float,
) -> "Dataset":
    """
    Abstract method to flip labels in the dataset.

    Args:
        dataset: The dataset to modify
        indices: List of indices to consider for flipping
        poisoned_percent: Percentage of labels to change (0-100)

    Returns:
        Modified dataset with flipped labels
    """
    pass

NonTargetedLabelFlippingStrategy

Bases: LabelFlippingStrategy

Implementation of non-targeted label flipping strategy.

Source code in nebula/addons/attacks/dataset/labelflipping.py
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
class NonTargetedLabelFlippingStrategy(LabelFlippingStrategy):
    """Implementation of non-targeted label flipping strategy."""

    def flip_labels(
        self,
        dataset,
        indices: list[int],
        poisoned_percent: float,
    ) -> "Dataset":
        """
        Flips labels randomly to different classes.

        Args:
            dataset: The dataset to modify
            indices: List of indices to consider for flipping
            poisoned_percent: Percentage of labels to change (0-100)

        Returns:
            Modified dataset with flipped labels
        """
        new_dataset = copy.deepcopy(dataset)
        if not isinstance(new_dataset.targets, np.ndarray):
            new_dataset.targets = np.array(new_dataset.targets)
        else:
            new_dataset.targets = new_dataset.targets.copy()

        num_indices = len(indices)
        num_flipped = int(poisoned_percent * num_indices / 100.0)

        if num_indices == 0 or num_flipped > num_indices:
            return new_dataset

        flipped_indices = random.sample(indices, num_flipped)
        class_list = list(set(new_dataset.targets.tolist()))

        for i in flipped_indices:
            current_label = new_dataset.targets[i]
            new_label = random.choice(class_list)
            while new_label == current_label:
                new_label = random.choice(class_list)
            new_dataset.targets[i] = new_label

        return new_dataset

flip_labels(dataset, indices, poisoned_percent)

Flips labels randomly to different classes.

Parameters:

Name Type Description Default
dataset

The dataset to modify

required
indices list[int]

List of indices to consider for flipping

required
poisoned_percent float

Percentage of labels to change (0-100)

required

Returns:

Type Description
Dataset

Modified dataset with flipped labels

Source code in nebula/addons/attacks/dataset/labelflipping.py
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def flip_labels(
    self,
    dataset,
    indices: list[int],
    poisoned_percent: float,
) -> "Dataset":
    """
    Flips labels randomly to different classes.

    Args:
        dataset: The dataset to modify
        indices: List of indices to consider for flipping
        poisoned_percent: Percentage of labels to change (0-100)

    Returns:
        Modified dataset with flipped labels
    """
    new_dataset = copy.deepcopy(dataset)
    if not isinstance(new_dataset.targets, np.ndarray):
        new_dataset.targets = np.array(new_dataset.targets)
    else:
        new_dataset.targets = new_dataset.targets.copy()

    num_indices = len(indices)
    num_flipped = int(poisoned_percent * num_indices / 100.0)

    if num_indices == 0 or num_flipped > num_indices:
        return new_dataset

    flipped_indices = random.sample(indices, num_flipped)
    class_list = list(set(new_dataset.targets.tolist()))

    for i in flipped_indices:
        current_label = new_dataset.targets[i]
        new_label = random.choice(class_list)
        while new_label == current_label:
            new_label = random.choice(class_list)
        new_dataset.targets[i] = new_label

    return new_dataset

TargetedLabelFlippingStrategy

Bases: LabelFlippingStrategy

Implementation of targeted label flipping strategy.

Source code in nebula/addons/attacks/dataset/labelflipping.py
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
class TargetedLabelFlippingStrategy(LabelFlippingStrategy):
    """Implementation of targeted label flipping strategy."""

    def __init__(self, target_label: int, target_changed_label: int):
        """
        Initialize targeted label flipping strategy.

        Args:
            target_label: The label to change
            target_changed_label: The label to change to
        """
        self.target_label = target_label
        self.target_changed_label = target_changed_label

    def flip_labels(
        self,
        dataset,
        indices: list[int],
        poisoned_percent: float,
    ) -> "Dataset":
        """
        Flips labels from target_label to target_changed_label.

        Args:
            dataset: The dataset to modify
            indices: List of indices to consider for flipping
            poisoned_percent: Percentage of labels to change (0-100)

        Returns:
            Modified dataset with flipped labels
        """
        new_dataset = copy.deepcopy(dataset)
        if not isinstance(new_dataset.targets, np.ndarray):
            new_dataset.targets = np.array(new_dataset.targets)
        else:
            new_dataset.targets = new_dataset.targets.copy()

        for i in indices:
            if int(new_dataset.targets[i]) == self.target_label:
                new_dataset.targets[i] = self.target_changed_label

        if self.target_label in new_dataset.targets:
            logging.info(f"[{self.__class__.__name__}] Target label {self.target_label} still present after flipping.")
        else:
            logging.info(
                f"[{self.__class__.__name__}] Target label {self.target_label} successfully flipped to {self.target_changed_label}."
            )

        return new_dataset

__init__(target_label, target_changed_label)

Initialize targeted label flipping strategy.

Parameters:

Name Type Description Default
target_label int

The label to change

required
target_changed_label int

The label to change to

required
Source code in nebula/addons/attacks/dataset/labelflipping.py
53
54
55
56
57
58
59
60
61
62
def __init__(self, target_label: int, target_changed_label: int):
    """
    Initialize targeted label flipping strategy.

    Args:
        target_label: The label to change
        target_changed_label: The label to change to
    """
    self.target_label = target_label
    self.target_changed_label = target_changed_label

flip_labels(dataset, indices, poisoned_percent)

Flips labels from target_label to target_changed_label.

Parameters:

Name Type Description Default
dataset

The dataset to modify

required
indices list[int]

List of indices to consider for flipping

required
poisoned_percent float

Percentage of labels to change (0-100)

required

Returns:

Type Description
Dataset

Modified dataset with flipped labels

Source code in nebula/addons/attacks/dataset/labelflipping.py
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def flip_labels(
    self,
    dataset,
    indices: list[int],
    poisoned_percent: float,
) -> "Dataset":
    """
    Flips labels from target_label to target_changed_label.

    Args:
        dataset: The dataset to modify
        indices: List of indices to consider for flipping
        poisoned_percent: Percentage of labels to change (0-100)

    Returns:
        Modified dataset with flipped labels
    """
    new_dataset = copy.deepcopy(dataset)
    if not isinstance(new_dataset.targets, np.ndarray):
        new_dataset.targets = np.array(new_dataset.targets)
    else:
        new_dataset.targets = new_dataset.targets.copy()

    for i in indices:
        if int(new_dataset.targets[i]) == self.target_label:
            new_dataset.targets[i] = self.target_changed_label

    if self.target_label in new_dataset.targets:
        logging.info(f"[{self.__class__.__name__}] Target label {self.target_label} still present after flipping.")
    else:
        logging.info(
            f"[{self.__class__.__name__}] Target label {self.target_label} successfully flipped to {self.target_changed_label}."
        )

    return new_dataset