Add BGP finite-state machine

Implement BGPFsm.

4 files changed, 586 insertions, 28 deletions

diff --git a/src/main/java/com/lumaserv/bgp/BGPFsm.java b/src/main/java/com/lumaserv/bgp/BGPFsm.java
new file mode 100644
index 0000000..b38ce6a
--- /dev/null
+++ b/src/main/java/com/lumaserv/bgp/BGPFsm.java
@@ -0,0 +1,446 @@
+package com.lumaserv.bgp;
+
+import com.lumaserv.bgp.protocol.BGPPacket;
+import com.lumaserv.bgp.protocol.message.BGPNotification;
+import com.lumaserv.bgp.protocol.message.BGPOpen;
+import com.lumaserv.bgp.protocol.message.BGPUpdate;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.IOException;
+import java.util.Timer;
+import java.util.TimerTask;
+
+public class BGPFsm {
+    final int CONFIG_HOLD_TIME = 80;  // FIXME use from configuration
+    final byte CONFIG_VERSION = 4;
+
+    public final State IDLE;
+    public final State CONNECT;
+    public final State ACTIVE;
+    public final State OPEN_SENT;
+    public final State OPEN_CONFIRM;
+    public final State ESTABLISHED;
+
+    @Getter
+    @Setter
+    State currentState;
+    boolean delayOpen;
+    int holdTime;
+    int connectRetryCounter;
+    int connectRetryTime = 5; //120;
+    @Setter
+    BGPSession session;
+    Timer keepAliveTimer;
+    TimerTask keepAliveTimerTask;
+    Timer holdTimer;
+    TimerTask holdTimerTask;
+    Timer connectRetryTimer;
+    TimerTask connectRetryTimerTask;
+
+    class State {
+        void setState(State next) {
+            System.out.println("State transition: " + BGPFsm.this + ":" + this + " -> " + next);
+            // if (next == IDLE)
+            //     throw new RuntimeException();
+            currentState = next;
+        }
+        void manualStart() { throw new RuntimeException("Not implemented: " + this); }
+        void automaticStart() { throw new RuntimeException("Not implemented: " + this); }
+        void manualStartPassive() { throw new RuntimeException("Not implemented: " + this); }
+        void automaticStartPassive() { throw new RuntimeException("Not implemented: " + this); }
+        void manualStop() { throw new RuntimeException("Not implemented: " + this); }
+        void automaticStop() { throw new RuntimeException("Not implemented: " + this); }
+        void holdTimerExpires() { throw new RuntimeException("Not implemented: " + this); }
+        void keepAliveTimerExpires() { throw new RuntimeException("Not implemented: " + this); }
+        void connectRetryTimerExpires() { throw new RuntimeException("Not implemented: " + this); }
+        void tcpCRAcked() { throw new RuntimeException("Not implemented: " + this); }
+        void tcpConnectionConfirmed() { throw new RuntimeException("Not implemented: " + this); }
+        void tcpConnectionFails() { throw new RuntimeException("Not implemented: " + this); }
+        void openMsg(BGPOpen packet)  { throw new RuntimeException("Not implemented: " + this); }
+        void keepAliveMsg() { throw new RuntimeException("Not implemented: " + this); }
+        void updateMsg(BGPUpdate packet) { throw new RuntimeException("Not implemented: " + this); }
+        void notifMsgVerErr(BGPNotification msg) { throw new RuntimeException("Not implemented: " + this); }
+        void notifMsg(BGPNotification msg) { throw new RuntimeException("Not implemented: " + this); }
+    }
+
+    class Idle extends State {
+        protected void start() {
+            // - initializes all BGP resources for the peer connection,
+            setConnectRetryTimer(connectRetryTime);
+            setState(CONNECT);
+            // - listens for a connection that may be initiated by the remote
+            //   BGP peer, and
+            connectRetryCounter = 0;
+            session.startOutgoing();
+        }
+
+        @Override
+        void manualStart() {
+            start();
+        }
+
+        @Override
+        void automaticStart() {
+            start();
+        }
+
+        @Override
+        void manualStop() {
+            // ignore
+        }
+
+        @Override
+        void automaticStop() {
+            // ignore
+        }
+
+        protected void startPassive() {
+            // - initializes all BGP resources,
+            // - sets the ConnectRetryCounter to zero,
+            setConnectRetryTimer(connectRetryTime);
+            // - listens for a connection that may be initiated by the remote
+            //   peer, and
+            setState(ACTIVE);
+            session.startIncoming();
+        }
+
+        @Override
+        void manualStartPassive() {
+            startPassive();
+        }
+
+        @Override
+        void automaticStartPassive() {
+            startPassive();
+        }
+
+        // TODO implement DampPeerOscillations related events?
+
+        // TODO
+        // Any other event (Events 9-12, 15-28) received in the Idle state
+        // does not cause change in the state of the local system.
+        @Override
+        void connectRetryTimerExpires() {} // Ignore event 9
+        @Override
+        void holdTimerExpires() {} // Ignore event 10
+        @Override
+        void keepAliveTimerExpires() {} // Ignore event 11
+        // Event 12 not implemented
+        // Event 15 not implemented
+        @Override
+        void tcpCRAcked() {} // Ignore event 16
+        @Override
+        void tcpConnectionConfirmed() {} // Ignore event 17
+        @Override
+        void tcpConnectionFails() {} // Ignore event 18
+        @Override
+        void openMsg(BGPOpen msg) {} // Ignore event 19
+        // Event 20 not implemented
+        //@Override
+        // TODO void bgpHeaderErr() {} // Ignore event 21
+        @Override
+        public String toString() { return "Idle"; }
+    }
+
+    class Connect extends State {
+        // TODO ignore start events 1, 3-7
+
+        @Override
+        void manualStop() {
+            session.dropConnection();
+            // - releases all BGP resources,
+            connectRetryCounter = 0;
+            setConnectRetryTimer(0);
+            setState(IDLE);
+        }
+
+        @Override
+        void connectRetryTimerExpires() {
+            System.out.println("connectRetryTimerExpires");
+            session.dropConnection();
+            setConnectRetryTimer(connectRetryTime);
+            // - stops the DelayOpenTimer and resets the timer to zero,
+            session.retryConnection();
+            // - continues to listen for a connection that may be initiated by
+            //   the remote BGP peer, and
+        }
+
+        // TODO DelayOpenTimer_Expires event (Event 12)
+        // TODO TcpConnection_Valid event (Event 14)
+        // TODO Tcp_CR_Invalid event (Event 15)
+
+        @Override
+        void tcpCRAcked() { // Or incoming tcpConnectionConfirmed
+            if (delayOpen) {
+                setConnectRetryTimer(0);
+                // - sets the DelayOpenTimer to the initial value, and
+            } else {
+                setConnectRetryTimer(0);
+                // - completes BGP initialization
+
+                BGPSessionConfiguration config = session.getConfiguration();
+                BGPOpen request = new BGPOpen()
+                    .setAsn(config.getLocalAs())
+                    .setHoldTime(CONFIG_HOLD_TIME)
+                    .setVersion(CONFIG_VERSION)
+                    .setIdentifier(config.getLocalIdentifier());
+                try {
+                    session.sendOpen(request); 
+                    setHoldTimer(4 * 60);
+                    setState(OPEN_SENT);
+                } catch (IOException ex) {
+                    tcpConnectionFails();
+                }
+            }
+        }
+
+        // FIXME
+        @Override
+        void tcpConnectionFails() {
+            setConnectRetryTimer(0);
+            // - releases all BGP resources,
+            // - (optionally) performs peer oscillation damping if the
+            //   DampPeerOscillations attribute is set to TRUE, and
+            session.dropConnection();
+            connectRetryCounter++;
+            setState(IDLE);
+        }
+
+        // TODO
+        // State notifMsgVerErr() {
+        //     // - sets the ConnectRetryTimer to zero,
+        //     // - releases all BGP resources,
+        //     // - drops the TCP connection, and
+        //     return IDLE;
+        // }
+
+        // State bgpOpen() {
+        //     return
+        // }
+
+        @Override
+        public String toString() { return "Connect"; }
+    }
+
+    class Active extends State {
+        @Override
+        void tcpConnectionConfirmed() {
+            if (delayOpen) {
+                setConnectRetryTimer(0);
+                // - sets the DelayOpenTimer to the initial value
+                //   (DelayOpenTime), and
+                // - stays in the Active state.
+            } else {
+                setConnectRetryTimer(0);
+                // - completes the BGP initialization,
+
+                BGPSessionConfiguration config = session.getConfiguration();
+                BGPOpen request = new BGPOpen()
+                    .setAsn(config.getLocalAs())
+                    .setHoldTime(CONFIG_HOLD_TIME)
+                    .setVersion(CONFIG_VERSION)
+                    .setIdentifier(config.getLocalIdentifier());
+                try {
+                    session.sendOpen(request);
+                    setHoldTimer(4 * 60);
+                    setState(OPEN_SENT);
+                } catch (IOException ex) {
+                    tcpConnectionFails();
+                }
+            }
+        }
+
+        @Override
+        public String toString() { return "Active"; }
+    }
+
+    class OpenSent extends State {
+        @Override
+        void openMsg(BGPOpen msg) {
+            if(session.getConfiguration().getRemoteAs() == msg.getAsn()) {
+                // - resets the DelayOpenTimer to zero,
+                // - sets the BGP ConnectRetryTimer to zero,
+                session.keepAlive();
+                if (msg.getHoldTime() < CONFIG_HOLD_TIME)
+                    holdTime = msg.getHoldTime();
+                else
+                    holdTime = CONFIG_HOLD_TIME;
+                setKeepAliveTimer();
+                setHoldTimer(holdTime);
+                setState(OPEN_CONFIRM);
+            } else {
+                System.out.println("Bad asn");
+                //  - sends a NOTIFICATION message with the appropriate error code,
+                // - sets the ConnectRetryTimer to zero,
+                // - releases all BGP resources,
+                session.dropConnection();
+                connectRetryCounter++;
+                // - (optionally) performs peer oscillation damping if the
+                //   DampPeerOscillations attribute is TRUE, and
+
+                setState(IDLE);
+            }
+        }
+
+        @Override
+        public String toString() { return "OpenSent"; }
+    }
+
+    class OpenConfirm extends State {
+        @Override
+        void keepAliveMsg() {
+            // - restarts the HoldTimer and
+            setState(ESTABLISHED);
+            session.getConfiguration().getListener().onOpen(session);
+        }
+
+        // TODO implement keepAliveTimerExpires and holdTimerExpires
+
+        @Override
+        void tcpConnectionFails() {
+            tcpConnectionFailsOrNotifMsg();
+        }
+
+        @Override
+        void notifMsg(BGPNotification msg) {
+            tcpConnectionFailsOrNotifMsg();
+        }
+
+        protected void tcpConnectionFailsOrNotifMsg() {
+            setConnectRetryTimer(0);
+            // - releases all BGP resources,
+            session.dropConnection();
+            // - increments the ConnectRetryCounter by 1,
+            // - (optionally) performs peer oscillation damping if the
+            //   DampPeerOscillations attribute is set to TRUE, and
+            setState(IDLE);
+        }
+
+        @Override
+        public String toString() { return "OpenConfirm"; }
+    }
+
+    class Established extends State {
+        @Override
+        void keepAliveMsg() {
+            setHoldTimer(holdTime);
+        }
+
+        @Override
+        void updateMsg(BGPUpdate packet) {
+            // - processes the message,
+            setHoldTimer(holdTime);
+        }
+
+        @Override
+        void holdTimerExpires() {
+            // - sends a NOTIFICATION message with the Error Code Hold Timer
+            //   Expired,
+            setConnectRetryTimer(0);
+            // - releases all BGP resources,
+            session.dropConnection();
+            connectRetryCounter++;
+            // - (optionally) performs peer oscillation damping if the
+            //   DampPeerOscillations attribute is set to TRUE, and
+            setState(IDLE);
+        }
+
+        @Override
+        void keepAliveTimerExpires() {
+            session.keepAlive();
+            setKeepAliveTimer();
+        }
+
+        @Override
+        void tcpConnectionFails() {
+            tcpConnectionFailsOrNotifMsg();
+        }
+
+        @Override
+        void notifMsg(BGPNotification msg) {
+            tcpConnectionFailsOrNotifMsg();
+        }
+
+        protected void tcpConnectionFailsOrNotifMsg() {
+            setConnectRetryTimer(0);
+            // - deletes all routes associated with this connection,
+            // - releases all BGP resources,
+            session.dropConnection();
+            // - increments the ConnectRetryCounter by 1,
+            setState(IDLE);
+        }
+
+        @Override
+        public String toString() { return "Established"; }
+    }
+
+    BGPFsm() {
+        IDLE = new Idle();
+        CONNECT = new Connect();
+        ACTIVE = new Active();
+        OPEN_SENT = new OpenSent();
+        OPEN_CONFIRM = new OpenConfirm();
+        ESTABLISHED = new Established();
+        currentState = IDLE;
+        keepAliveTimer = new Timer();
+        holdTimer = new Timer();
+        connectRetryTimer = new Timer();
+    }
+
+    private void setKeepAliveTimer() {
+        //        System.out.println("setKeepAliveTimer: " + holdTime / 3);
+
+        if (keepAliveTimerTask != null) {
+            keepAliveTimerTask.cancel();
+            keepAliveTimerTask = null;
+        }
+
+        if (holdTime <= 0)
+            return;
+
+        keepAliveTimerTask = new TimerTask() {
+                public void run() {
+                    currentState.keepAliveTimerExpires();
+                }
+            };
+        keepAliveTimer.schedule(keepAliveTimerTask, 1000 * holdTime / 3);
+    }
+
+    private void setHoldTimer(int delay) {
+        //        System.out.println("setHoldTimer: " + delay);
+        if (holdTimerTask != null) {
+            holdTimerTask.cancel();
+            holdTimerTask = null;
+        }
+
+        if (delay <= 0)
+            return;
+
+        holdTimerTask = new TimerTask() {
+                public void run() {
+                    currentState.holdTimerExpires();
+                }
+            };
+        holdTimer.schedule(holdTimerTask, 1000 * delay);
+    }
+
+    private void setConnectRetryTimer(int delay) {
+        System.out.println("setHoldTimer(" + this + "):" + delay);
+        if (connectRetryTimerTask != null) {
+            connectRetryTimerTask.cancel();
+            connectRetryTimerTask = null;
+        }
+
+        if (delay <= 0)
+            return;
+
+        connectRetryTimerTask = new TimerTask() {
+                public void run() {
+                    currentState.connectRetryTimerExpires();
+                }
+            };
+        connectRetryTimer.schedule(connectRetryTimerTask, 1000 * delay);
+    }
+}
diff --git a/src/main/java/com/lumaserv/bgp/BGPServer.java b/src/main/java/com/lumaserv/bgp/BGPServer.java
index b8f934a..695ef96 100644
--- a/src/main/java/com/lumaserv/bgp/BGPServer.java
+++ b/src/main/java/com/lumaserv/bgp/BGPServer.java
@@ -42,34 +42,47 @@ public class BGPServer implements Runnable {
         while (true) {
             try {
                 Socket socket = serverSocket.accept();
-                BGPPacket packet = BGPPacket.read(socket.getInputStream());
-                if(packet.getType() != BGPPacket.Type.OPEN)
-                    continue;
-                BGPOpen request = new BGPOpen(packet.getMessage());
+                System.out.println("Accept");
+
                 BGPSessionConfiguration config = sessionConfigurations.stream()
-                        .filter(c -> c.getRemoteAs() == request.getAsn())
+                        .filter(c -> socket.getInetAddress().equals(c.getRemoteAddr()))
                         .findFirst()
                         .orElse(null);
-                if(config == null)
+                System.out.println("Config: " + config);
+                if(config == null) {
+                    System.out.println("Peer not found:" + socket.getInetAddress());
+                    socket.close();
                     continue;
-                BGPSession session = new BGPSession(socket, config);
-                config.getListener().onOpen(session);
-                BGPOpen response = new BGPOpen()
-                        .setAsn(config.getLocalAs())
-                        .setHoldTime(request.getHoldTime())
-                        .setVersion(request.getVersion())
-                        .setIdentifier(config.getLocalIdentifier());
-                try {
-                    socket.getOutputStream().write(new BGPPacket().setType(BGPPacket.Type.OPEN).setMessage(response.build()).build());
-                } catch (IOException e) {
-                    e.printStackTrace();
                 }
-                session.keepAlive();
-                new Thread(session).start();
+
+                BGPFsm fsm = new BGPFsm();
+                BGPSession session = new BGPSession(socket, config, fsm);
+                fsm.setSession(session);
+                System.out.println("automaticStartPassive");
+                fsm.getCurrentState().automaticStartPassive();
+
+                // System.out.println("tcpConnectionConfirmed in " + fsm.getCurrentState());
+                // fsm.getCurrentState().tcpConnectionConfirmed();
             } catch (IOException ex) {
                 ex.printStackTrace();
             }
         }
     }
 
+    public boolean connect(BGPSessionConfiguration config, String host) throws IOException {
+        return connect(config, host, 179);
+    }
+
+    public boolean connect(BGPSessionConfiguration config, String host, int port) throws IOException {
+        try {
+            BGPFsm fsm = new BGPFsm();
+            BGPSession session = new BGPSession(config, fsm, host, port);
+            fsm.setSession(session);
+            fsm.getCurrentState().automaticStart();
+            return true;
+        } catch (IOException ex) {
+            ex.printStackTrace();
+            throw(ex);
+        }
+    }
 }
