Skip to content

Documentation for Blacklist Module

BlackList

Manages a dynamic blacklist and a list of recently disconnected nodes in a distributed system.

The blacklist tracks nodes that are temporarily excluded from communication or interaction due to malicious behavior or disconnection events. Nodes remain blacklisted for a fixed period defined by max_time_listed.

The recently disconnected list tracks peers that were recently disconnected and may need to be temporarily avoided.

Key features: - Asynchronous locks for concurrent safety. - Periodic cleaning of the blacklist via a background coroutine. - Integration with an event manager to publish changes.

Source code in nebula/core/network/blacklist.py 
 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
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
144
145
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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
class BlackList:
    """
    Manages a dynamic blacklist and a list of recently disconnected nodes in a distributed system.

    The blacklist tracks nodes that are temporarily excluded from communication or interaction due to malicious behavior
    or disconnection events. Nodes remain blacklisted for a fixed period defined by `max_time_listed`.

    The recently disconnected list tracks peers that were recently disconnected and may need to be temporarily avoided.

    Key features:
    - Asynchronous locks for concurrent safety.
    - Periodic cleaning of the blacklist via a background coroutine.
    - Integration with an event manager to publish changes.
    """

    def __init__(self, max_time_listed=BLACKLIST_EXPIRATION_TIME):
        """
        Initialize the BlackList with the specified expiration time.

        Args:
            max_time_listed (int): Maximum time in seconds for nodes to remain blacklisted.
        """
        self._max_time_listed = max_time_listed
        self._blacklisted_nodes = {}
        self._recently_disconnected = set()
        self._blacklisted_nodes_lock = Locker("blacklisted_nodes_lock", async_lock=True)
        self._recently_disconnected_lock = Locker("recently_disconnected_lock", async_lock=True)
        self._running = asyncio.Event()
        self._background_tasks = []  # Track background tasks

    async def apply_restrictions(self, nodes) -> set | None:
        """
        Applies both blacklist and recently disconnected restrictions to a given set of nodes.

        Args:
            nodes (set): Set of peer node addresses.

        Returns:
            set | None: Filtered set excluding blacklisted and recently disconnected nodes, or None if input is empty.
        """
        nodes_allowed = await self.verify_allowed_nodes(nodes)
        # logging.info(f"nodes allowed after appliying blacklist restricttions: {nodes_allowed}")
        if nodes_allowed:
            nodes_allowed = await self.verify_not_recently_disc(nodes_allowed)
            # logging.info(f"nodes allowed after seen recently disconnection restrictions: {nodes_allowed}")
        return nodes_allowed

    async def clear_restrictions(self):
        """
        Clears both the blacklist and the recently disconnected list.
        """
        await self.clear_blacklist()
        await self.clear_recently_disconected()

    """                                                     ##############################
                                                            #          BLACKLIST         #
                                                            ##############################
    """

    async def add_to_blacklist(self, addr):
        """
        Adds a node to the blacklist and starts the cleaner task if not already running.

        Args:
            addr (str): Address of the node to blacklist.
        """
        logging.info(f"Update blackList | addr listed: {addr}")
        await self._blacklisted_nodes_lock.acquire_async()
        expiration_time = time.time()
        self._blacklisted_nodes[addr] = expiration_time
        if not self._running.is_set():
            self._running.set()
            asyncio.create_task(self._start_blacklist_cleaner())
        await self._blacklisted_nodes_lock.release_async()
        nbe = NodeBlacklistedEvent(addr, blacklisted=True)
        event_manager = EventManager.get_instance()
        if event_manager is not None:
            asyncio.create_task(event_manager.publish_node_event(nbe))

    async def get_blacklist(self) -> set:
        """
        Adds a node to the blacklist and starts the cleaner task if not already running.

        Args:
            addr (str): Address of the node to blacklist.
        """
        bl = None
        await self._blacklisted_nodes_lock.acquire_async()
        if self._blacklisted_nodes:
            bl = set(self._blacklisted_nodes.keys())
        await self._blacklisted_nodes_lock.release_async()
        return bl or set()

    async def clear_blacklist(self):
        """
        Clears the blacklist entirely.
        """
        await self._blacklisted_nodes_lock.acquire_async()
        logging.info("🧹 Removing nodes from blacklist")
        self._blacklisted_nodes.clear()
        await self._blacklisted_nodes_lock.release_async()

    async def _start_blacklist_cleaner(self):
        """
        Background task that periodically removes expired entries from the blacklist.
        """
        while self._running.is_set():
            await self._blacklist_clean()
            await self._blacklist_cleaner_wait()

    async def _blacklist_clean(self):
        """
        Removes nodes from the blacklist whose expiration time has passed.
        """
        await self._blacklisted_nodes_lock.acquire_async()
        logging.info("BlackList cleaner has waken up")
        now = time.time()
        new_bl = {}

        for addr, timer in self._blacklisted_nodes.items():
            if timer + self._max_time_listed >= now:
                new_bl[addr] = timer
            else:
                logging.info(f"Removing addr{addr} from blacklisted nodes...")

        self._blacklisted_nodes = new_bl
        if not new_bl:
            self._running.clear()
        await self._blacklisted_nodes_lock.release_async()

    async def _blacklist_cleaner_wait(self):
        """
        Waits for the blacklist cleaner delay duration.
        """
        try:
            await asyncio.sleep(self._max_time_listed)
        except TimeoutError:
            pass

    async def node_in_blacklist(self, addr):
        """
        Checks whether a given address is currently blacklisted.

        Args:
            addr (str): Node address.

        Returns:
            bool: True if the node is blacklisted, False otherwise.
        """
        blacklisted = False
        await self._blacklisted_nodes_lock.acquire_async()
        if self._blacklisted_nodes:
            blacklisted = addr in self._blacklisted_nodes.keys()
        await self._blacklisted_nodes_lock.release_async()
        return blacklisted

    async def verify_allowed_nodes(self, nodes: set) -> set | None:
        """
        Filters out blacklisted nodes from the given set.

        Args:
            nodes (set): Set of node addresses to check.

        Returns:
            set | None: Nodes not in the blacklist, or None if input is empty.
        """
        if not nodes:
            return None
        nodes_not_listed = nodes
        await self._blacklisted_nodes_lock.acquire_async()
        blacklist = self._blacklisted_nodes
        if blacklist:
            nodes_not_listed = nodes.difference(blacklist)
        await self._blacklisted_nodes_lock.release_async()
        return nodes_not_listed

    """                                                     ##############################
                                                            #    RECENTLY DISCONNECTED   #
                                                            ##############################
    """

    async def add_recently_disconnected(self, addr):
        """
        Marks a node as recently disconnected and schedules its expiration.

        Args:
            addr (str): Address of the disconnected node.
        """
        logging.info(f"Recently disconnected from: {addr}")
        await self._recently_disconnected_lock.acquire_async()
        self._recently_disconnected.add(addr)
        await self._recently_disconnected_lock.release_async()
        task = asyncio.create_task(self._remove_recently_disc(addr), name=f"BlackList_remove_recently_{addr}")
        self._background_tasks.append(task)
        nbe = NodeBlacklistedEvent(addr)
        event_manager = EventManager.get_instance()
        if event_manager is not None:
            asyncio.create_task(event_manager.publish_node_event(nbe))

    async def clear_recently_disconected(self):
        """
        Clears the list of recently disconnected nodes.
        """
        await self._recently_disconnected_lock.acquire_async()
        logging.info("🧹 Removing nodes from Recently Disconencted list")
        self._recently_disconnected.clear()
        await self._recently_disconnected_lock.release_async()

    async def get_recently_disconnected(self):
        """
        Retrieves a copy of the recently disconnected nodes.

        Returns:
            set: Addresses of recently disconnected nodes.
        """
        rd = None
        await self._recently_disconnected_lock.acquire_async()
        rd = self._recently_disconnected.copy()
        await self._recently_disconnected_lock.release_async()
        return rd

    async def _remove_recently_disc(self, addr):
        """
        Waits for the expiration time and then removes the node from the recently disconnected list.

        Args:
            addr (str): Address to remove after expiration.
        """
        await asyncio.sleep(RECENTLY_DISCONNECTED_EXPIRE_TIME)
        await self._recently_disconnected_lock.acquire_async()
        self._recently_disconnected.discard(addr)
        logging.info(f"Recently disconnection timeout expired for souce: {addr}")
        await self._recently_disconnected_lock.release_async()

    async def verify_not_recently_disc(self, nodes: set) -> set | None:
        """
        Filters out recently disconnected nodes from the given set.

        Args:
            nodes (set): Set of node addresses to filter.

        Returns:
            set | None: Set of nodes not recently disconnected, or None if input is empty.
        """
        if not nodes:
            return None
        nodes_not_listed = nodes
        await self._recently_disconnected_lock.acquire_async()
        rec_disc = self._recently_disconnected
        # logging.info(f"recently disconencted nodes: {rec_disc}")
        if rec_disc:
            nodes_not_listed = nodes.difference(rec_disc)
        await self._recently_disconnected_lock.release_async()
        return nodes_not_listed

    async def stop(self):
        """
        Stop the BlackList by clearing all data and stopping background tasks.
        """
        logging.info("🛑  Stopping BlackList...")

        # Stop the background cleaner
        self._running.clear()

        # Cancel all background tasks
        if self._background_tasks:
            logging.info(f"🛑  Cancelling {len(self._background_tasks)} background tasks...")
            for task in self._background_tasks:
                if not task.done():
                    task.cancel()
                    try:
                        await task
                    except asyncio.CancelledError:
                        pass
            self._background_tasks.clear()
            logging.info("🛑  All background tasks cancelled")

        # Clear all data
        try:
            async with self._blacklisted_nodes_lock:
                self._blacklisted_nodes.clear()
        except Exception as e:
            logging.warning(f"Error clearing blacklist: {e}")

        try:
            async with self._recently_disconnected_lock:
                self._recently_disconnected.clear()
        except Exception as e:
            logging.warning(f"Error clearing recently disconnected: {e}")

        logging.info("✅  BlackList stopped successfully")

__init__(max_time_listed=BLACKLIST_EXPIRATION_TIME)

Initialize the BlackList with the specified expiration time.

Parameters:

Name Type Description Default
max_time_listed int

Maximum time in seconds for nodes to remain blacklisted.

 BLACKLIST_EXPIRATION_TIME
Source code in nebula/core/network/blacklist.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def __init__(self, max_time_listed=BLACKLIST_EXPIRATION_TIME):
    """
    Initialize the BlackList with the specified expiration time.

    Args:
        max_time_listed (int): Maximum time in seconds for nodes to remain blacklisted.
    """
    self._max_time_listed = max_time_listed
    self._blacklisted_nodes = {}
    self._recently_disconnected = set()
    self._blacklisted_nodes_lock = Locker("blacklisted_nodes_lock", async_lock=True)
    self._recently_disconnected_lock = Locker("recently_disconnected_lock", async_lock=True)
    self._running = asyncio.Event()
    self._background_tasks = []  # Track background tasks

add_recently_disconnected(addr) async

Marks a node as recently disconnected and schedules its expiration.

Parameters:

Name Type Description Default
addr str

Address of the disconnected node.

 required
Source code in nebula/core/network/blacklist.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
async def add_recently_disconnected(self, addr):
    """
    Marks a node as recently disconnected and schedules its expiration.

    Args:
        addr (str): Address of the disconnected node.
    """
    logging.info(f"Recently disconnected from: {addr}")
    await self._recently_disconnected_lock.acquire_async()
    self._recently_disconnected.add(addr)
    await self._recently_disconnected_lock.release_async()
    task = asyncio.create_task(self._remove_recently_disc(addr), name=f"BlackList_remove_recently_{addr}")
    self._background_tasks.append(task)
    nbe = NodeBlacklistedEvent(addr)
    event_manager = EventManager.get_instance()
    if event_manager is not None:
        asyncio.create_task(event_manager.publish_node_event(nbe))

