rpki: support reset/softreset/enable operation

gobgp rpki server <ip address> reset|softreset|enable

reset: close a tcp conneciton and retry to connect and send ResetQuery
softreset: send ResetQuery
enable: send SerialQuery

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

2 files changed, 65 insertions, 31 deletions

diff --git a/server/rpki.go b/server/rpki.go
index 326203ec..1864fa25 100644
--- a/server/rpki.go
+++ b/server/rpki.go
@@ -80,7 +80,6 @@ func newROAManager(as uint32, conf config.RpkiServers) (*roaManager, error) {
 		client := &roaClient{
 			host:    net.JoinHostPort(c.Address.String(), strconv.Itoa(int(c.Port))),
 			eventCh: m.eventCh,
-			state:   &entry.RpkiServerState,
 		}
 		m.clientMap[client.host] = client
 		client.t.Go(client.tryConnect)
@@ -89,6 +88,25 @@ func newROAManager(as uint32, conf config.RpkiServers) (*roaManager, error) {
 	return m, nil
 }
 
+func (m *roaManager) operate(op api.Operation, address string) error {
+	for network, client := range m.clientMap {
+		add, _ := splitHostPort(network)
+		if add == address {
+			switch op {
+			case api.Operation_ENABLE:
+				client.enable()
+			case api.Operation_DISABLE:
+			case api.Operation_RESET:
+				client.reset()
+			case api.Operation_SOFTRESET:
+				client.softReset()
+			}
+			return nil
+		}
+	}
+	return fmt.Errorf("roa server not found %s", address)
+}
+
 func (c *roaManager) recieveROA() chan *roaClientEvent {
 	return c.eventCh
 }
@@ -108,7 +126,6 @@ func (m *roaManager) handleROAEvent(ev *roaClientEvent) {
 		client.state.Downtime = time.Now().Unix()
 		// clear state
 		client.state.RpkiMessages = config.RpkiMessages{}
-		client.conn.Close()
 		client.conn = nil
 		client.t = tomb.Tomb{}
 		client.t.Go(client.tryConnect)
@@ -119,7 +136,7 @@ func (m *roaManager) handleROAEvent(ev *roaClientEvent) {
 		client.t = tomb.Tomb{}
 		client.t.Go(client.established)
 	case RTR:
-		m.handleRTRMsg(ev.src, client.state, ev.data)
+		m.handleRTRMsg(client, &client.state, ev.data)
 	}
 }
 
@@ -166,13 +183,16 @@ func addROA(host string, tree *radix.Tree, as uint32, prefix []byte, prefixLen,
 	}
 }
 
-func (c *roaManager) handleRTRMsg(host string, state *config.RpkiServerState, buf []byte) {
+func (c *roaManager) handleRTRMsg(client *roaClient, state *config.RpkiServerState, buf []byte) {
 	received := &state.RpkiMessages.RpkiReceived
 
 	m, _ := bgp.ParseRTR(buf)
 	if m != nil {
+
 		switch msg := m.(type) {
 		case *bgp.RTRSerialNotify:
+			client.sessionID = msg.RTRCommon.SessionID
+			client.serialNumber = msg.RTRCommon.SerialNumber
 			received.SerialNotify++
 		case *bgp.RTRSerialQuery:
 		case *bgp.RTRResetQuery:
@@ -187,9 +207,11 @@ func (c *roaManager) handleRTRMsg(host string, state *config.RpkiServerState, bu
 				received.Ipv6Prefix++
 				tree = c.roas[bgp.RF_IPv6_UC]
 			}
-			addROA(host, tree, msg.AS, msg.Prefix, msg.PrefixLen, msg.MaxLen)
+			addROA(client.host, tree, msg.AS, msg.Prefix, msg.PrefixLen, msg.MaxLen)
 		case *bgp.RTREndOfData:
 			received.EndOfData++
+			client.sessionID = msg.RTRCommon.SessionID
+			client.serialNumber = msg.RTRCommon.SerialNumber
 		case *bgp.RTRCacheReset:
 			received.CacheReset++
 		case *bgp.RTRErrorReport:
@@ -342,14 +364,42 @@ func (c *roaManager) validate(pathList []*table.Path) {
 }
 
 type roaClient struct {
-	t       tomb.Tomb
-	host    string
-	conn    *net.TCPConn
-	state   *config.RpkiServerState
-	eventCh chan *roaClientEvent
+	t            tomb.Tomb
+	host         string
+	conn         *net.TCPConn
+	state        config.RpkiServerState
+	eventCh      chan *roaClientEvent
+	sessionID    uint16
+	serialNumber uint32
+}
+
+func (c *roaClient) enable() error {
+	if c.conn != nil {
+		r := bgp.NewRTRSerialQuery(c.sessionID, c.serialNumber)
+		data, _ := r.Serialize()
+		_, err := c.conn.Write(data)
+		if err != nil {
+			return err
+		}
+		c.state.RpkiMessages.RpkiSent.SerialQuery++
+	}
+	return nil
 }
 
-func (c *roaClient) kill() {
+func (c *roaClient) softReset() error {
+	if c.conn != nil {
+		r := bgp.NewRTRResetQuery()
+		data, _ := r.Serialize()
+		_, err := c.conn.Write(data)
+		if err != nil {
+			return err
+		}
+		c.state.RpkiMessages.RpkiSent.ResetQuery++
+	}
+	return nil
+}
+
+func (c *roaClient) reset() {
 	c.t.Kill(nil)
 	if c.conn != nil {
 		c.conn.Close()
@@ -383,16 +433,12 @@ func (c *roaClient) established() error {
 		}
 	}
 
-	r := bgp.NewRTRResetQuery()
-	data, _ := r.Serialize()
-	_, err := c.conn.Write(data)
+	err := c.softReset()
 	if err != nil {
 		disconnected()
 		return nil
 	}
 
-	c.state.RpkiMessages.RpkiSent.ResetQuery++
-
 	reader := bufio.NewReader(c.conn)
 	scanner := bufio.NewScanner(reader)
 	scanner.Split(bgp.SplitRTR)
diff --git a/server/server.go b/server/server.go
index c5875299..a72c9d1d 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2375,22 +2375,7 @@ func (server *BgpServer) handleModMrt(grpcReq *GrpcRequest) {
 
 func (server *BgpServer) handleModRpki(grpcReq *GrpcRequest) {
 	arg := grpcReq.Data.(*api.ModRpkiArguments)
-	configured := false
-	if len(server.bgpConfig.RpkiServers.RpkiServerList) > 0 {
-		configured = true
-	}
 
-	if arg.Operation == api.Operation_ADD {
-		if configured {
-			grpcDone(grpcReq, fmt.Errorf("already enabled"))
-			return
-		}
-	} else {
-		if !configured {
-			grpcDone(grpcReq, fmt.Errorf("not enabled yet"))
-			return
-		}
-	}
 	switch arg.Operation {
 	case api.Operation_ADD:
 		r := config.RpkiServer{}
@@ -2400,6 +2385,9 @@ func (server *BgpServer) handleModRpki(grpcReq *GrpcRequest) {
 		server.roaManager, _ = newROAManager(server.bgpConfig.Global.GlobalConfig.As, server.bgpConfig.RpkiServers)
 		grpcDone(grpcReq, nil)
 		return
+	case api.Operation_ENABLE, api.Operation_DISABLE, api.Operation_RESET, api.Operation_SOFTRESET:
+		grpcDone(grpcReq, server.roaManager.operate(arg.Operation, arg.Address))
+		return
 	}
 	grpcDone(grpcReq, fmt.Errorf("not supported yet"))
 }