diff --git a/src/main/java/com/lumaserv/bgp/BGPSession.java b/src/main/java/com/lumaserv/bgp/BGPSession.java
index 3ffadd8..94a4f5a 100644
--- a/src/main/java/com/lumaserv/bgp/BGPSession.java
+++ b/src/main/java/com/lumaserv/bgp/BGPSession.java
@@ -1,12 +1,15 @@
 package com.lumaserv.bgp;
 
 import com.lumaserv.bgp.protocol.BGPPacket;
+import com.lumaserv.bgp.protocol.message.BGPNotification;
+import com.lumaserv.bgp.protocol.message.BGPOpen;
 import com.lumaserv.bgp.protocol.message.BGPUpdate;
 import lombok.Getter;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketException;
 
@@ -14,15 +17,32 @@ public class BGPSession implements Runnable {
 
     @Getter
     final BGPSessionConfiguration configuration;
-    final InputStream inputStream;
-    final OutputStream outputStream;
+    final String host;
+    final int port;
+    Socket socket;
+    InputStream inputStream;
+    OutputStream outputStream;
     @Getter
     boolean closed;
+    boolean outgoing;
+    BGPFsm fsm;
+    Thread thread;
 
-    public BGPSession(Socket socket, BGPSessionConfiguration configuration) throws IOException {
+    public BGPSession(Socket socket, BGPSessionConfiguration configuration, BGPFsm fsm) throws IOException {
         this.configuration = configuration;
+        this.socket = socket;
         this.inputStream = socket.getInputStream();
         this.outputStream = socket.getOutputStream();
+        this.fsm = fsm;
+        this.host = null;
+        this.port = -1;
+    }
+
+    public BGPSession(BGPSessionConfiguration configuration, BGPFsm fsm, String host, int port) throws IOException {
+        this.configuration = configuration;
+        this.fsm = fsm;
+        this.host = host;
+        this.port = port;
     }
 
     public void keepAlive() {
@@ -36,28 +56,105 @@ public class BGPSession implements Runnable {
     private void handle(BGPPacket packet) {
         switch (packet.getType()) {
             case NOTIFICATION:
-                closed = true;
-                configuration.getListener().onClose(this);
+                BGPNotification notif = new BGPNotification(packet.getMessage());
+                if (notif.getMajorErrorCode() == BGPNotification.Error.MESSAGE_HEADER_ERROR.getCode() &&
+                    notif.getMinorErrorCode() == BGPNotification.OpenMessageError.UNSUPPORTED_VERSION_NUMBER.getCode())
+                    fsm.getCurrentState().notifMsgVerErr(notif);
+                else
+                    fsm.getCurrentState().notifMsg(notif);
                 break;
             case KEEPALIVE:
-                keepAlive();
+                fsm.getCurrentState().keepAliveMsg();
+                //keepAlive();
+                break;
+            case OPEN:
+                BGPOpen open = new BGPOpen(packet.getMessage());
+                fsm.getCurrentState().openMsg(open);
                 break;
             case UPDATE: {
-                configuration.getListener().onUpdate(this, new BGPUpdate(packet.getMessage()));
+                BGPUpdate update = new BGPUpdate(packet.getMessage());
+                fsm.getCurrentState().updateMsg(update);
+                configuration.getListener().onUpdate(this, update);
                 break;
             }
         }
     }
 
+    public void startIncoming() {
+        outgoing = false;
+        thread = new Thread(this);
+        thread.start();
+    }
+
+    public void startOutgoing() {
+        outgoing = true;
+        thread = new Thread(this);
+        thread.start();
+    }
+
+    public void sendOpen(BGPOpen request) throws IOException {
+        outputStream.write(new BGPPacket().setType(BGPPacket.Type.OPEN).setMessage(request.build()).build());
+System.out.println("Sent open");
+    }
+
+    public void retryConnection() {
+        if (!socket.isConnected()) {
+            try {
+                socket.close();
+            } catch (IOException ex) {
+                ex.printStackTrace();
+                throw new RuntimeException(ex);
+            }
+        }
+    }
+
+    public void dropConnection() {
+        closed = true;
+        try {
+            if (outputStream != null)
+                outputStream.close();
+            if (inputStream != null)
+                inputStream.close();
+            socket.close();
+        } catch (IOException ex) {
+            ex.printStackTrace();
+            throw new RuntimeException(ex);
+        }
+    }
+
     public void run() {
         try {
+            if (socket == null) {
+                while (true) {
+                    try {
+                        System.out.println("Outgoing");
+                        socket = new Socket();
+                        socket.connect(new InetSocketAddress(host, port));
+                        break;
+                    } catch(IOException ex) {
+                        if (closed) {
+                            throw new IOException("Closed", ex);
+                        }
+                    }
+                }
+                System.out.println("Outgoing accepted");
+                inputStream = socket.getInputStream();
+                outputStream = socket.getOutputStream();
+                fsm.getCurrentState().tcpCRAcked();
+            } else {
+                System.out.println("Incoming");
+                fsm.getCurrentState().tcpConnectionConfirmed();
+            }
             while (!closed) {
                 handle(BGPPacket.read(inputStream));
             }
+            System.out.println("Closed!");
         } catch (IOException ex) {
+            if (!closed)
+                fsm.getCurrentState().tcpConnectionFails();
+            ex.printStackTrace();
             closed = true;
             configuration.getListener().onClose(this);
         }
     }
-
 }
diff --git a/src/main/java/com/lumaserv/bgp/BGPSessionConfiguration.java b/src/main/java/com/lumaserv/bgp/BGPSessionConfiguration.java
index 8102f81..9c24d0a 100644
--- a/src/main/java/com/lumaserv/bgp/BGPSessionConfiguration.java
+++ b/src/main/java/com/lumaserv/bgp/BGPSessionConfiguration.java
@@ -4,6 +4,8 @@ import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.Setter;
 
+import java.net.InetAddress;
+
 @AllArgsConstructor
 @Setter
 @Getter
@@ -14,6 +16,6 @@ public class BGPSessionConfiguration {
     byte[] localIdentifier;
     int remoteAs;
     byte[] remoteIdentifier;
+    InetAddress remoteAddr;
     BGPListener listener;
-
 }