add_to_blacklist(addr) async

Adds a node to the blacklist and starts the cleaner task if not already running.

Parameters:

Name Type Description Default
addr str

Address of the node to blacklist.

 required
Source code in nebula/core/network/blacklist.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def add_to_blacklist(self, addr):
    """
    Adds a node to the blacklist and starts the cleaner task if not already running.

    Args:
        addr (str): Address of the node to blacklist.
    """
    logging.info(f"Update blackList | addr listed: {addr}")
    await self._blacklisted_nodes_lock.acquire_async()
    expiration_time = time.time()
    self._blacklisted_nodes[addr] = expiration_time
    if not self._running.is_set():
        self._running.set()
        asyncio.create_task(self._start_blacklist_cleaner())
    await self._blacklisted_nodes_lock.release_async()
    nbe = NodeBlacklistedEvent(addr, blacklisted=True)
    event_manager = EventManager.get_instance()
    if event_manager is not None:
        asyncio.create_task(event_manager.publish_node_event(nbe))

apply_restrictions(nodes) async

Applies both blacklist and recently disconnected restrictions to a given set of nodes.

Parameters:

Name Type Description Default
nodes set

Set of peer node addresses.

 required

Returns:

Type Description
set | None

set | None: Filtered set excluding blacklisted and recently disconnected nodes, or None if input is empty.
Source code in nebula/core/network/blacklist.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def apply_restrictions(self, nodes) -> set | None:
    """
    Applies both blacklist and recently disconnected restrictions to a given set of nodes.

    Args:
        nodes (set): Set of peer node addresses.

    Returns:
        set | None: Filtered set excluding blacklisted and recently disconnected nodes, or None if input is empty.
    """
    nodes_allowed = await self.verify_allowed_nodes(nodes)
    # logging.info(f"nodes allowed after appliying blacklist restricttions: {nodes_allowed}")
    if nodes_allowed:
        nodes_allowed = await self.verify_not_recently_disc(nodes_allowed)
        # logging.info(f"nodes allowed after seen recently disconnection restrictions: {nodes_allowed}")
    return nodes_allowed

