From ebdb96f8930b6c8196d31268e37e84f4e1374cbf Mon Sep 17 00:00:00 2001
From: Massimo Candela <massimo@us.ntt.net>
Date: Sat, 6 Jan 2024 20:05:14 +0100
Subject: [PATCH] added parameter to ignore collectors' peers (#1217)

---
 config.yml.example             |  3 +++
 docs/configuration.md          |  1 +
 src/connectors/connector.js    | 41 ++++++++++++++++++++++++++++++++--
 src/connectors/connectorRIS.js |  3 +++
 tests/config.test.yml          |  3 +++
 5 files changed, 49 insertions(+), 2 deletions(-)

diff --git a/config.yml.example b/config.yml.example
index c228553..6897f07 100644
--- a/config.yml.example
+++ b/config.yml.example
@@ -2,6 +2,9 @@ connectors:
   - file: connectorRIS
     name: ris
     params:
+      blacklistSources:
+        - 123.0.0.1/24
+        - 3453
       carefulSubscription: true
       url: ws://ris-live.ripe.net/v1/ws/
       perMessageDeflate: true
diff --git a/docs/configuration.md b/docs/configuration.md
index f01ae45..9890e7b 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -131,6 +131,7 @@ Parameters for this connector module:
 |carefulSubscription| If this parameter is set to true (default), the RIS server will stream only the data related to our prefix. This is an advanced parameter useful only for research purposes. |
 |perMessageDeflate| Enable gzip compression on the connection. |
 |disableCanary| ConnectorRIS automatically receives BGP announcements about [RIS beacons](https://www.ripe.net/analyse/internet-measurements/routing-information-service-ris/current-ris-routing-beacons). RIS beacons are prefixes periodically announced and withdrawn for research purposes. BGPalerter uses these beacons to detect faulty data streams. By setting this parameter to true, you will disable such a check.|
+|blacklistSources| A list of prefixes, IPs, and AS numbers of blacklisted collectors' peers. Data coming from these collectors' peers will be ignored.|
 
 #### connectorRISDump
 It connects to the RIPEstat's BGPlay API and retrieves a RIS dump about the monitored resources. The retrieved dump is 2 hours old, due to limitations on the API side. 
diff --git a/src/connectors/connector.js b/src/connectors/connector.js
index 8272d06..73b82b6 100644
--- a/src/connectors/connector.js
+++ b/src/connectors/connector.js
@@ -33,6 +33,7 @@
 
 import axios from "redaxios";
 import axiosEnrich from "../utils/axiosEnrich";
+import ipUtils from 'ip-sub';
 
 export default class Connector {
 
@@ -48,12 +49,42 @@ export default class Connector {
         this.errorCallback = null;
         this.disconnectCallback = null;
 
-
         this.axios = axiosEnrich(axios,
             (!this.params.noProxy && env.agent) ? env.agent : null,
             `${env.clientId}/${env.version}`);
     }
 
+    _parseFilters = (callback) => {
+        const {blacklistSources=[]} = this.params;
+
+        if (blacklistSources) {
+            const filters = {
+                asns: blacklistSources.filter(i => Number.isInteger(i)),
+                prefixes: blacklistSources.filter(i => ipUtils.isValidPrefix(i) || ipUtils.isValidIP(i)).map(i => ipUtils.toPrefix(i)),
+            }
+
+            const generateCallback = (filters, callback) => {
+                return (message) => {
+                    const {data} = message;
+
+                    if (data && (data.peerAS || data.peer)) {
+                        const messagePeer = ipUtils.toPrefix(data.peer);
+                        if (!filters.prefixes.some(prefix => ipUtils.isEqualPrefix(prefix, messagePeer) || ipUtils.isSubnet(prefix, messagePeer))
+                            && !filters.asns.includes(data.peerAS)) {
+                            return callback(message);
+                        }
+                    } else {
+                        return callback(message);
+                    }
+                }
+            }
+
+            return generateCallback(filters, callback);
+        } else {
+            return null;
+        }
+    }
+
     connect = () =>
         new Promise((resolve, reject) => reject(new Error('The method connect MUST be implemented')));
 
@@ -93,7 +124,13 @@ export default class Connector {
     };
 
     onMessage = (callback) => {
-        this.messageCallback = callback;
+        const filterCallback = this._parseFilters(callback);
+
+        if (filterCallback) {
+            this.messageCallback = filterCallback;
+        } else {
+            this.messageCallback = callback;
+        }
     };
 
     onError = (callback) => {
diff --git a/src/connectors/connectorRIS.js b/src/connectors/connectorRIS.js
index b9b89a8..998d808 100644
--- a/src/connectors/connectorRIS.js
+++ b/src/connectors/connectorRIS.js
@@ -386,6 +386,7 @@ export default class ConnectorRIS extends Connector {
                 const possibleRIS = message["possibleRIS"] || false;
                 const withdrawals = (message["withdrawals"] || []).filter(prefix => acceptPrefix(prefix, possibleRIS));
                 const peer = message["peer"];
+                const peerAS = message["peer_asn"];
                 const communities = message["community"] || [];
                 const timestamp = message["timestamp"] * 1000;
                 let path, originAS;
@@ -411,6 +412,7 @@ export default class ConnectorRIS extends Connector {
                                     type: "announcement",
                                     prefix,
                                     peer,
+                                    peerAS,
                                     path,
                                     originAS,
                                     nextHop,
@@ -428,6 +430,7 @@ export default class ConnectorRIS extends Connector {
                         type: "withdrawal",
                         prefix,
                         peer,
+                        peerAS,
                         timestamp
                     });
                 }
diff --git a/tests/config.test.yml b/tests/config.test.yml
index b3ffa4c..ca5aad5 100644
--- a/tests/config.test.yml
+++ b/tests/config.test.yml
@@ -4,6 +4,9 @@ connectors:
   - file: connectorTest
     name: tes
     params:
+      blacklistSources:
+        - 124.0.0.3
+        - 3453
       testType: withdrawal
 
 monitors: