rpki: support dynamic add/delete operations via gRPC

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

3 files changed, 88 insertions, 41 deletions

diff --git a/server/rpki.go b/server/rpki.go
index 329ee053..f4c2ca32 100644
--- a/server/rpki.go
+++ b/server/rpki.go
@@ -21,7 +21,6 @@ import (
 	"io"
 	"net"
 	"sort"
-	"strconv"
 	"time"
 
 	log "github.com/Sirupsen/logrus"
@@ -134,34 +133,58 @@ type roaClientEvent struct {
 type roaManager struct {
 	AS        uint32
 	Roas      map[bgp.RouteFamily]*radix.Tree
-	config    []config.RpkiServer
 	eventCh   chan *roaClientEvent
 	clientMap map[string]*roaClient
 }
 
-func NewROAManager(as uint32, servers []config.RpkiServer) (*roaManager, error) {
+func NewROAManager(as uint32) (*roaManager, error) {
 	m := &roaManager{
-		AS:     as,
-		Roas:   make(map[bgp.RouteFamily]*radix.Tree),
-		config: servers,
+		AS:   as,
+		Roas: make(map[bgp.RouteFamily]*radix.Tree),
 	}
 	m.Roas[bgp.RF_IPv4_UC] = radix.New()
 	m.Roas[bgp.RF_IPv6_UC] = radix.New()
 	m.eventCh = make(chan *roaClientEvent)
 	m.clientMap = make(map[string]*roaClient)
+	return m, nil
+}
 
-	for _, entry := range servers {
-		c := entry.Config
-		// should be set somewhere else
-		if c.RecordLifetime == 0 {
-			c.RecordLifetime = 3600
-		}
-		client := NewRoaClient(c.Address, strconv.Itoa(int(c.Port)), m.eventCh, c.RecordLifetime)
-		m.clientMap[client.host] = client
-		client.t.Go(client.tryConnect)
+func (m *roaManager) SetAS(as uint32) error {
+	if m.AS != 0 {
+		return fmt.Errorf("AS was already configured")
 	}
+	m.AS = as
+	return nil
+}
 
-	return m, nil
+func (m *roaManager) AddServer(host string, lifetime int64) error {
+	if m.AS == 0 {
+		return fmt.Errorf("AS isn't configured yet")
+	}
+	address, port, err := net.SplitHostPort(host)
+	if err != nil {
+		return err
+	}
+	if lifetime == 0 {
+		lifetime = 3600
+	}
+	if _, ok := m.clientMap[host]; ok {
+		return fmt.Errorf("roa server exists %s", host)
+	}
+	client := NewRoaClient(address, port, m.eventCh, lifetime)
+	m.clientMap[host] = client
+	client.t.Go(client.tryConnect)
+	return nil
+}
+
+func (m *roaManager) DeleteServer(host string) error {
+	client, ok := m.clientMap[host]
+	if !ok {
+		return fmt.Errorf("roa server doesn't exists %s", host)
+	}
+	client.reset()
+	delete(m.clientMap, host)
+	return nil
 }
 
 func (m *roaManager) deleteAllROA(network string) {
diff --git a/server/rpki_test.go b/server/rpki_test.go
index fb270cf7..c58ef32b 100644
--- a/server/rpki_test.go
+++ b/server/rpki_test.go
@@ -61,7 +61,7 @@ func validateOne(tree *radix.Tree, cidr, aspathStr string) config.RpkiValidation
 func TestValidate0(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("192.168.0.0").To4(), 24, 32, 100, ""))
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("192.168.0.0").To4(), 24, 24, 200, ""))
 
@@ -90,7 +90,7 @@ func TestValidate0(t *testing.T) {
 func TestValidate1(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 16, 65000, ""))
 
 	var r config.RpkiValidationResultType
@@ -106,7 +106,7 @@ func TestValidate1(t *testing.T) {
 func TestValidate2(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 
 	var r config.RpkiValidationResultType
 
@@ -121,7 +121,7 @@ func TestValidate2(t *testing.T) {
 func TestValidate3(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 16, 65000, ""))
 
 	var r config.RpkiValidationResultType
@@ -133,7 +133,7 @@ func TestValidate3(t *testing.T) {
 	r = validateOne(tree, "10.0.0.0/17", "65000")
 	assert.Equal(r, config.RPKI_VALIDATION_RESULT_TYPE_INVALID)
 
-	manager, _ = NewROAManager(0, []config.RpkiServer{})
+	manager, _ = NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 24, 65000, ""))
 
 	tree = manager.Roas[bgp.RF_IPv4_UC]
@@ -144,7 +144,7 @@ func TestValidate3(t *testing.T) {
 func TestValidate4(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 16, 65000, ""))
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 16, 65001, ""))
 
@@ -160,7 +160,7 @@ func TestValidate4(t *testing.T) {
 func TestValidate5(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 17, 17, 65000, ""))
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.128.0").To4(), 17, 17, 65000, ""))
 
@@ -173,7 +173,7 @@ func TestValidate5(t *testing.T) {
 func TestValidate6(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 8, 32, 0, ""))
 
 	var r config.RpkiValidationResultType
@@ -191,7 +191,7 @@ func TestValidate6(t *testing.T) {
 func TestValidate7(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 24, 65000, ""))
 
 	var r config.RpkiValidationResultType
@@ -209,7 +209,7 @@ func TestValidate7(t *testing.T) {
 func TestValidate8(t *testing.T) {
 	assert := assert.New(t)
 
-	manager, _ := NewROAManager(0, []config.RpkiServer{})
+	manager, _ := NewROAManager(0)
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 24, 0, ""))
 	manager.addROA(NewROA(bgp.AFI_IP, net.ParseIP("10.0.0.0").To4(), 16, 24, 65000, ""))
 
diff --git a/server/server.go b/server/server.go
index 1102ab44..64ec3f53 100644
--- a/server/server.go
+++ b/server/server.go
@@ -148,7 +148,6 @@ type BgpServer struct {
 	updatedPeerCh chan config.Neighbor
 	fsmincomingCh *channels.InfiniteChannel
 	fsmStateCh    chan *FsmMsg
-	rpkiConfigCh  chan []config.RpkiServer
 	acceptCh      chan *net.TCPConn
 	zapiMsgCh     chan *zebra.Message
 
@@ -171,12 +170,11 @@ func NewBgpServer() *BgpServer {
 	b.addedPeerCh = make(chan config.Neighbor)
 	b.deletedPeerCh = make(chan config.Neighbor)
 	b.updatedPeerCh = make(chan config.Neighbor)
-	b.rpkiConfigCh = make(chan []config.RpkiServer)
 	b.GrpcReqCh = make(chan *GrpcRequest, 1)
 	b.policyUpdateCh = make(chan config.RoutingPolicy)
 	b.neighborMap = make(map[string]*Peer)
 	b.watchers = Watchers(make(map[watcherType]watcher))
-	b.roaManager, _ = NewROAManager(0, nil)
+	b.roaManager, _ = NewROAManager(0)
 	b.policy = table.NewRoutingPolicy()
 	return &b
 }
@@ -207,8 +205,6 @@ func (server *BgpServer) Listeners(addr string) []*net.TCPListener {
 }
 
 func (server *BgpServer) Serve() {
-	server.roaManager, _ = NewROAManager(0, nil)
-
 	w, _ := newGrpcIncomingWatcher()
 	server.watchers[WATCHER_GRPC_INCOMING] = w
 
@@ -336,8 +332,6 @@ func (server *BgpServer) Serve() {
 	CONT:
 
 		select {
-		case c := <-server.rpkiConfigCh:
-			server.roaManager, _ = NewROAManager(server.bgpConfig.Global.Config.As, c)
 		case rmsg := <-server.roaManager.ReceiveROA():
 			server.roaManager.HandleROAEvent(rmsg)
 		case zmsg := <-server.zapiMsgCh:
@@ -1060,8 +1054,37 @@ func (server *BgpServer) SetGlobalType(g config.Global) error {
 	return nil
 }
 
-func (server *BgpServer) SetRpkiConfig(c []config.RpkiServer) {
-	server.rpkiConfigCh <- c
+func (server *BgpServer) SetRpkiConfig(c []config.RpkiServer) error {
+	ch := make(chan *GrpcResponse)
+	server.GrpcReqCh <- &GrpcRequest{
+		RequestType: REQ_MOD_RPKI,
+		Data: &api.ModRpkiArguments{
+			Operation: api.Operation_INITIALIZE,
+			Asn:       server.bgpConfig.Global.Config.As,
+		},
+		ResponseCh: ch,
+	}
+	if err := (<-ch).Err(); err != nil {
+		return err
+	}
+
+	for _, s := range c {
+		ch := make(chan *GrpcResponse)
+		server.GrpcReqCh <- &GrpcRequest{
+			RequestType: REQ_MOD_RPKI,
+			Data: &api.ModRpkiArguments{
+				Operation: api.Operation_ADD,
+				Address:   s.Config.Address,
+				Port:      s.Config.Port,
+				Lifetime:  s.Config.RecordLifetime,
+			},
+			ResponseCh: ch,
+		}
+		if err := (<-ch).Err(); err != nil {
+			return err
+		}
+	}
+	return nil
 }
 
 func (server *BgpServer) SetBmpConfig(c []config.BmpServer) error {
@@ -2836,13 +2859,14 @@ func (server *BgpServer) handleModRpki(grpcReq *GrpcRequest) {
 	arg := grpcReq.Data.(*api.ModRpkiArguments)
 
 	switch arg.Operation {
+	case api.Operation_INITIALIZE:
+		grpcDone(grpcReq, server.roaManager.SetAS(arg.Asn))
+		return
 	case api.Operation_ADD:
-		r := config.RpkiServer{}
-		r.Config.Address = arg.Address
-		r.Config.Port = arg.Port
-		server.bgpConfig.RpkiServers = append(server.bgpConfig.RpkiServers, r)
-		server.roaManager, _ = NewROAManager(server.bgpConfig.Global.Config.As, server.bgpConfig.RpkiServers)
-		grpcDone(grpcReq, nil)
+		grpcDone(grpcReq, server.roaManager.AddServer(net.JoinHostPort(arg.Address, strconv.Itoa(int(arg.Port))), arg.Lifetime))
+		return
+	case api.Operation_DEL:
+		grpcDone(grpcReq, server.roaManager.DeleteServer(arg.Address))
 		return
 	case api.Operation_ENABLE, api.Operation_DISABLE, api.Operation_RESET, api.Operation_SOFTRESET:
 		grpcDone(grpcReq, server.roaManager.operate(arg.Operation, arg.Address))