Skip to content

Documentation for Communicationattack Module

CommunicationAttack

Bases: Attack

Source code in nebula/addons/attacks/communications/communicationattack.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 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
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
class CommunicationAttack(Attack):
    def __init__(
        self,
        engine,
        target_class,
        target_method,
        round_start_attack,
        round_stop_attack,
        attack_interval,
        decorator_args=None,
        selectivity_percentage: int = 100,
        selection_interval: int = None,
    ):
        super().__init__()
        self.engine = engine
        self.target_class = target_class
        self.target_method = target_method
        self.decorator_args = decorator_args
        self.round_start_attack = round_start_attack
        self.round_stop_attack = round_stop_attack
        self.attack_interval = attack_interval
        self.original_method = getattr(target_class, target_method, None)
        self.selectivity_percentage = selectivity_percentage
        self.selection_interval = selection_interval
        self.last_selection_round = 0
        self.targets = set()

        if not self.original_method:
            raise AttributeError(f"Method {target_method} not found in class {target_class}")

    @abstractmethod
    def decorator(self, *args):
        """Decorator that adds malicious behavior to the execution of the original method."""
        pass

    async def select_targets(self):
        """
        Selects a subset of neighboring nodes as attack targets based on the configured selectivity percentage.

        This method determines which neighboring nodes should be targeted in the current round of attack.
        If the selectivity percentage is less than 100%, it samples a subset of the currently connected direct neighbors.
        The selection behavior can be influenced by a `selection_interval`:
            - If `selection_interval` is set, target selection occurs only at rounds that are multiples of this interval.
            - If no interval is defined but no targets have been selected yet, targets are selected once.
        If the selectivity is 100%, all direct neighbors are selected as targets.

        Target addresses are retrieved from the CommunicationsManager (only direct connections).
        The number of selected targets is at least 1.

        Logs are emitted at each selection event to indicate which targets were chosen.

        Increments the internal `last_selection_round` counter after execution.

        Notes:
            - The `self.targets` attribute is updated in-place.
            - The `self.last_selection_round` attribute tracks when the selection was last performed.

    """
        if self.selectivity_percentage != 100:
            if self.selection_interval:
                if self.last_selection_round % self.selection_interval == 0:
                    logging.info("Recalculating targets...")
                    all_nodes = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
                    num_targets = max(1, int(len(all_nodes) * (self.selectivity_percentage / 100)))
                    self.targets = set(random.sample(list(all_nodes), num_targets))
            elif not self.targets:
                logging.info("Calculating targets...")
                all_nodes = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
                num_targets = max(1, int(len(all_nodes) * (self.selectivity_percentage / 100)))
                self.targets = set(random.sample(list(all_nodes), num_targets))
        else:
            logging.info("All neighbors selected as targets")
            self.targets = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)

        logging.info(f"Selected {self.selectivity_percentage}% targets from neighbors: {self.targets}")
        self.last_selection_round += 1

    async def _inject_malicious_behaviour(self):
        """Inject malicious behavior into the target method."""
        decorated_method = self.decorator(self.decorator_args)(self.original_method)

        setattr(
            self.target_class,
            self.target_method,
            types.MethodType(decorated_method, self.target_class),
        )

    async def _restore_original_behaviour(self):
        """Restore the original behavior of the target method."""
        setattr(self.target_class, self.target_method, self.original_method)

    async def attack(self):
        """Perform the attack logic based on the current round."""
        if self.engine.round not in range(self.round_start_attack, self.round_stop_attack + 1):
            pass
        elif self.engine.round == self.round_stop_attack:
            logging.info(f"[{self.__class__.__name__}] Stoping attack")
            await self._restore_original_behaviour()
        elif (self.engine.round == self.round_start_attack) or (
            (self.engine.round - self.round_start_attack) % self.attack_interval == 0
        ):
            await self.select_targets()
            logging.info(f"[{self.__class__.__name__}] Performing attack")
            await self._inject_malicious_behaviour()
        else:
            await self._restore_original_behaviour()

attack() async

Perform the attack logic based on the current round.

Source code in nebula/addons/attacks/communications/communicationattack.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def attack(self):
    """Perform the attack logic based on the current round."""
    if self.engine.round not in range(self.round_start_attack, self.round_stop_attack + 1):
        pass
    elif self.engine.round == self.round_stop_attack:
        logging.info(f"[{self.__class__.__name__}] Stoping attack")
        await self._restore_original_behaviour()
    elif (self.engine.round == self.round_start_attack) or (
        (self.engine.round - self.round_start_attack) % self.attack_interval == 0
    ):
        await self.select_targets()
        logging.info(f"[{self.__class__.__name__}] Performing attack")
        await self._inject_malicious_behaviour()
    else:
        await self._restore_original_behaviour()

decorator(*args) abstractmethod

Decorator that adds malicious behavior to the execution of the original method.

Source code in nebula/addons/attacks/communications/communicationattack.py 
41
42
43
44
@abstractmethod
def decorator(self, *args):
    """Decorator that adds malicious behavior to the execution of the original method."""
    pass

select_targets() async

Selects a subset of neighboring nodes as attack targets based on the configured selectivity percentage.

This method determines which neighboring nodes should be targeted in the current round of attack. If the selectivity percentage is less than 100%, it samples a subset of the currently connected direct neighbors. The selection behavior can be influenced by a selection_interval: - If selection_interval is set, target selection occurs only at rounds that are multiples of this interval. - If no interval is defined but no targets have been selected yet, targets are selected once. If the selectivity is 100%, all direct neighbors are selected as targets.

Target addresses are retrieved from the CommunicationsManager (only direct connections). The number of selected targets is at least 1.

Logs are emitted at each selection event to indicate which targets were chosen.

Increments the internal last_selection_round counter after execution.

Notes
  • The self.targets attribute is updated in-place.
  • The self.last_selection_round attribute tracks when the selection was last performed.
Source code in nebula/addons/attacks/communications/communicationattack.py 
46
47
48
49
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
async def select_targets(self):
    """
    Selects a subset of neighboring nodes as attack targets based on the configured selectivity percentage.

    This method determines which neighboring nodes should be targeted in the current round of attack.
    If the selectivity percentage is less than 100%, it samples a subset of the currently connected direct neighbors.
    The selection behavior can be influenced by a `selection_interval`:
        - If `selection_interval` is set, target selection occurs only at rounds that are multiples of this interval.
        - If no interval is defined but no targets have been selected yet, targets are selected once.
    If the selectivity is 100%, all direct neighbors are selected as targets.

    Target addresses are retrieved from the CommunicationsManager (only direct connections).
    The number of selected targets is at least 1.

    Logs are emitted at each selection event to indicate which targets were chosen.

    Increments the internal `last_selection_round` counter after execution.

    Notes:
        - The `self.targets` attribute is updated in-place.
        - The `self.last_selection_round` attribute tracks when the selection was last performed.

"""
    if self.selectivity_percentage != 100:
        if self.selection_interval:
            if self.last_selection_round % self.selection_interval == 0:
                logging.info("Recalculating targets...")
                all_nodes = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
                num_targets = max(1, int(len(all_nodes) * (self.selectivity_percentage / 100)))
                self.targets = set(random.sample(list(all_nodes), num_targets))
        elif not self.targets:
            logging.info("Calculating targets...")
            all_nodes = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)
            num_targets = max(1, int(len(all_nodes) * (self.selectivity_percentage / 100)))
            self.targets = set(random.sample(list(all_nodes), num_targets))
    else:
        logging.info("All neighbors selected as targets")
        self.targets = await CommunicationsManager.get_instance().get_addrs_current_connections(only_direct=True)

    logging.info(f"Selected {self.selectivity_percentage}% targets from neighbors: {self.targets}")
    self.last_selection_round += 1