clear_blacklist() async

Clears the blacklist entirely.

Source code in nebula/core/network/blacklist.py 
106
107
108
109
110
111
112
113
async def clear_blacklist(self):
    """
    Clears the blacklist entirely.
    """
    await self._blacklisted_nodes_lock.acquire_async()
    logging.info("🧹 Removing nodes from blacklist")
    self._blacklisted_nodes.clear()
    await self._blacklisted_nodes_lock.release_async()

clear_recently_disconected() async

Clears the list of recently disconnected nodes.

Source code in nebula/core/network/blacklist.py 
212
213
214
215
216
217
218
219
async def clear_recently_disconected(self):
    """
    Clears the list of recently disconnected nodes.
    """
    await self._recently_disconnected_lock.acquire_async()
    logging.info("🧹 Removing nodes from Recently Disconencted list")
    self._recently_disconnected.clear()
    await self._recently_disconnected_lock.release_async()

clear_restrictions() async

Clears both the blacklist and the recently disconnected list.

Source code in nebula/core/network/blacklist.py 
60
61
62
63
64
65
async def clear_restrictions(self):
    """
    Clears both the blacklist and the recently disconnected list.
    """
    await self.clear_blacklist()
    await self.clear_recently_disconected()

get_blacklist() async

Adds a node to the blacklist and starts the cleaner task if not already running.

Parameters:

Name Type Description Default
addr str

Address of the node to blacklist.

 required
Source code in nebula/core/network/blacklist.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def get_blacklist(self) -> set:
    """
    Adds a node to the blacklist and starts the cleaner task if not already running.

    Args:
        addr (str): Address of the node to blacklist.
    """
    bl = None
    await self._blacklisted_nodes_lock.acquire_async()
    if self._blacklisted_nodes:
        bl = set(self._blacklisted_nodes.keys())
    await self._blacklisted_nodes_lock.release_async()
    return bl or set()

get_recently_disconnected() async

Retrieves a copy of the recently disconnected nodes.

Returns:

Name Type Description
set

Addresses of recently disconnected nodes.
Source code in nebula/core/network/blacklist.py 
221
222
223
224
225
226
227
228
229
230
231
232
async def get_recently_disconnected(self):
    """
    Retrieves a copy of the recently disconnected nodes.

    Returns:
        set: Addresses of recently disconnected nodes.
    """
    rd = None
    await self._recently_disconnected_lock.acquire_async()
    rd = self._recently_disconnected.copy()
    await self._recently_disconnected_lock.release_async()
    return rd

node_in_blacklist(addr) async

Checks whether a given address is currently blacklisted.

Parameters:

Name Type Description Default
addr str

Node address.

 required

Returns:

Name Type Description
bool

True if the node is blacklisted, False otherwise.
Source code in nebula/core/network/blacklist.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
async def node_in_blacklist(self, addr):
    """
    Checks whether a given address is currently blacklisted.

    Args:
        addr (str): Node address.

    Returns:
        bool: True if the node is blacklisted, False otherwise.
    """
    blacklisted = False
    await self._blacklisted_nodes_lock.acquire_async()
    if self._blacklisted_nodes:
        blacklisted = addr in self._blacklisted_nodes.keys()
    await self._blacklisted_nodes_lock.release_async()
    return blacklisted

stop() async

Stop the BlackList by clearing all data and stopping background tasks.

Source code in nebula/core/network/blacklist.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
async def stop(self):
    """
    Stop the BlackList by clearing all data and stopping background tasks.
    """
    logging.info("🛑  Stopping BlackList...")

    # Stop the background cleaner
    self._running.clear()

    # Cancel all background tasks
    if self._background_tasks:
        logging.info(f"🛑  Cancelling {len(self._background_tasks)} background tasks...")
        for task in self._background_tasks:
            if not task.done():
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass
        self._background_tasks.clear()
        logging.info("🛑  All background tasks cancelled")

    # Clear all data
    try:
        async with self._blacklisted_nodes_lock:
            self._blacklisted_nodes.clear()
    except Exception as e:
        logging.warning(f"Error clearing blacklist: {e}")

    try:
        async with self._recently_disconnected_lock:
            self._recently_disconnected.clear()
    except Exception as e:
        logging.warning(f"Error clearing recently disconnected: {e}")

    logging.info("✅  BlackList stopped successfully")

verify_allowed_nodes(nodes) async

Filters out blacklisted nodes from the given set.

Parameters:

Name Type Description Default
nodes set

Set of node addresses to check.

 required

Returns:

Type Description
set | None

set | None: Nodes not in the blacklist, or None if input is empty.
Source code in nebula/core/network/blacklist.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def verify_allowed_nodes(self, nodes: set) -> set | None:
    """
    Filters out blacklisted nodes from the given set.

    Args:
        nodes (set): Set of node addresses to check.

    Returns:
        set | None: Nodes not in the blacklist, or None if input is empty.
    """
    if not nodes:
        return None
    nodes_not_listed = nodes
    await self._blacklisted_nodes_lock.acquire_async()
    blacklist = self._blacklisted_nodes
    if blacklist:
        nodes_not_listed = nodes.difference(blacklist)
    await self._blacklisted_nodes_lock.release_async()
    return nodes_not_listed

verify_not_recently_disc(nodes) async

Filters out recently disconnected nodes from the given set.

Parameters:

Name Type Description Default
nodes set

Set of node addresses to filter.

 required

Returns:

Type Description
set | None

set | None: Set of nodes not recently disconnected, or None if input is empty.
Source code in nebula/core/network/blacklist.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
async def verify_not_recently_disc(self, nodes: set) -> set | None:
    """
    Filters out recently disconnected nodes from the given set.

    Args:
        nodes (set): Set of node addresses to filter.

    Returns:
        set | None: Set of nodes not recently disconnected, or None if input is empty.
    """
    if not nodes:
        return None
    nodes_not_listed = nodes
    await self._recently_disconnected_lock.acquire_async()
    rec_disc = self._recently_disconnected
    # logging.info(f"recently disconencted nodes: {rec_disc}")
    if rec_disc:
        nodes_not_listed = nodes.difference(rec_disc)
    await self._recently_disconnected_lock.release_async()
    return nodes_not_listed