added parameter to ignore collectors' peers (#1217)

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

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. 

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) => {

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
                     });
                 }

tests/config.test.yml

@@ -4,6 +4,9 @@ connectors:
   - file: connectorTest
     name: tes
     params:
+      blacklistSources:
+        - 124.0.0.3
+        - 3453
       testType: withdrawal
 
 monitors: