wip

config.yml

@@ -1,12 +1,26 @@
 # Please, rename this file to config.yml
 
+connectors:
+  - file: connectorRIS
+    params:
+      url: wss://ris-live.ripe.net/v1/ws/
+      moreSpecific: true
+      type: UPDATE
+      host: rrc21
+      socketOptions:
+        includeRaw: false
+
+
 monitors:
-  - class: MonitorHijack
+  - file: monitorHijack
     channel: hijack
     name: basic-hijack-detection
 
 reports:
-  - class: ReportEmail
+  - file: reportFile
+    channels:
+      - hijack
+  - file: ReportEmail
     channels:
       - hijack
       - pippo
@@ -20,17 +34,6 @@ clearNotificationQueueAfterSeconds: 10 # If nothing happened in the meanwhile
 
 bufferSize: 10
 
-wsParams:
-  moreSpecific: true
-  type: UPDATE
-  host: rrc21
-  socketOptions:
-    includeRaw: false
-
-# The streaming services used to have real-time data
-# This is an array (use new lines and dashes!)
-websocketDataService: wss://ris-live.ripe.net/v1/ws/
-
 # The file containing the monitored prefixes. Please see monitored_prefixes_test.yml for an example
 # This is an array (use new lines and dashes!)
 monitoredPrefixesFiles:

connectorFactory.js

@@ -0,0 +1,36 @@
+import env from "./env";
+
+export default class ConnectorFactory {
+
+    constructor() {
+        this.disconnected = [];
+        this.connected = [];
+
+    }
+
+    loadConnectors = () => {
+        if (this.disconnected.length === 0) {
+            this.disconnected = config.reports.map(connector => new connector.class(connector.params, env));
+        }
+    };
+
+    connectConnectors = () =>
+        Promise.all(this.disconnected.map(connector => {
+            connector.connect()
+                .then(() => {
+                    this.connected.push();
+                });
+        }));
+
+    subscribeConnectors = (params) =>
+        new Promise((resolve, reject) => {
+
+            if (this.connectors.length === 0) {
+                reject(new Error("No connectors loaded"));
+            } else {
+
+                resolve(Promise.all(this.connectors.map(connector => connector.subscribe(params))));
+            }
+
+        });
+}

connectors/connector.js

@@ -0,0 +1,47 @@
+
+
+export default class Connector {
+
+    constructor(params, env){
+        this.config = env.config;
+        this.logger = env.logger;
+        this.params = params;
+        this.messageCallback = null;
+        this.connectCallback = null;
+        this.errorCallback = null;
+        this.closeCallback = null;
+    }
+
+    connect = () =>
+        new Promise((resolve, reject) => reject(new Error('The method connect has to be implemented')));
+
+
+    close = () => {
+        this.logger.log({
+            level: 'info',
+            message: 'Web socket disconnected'
+        });
+    };
+
+    subscribe = (input) => {
+        throw new Error('The method subscribe has to be implemented');
+    };
+
+    message = (message) => this.messageCallback(message);
+
+    transform = (message) => {
+        throw new Error('The method transform has to be implemented');
+    };
+
+    onConnect = (callback) => {
+        this.connectCallback = callback;
+    };
+
+    onMessage = (callback) => {
+        this.messageCallback = callback;
+    };
+
+    onClose = (callback) => {
+        this.closeCallback = callback;
+    };
+}

connectors/connectorRIS.js

@@ -0,0 +1,67 @@
+import WebSocket from "ws";
+
+export default class ConnectorRIS extends Connector{
+
+    constructor(params, env) {
+        super(params, env);
+    }
+
+    connect = () => {
+
+        // const ws = new WebSocket(this.params.url);
+        //
+        // ws.on('message', this.message);
+        //
+        // ws.on('open', () => {
+        //
+        // });
+        //
+        // ws.on('close', this.close);
+
+        new Promise((resolve, reject) => reject(new Error('The method connect has to be implemented')));
+    };
+
+    subscribe = (input) => {
+        ws.send(JSON.stringify({
+            type: "ris_subscribe",
+            data: this.params
+        }));
+    };
+
+    transform = (message) => {
+        message = message.data;
+        const components = [];
+        const announcements = message["announcements"] || [];
+        const withdrawals = message["withdrawals"] || [];
+        const peer = message["peer"];
+        const path = message["path"];
+
+        for (let announcement of announcements){
+            const nextHop = announcement["next_hop"];
+            const prefixes = announcement["prefixes"] || [];
+
+            for (let prefix of prefixes){
+                components.push({
+                    type: "announcement",
+                    prefix,
+                    peer,
+                    path,
+                    originAs: path[path.length - 1],
+                    nextHop
+                })
+            }
+        }
+
+        for (let prefix of withdrawals){
+            components.push({
+                type: "withdrawal",
+                prefix,
+                peer
+
+            })
+        }
+
+        return components;
+    };
+
+}

env.js

@@ -64,7 +64,7 @@ if (vector.env === 'prod') {
 config.monitors = (config.monitors || [])
     .map(item => {
         return {
-            class: require("./monitors/" + item.class).default,
+            class: require("./monitors/" + item.file).default,
             channel: item.channel,
             name: item.name
         };
@@ -74,12 +74,22 @@ config.reports = (config.reports || [])
     .map(item => {
 
         return {
-            class: require("./reports/" + item.class).default,
+            class: require("./reports/" + item.file).default,
             channels: item.channels
         };
 
     });
 
+config.connectors = (config.connectors || [])
+    .map(item => {
+
+        return {
+            class: require("./connectors/" + item.file).default,
+            params: item.params
+        };
+
+    });
+
 const input = new InputManager(config);
 
 vector.config = config;
@@ -89,4 +99,5 @@ vector.pubSub = pubSub;
 vector.monitors = config.monitors.map(monitor => new monitor.class(monitor.name, monitor.channel, vector));
 vector.reports = config.reports.map(report => new report.class(report.channels, vector));
 
+
 module.exports = vector;

index.js

@@ -3,6 +3,7 @@ import cluster from "cluster";
 import WebSocket from "ws";
 import sleep from "sleep";
 import Consumer from "./consumer";
+import Connector from "./connector";
 
 if (cluster.isMaster) {
 

monitors/monitor.js

@@ -47,7 +47,6 @@ export default class Monitor {
             if (id !== alert.id) {
                 throw new Error('Squash MUST receive a list of events all with the same ID.');
             }
-
         }
 
         return {
@@ -132,11 +131,6 @@ export default class Monitor {
 
         this.pubSub.publish(this.channel, alert);
 
-        this.logger.log({
-            level: 'verbose',
-            message: alert.message
-        });
-
         return alert;
     }
 

reports/reportFile.js

@@ -0,0 +1,17 @@
+import Report from "./report";
+import nodemailer from "nodemailer";
+
+export default class ReportEmail extends Report {
+
+    constructor(channels, env) {
+        super(channels, env);
+    }
+
+    report = (message, content) => {
+
+        this.logger.log({
+            level: 'verbose',
+            message: content.message
+        });
+    }
+}