Merge pull request #56 from cloudflare/feature/json-serials

Serial control

Author: lspgn

README.md

@@ -316,6 +316,18 @@ To use a data source that do not contains signatures or validity information, pa
 
 Cloudflare's prefix list removes duplicates and entries that are not routed on the Internet (>/24 IPv4 and >/48 IPv6).
 
+By default, the session ID will be randomly generated. The serial will start at zero.
+
+You can define a serial to start with the following way:
+* the JSON must contain a `serial` field in `metadata`; and
+* the flag `-useserial` must be set to 1 or 2
+
+When flag is set to 1, every change of file will increment the serial regardless of the current `serial` field.
+Make sure the refresh rate of GoRTR is more frequent than the refresh rate of the JSON.
+
+When flag is set to 2, GoRTR will set the value of the serial in the JSON. If an ID is missed or not updated,
+it will cause discrepancies on the client.
+
 ## Configurations
 
 ### Compatibility matrix

cmd/gortr/gortr.go

@@ -38,6 +38,10 @@ const (
 	METHOD_NONE = iota
 	METHOD_PASSWORD
 	METHOD_KEY
+
+	USE_SERIAL_DISABLE = iota
+	USE_SERIAL_START
+	USE_SERIAL_FULL
 )
 
 var (
@@ -78,7 +82,9 @@ var (
 	Verify    = flag.Bool("verify", true, "Check signature using provided public key (disable by passing -verify=false)")
 	PublicKey = flag.String("verify.key", "cf.pub", "Public key path (PEM file)")
 
-	CacheBin        = flag.String("cache", "https://rpki.cloudflare.com/rpki.json", "URL of the cached JSON data")
+	CacheBin  = flag.String("cache", "https://rpki.cloudflare.com/rpki.json", "URL of the cached JSON data")
+	UseSerial = flag.String("useserial", "disable", "Use serial contained in file (disable, startup, full)")
+
 	Etag            = flag.Bool("etag", true, "Enable Etag header")
 	UserAgent       = flag.String("useragent", fmt.Sprintf("Cloudflare-%v (+https://github.com/cloudflare/gortr)", AppVersion), "User-Agent header")
 	RefreshInterval = flag.Int("refresh", 600, "Refresh interval in seconds")
@@ -143,6 +149,11 @@ var (
 		"password": METHOD_PASSWORD,
 		//"key":   METHOD_KEY,
 	}
+	serialToId = map[string]int{
+		"disable": USE_SERIAL_DISABLE,
+		"startup": USE_SERIAL_START,
+		"full":    USE_SERIAL_FULL,
+	}
 )
 
 func initMetrics() {
@@ -331,6 +342,8 @@ func (e IdenticalEtag) Error() string {
 }
 
 func (s *state) updateFile(file string) error {
+	sessid, _ := s.server.GetSessionId(nil)
+
 	log.Debugf("Refreshing cache from %s", file)
 
 	s.lastts = time.Now().UTC()
@@ -354,6 +367,14 @@ func (s *state) updateFile(file string) error {
 		return err
 	}
 
+	if s.useSerial == USE_SERIAL_START || s.useSerial == USE_SERIAL_FULL {
+		//if serial, _ := s.server.GetCurrentSerial(sessid); roalistjson.Metadata.Serial != 0 && serial != roalistjson.Metadata.Serial  {
+		if _, valid := s.server.GetCurrentSerial(sessid); !valid || s.useSerial == USE_SERIAL_FULL {
+			// Set serial at beginning
+			s.server.SetSerial(uint32(roalistjson.Metadata.Serial))
+		}
+	}
+
 	if s.checktime {
 		validtime := time.Unix(int64(roalistjson.Metadata.Valid), 0).UTC()
 		if time.Now().UTC().After(validtime) {
@@ -383,12 +404,32 @@ func (s *state) updateFile(file string) error {
 		log.Infof("Slurm filtering: %v kept, %v removed, %v asserted", len(kept), len(removed), len(asserted))
 		roasjson = append(kept, asserted...)
 	}
+
+	roas, count, countv4, countv6 := processData(roasjson)
+	if err != nil {
+		return err
+	}
+
+	log.Infof("New update (%v uniques, %v total prefixes). %v bytes. Updating sha256 hash %x -> %x",
+		len(roas), count, len(s.lastconverted), s.lasthash, hsum)
+	s.lasthash = hsum
+
+	s.server.AddROAs(roas)
+
+	serial, _ := s.server.GetCurrentSerial(sessid)
+	log.Infof("Updated added, new serial %v", serial)
+	if s.sendNotifs {
+		log.Debugf("Sending notifications to clients")
+		s.server.NotifyClientsLatest()
+	}
+
 	s.lockJson.Lock()
 	s.exported = prefixfile.ROAList{
 		Metadata: prefixfile.MetaData{
 			Counts:    len(roasjson),
 			Generated: roalistjson.Metadata.Generated,
 			Valid:     roalistjson.Metadata.Valid,
+			Serial:    int(serial),
 			/*Signature:     roalistjson.Metadata.Signature,
 			SignatureDate: roalistjson.Metadata.SignatureDate,*/
 		},
@@ -406,25 +447,6 @@ func (s *state) updateFile(file string) error {
 
 	s.lockJson.Unlock()
 
-	roas, count, countv4, countv6 := processData(roasjson)
-	if err != nil {
-		return err
-	}
-
-	log.Infof("New update (%v uniques, %v total prefixes). %v bytes. Updating sha256 hash %x -> %x",
-		len(roas), count, len(s.lastconverted), s.lasthash, hsum)
-	s.lasthash = hsum
-
-	s.server.AddROAs(roas)
-
-	sessid, _ := s.server.GetSessionId(nil)
-	serial, _ := s.server.GetCurrentSerial(sessid)
-	log.Infof("Updated added, new serial %v", serial)
-	if s.sendNotifs {
-		log.Debugf("Sending notifications to clients")
-		s.server.NotifyClientsLatest()
-	}
-
 	if s.metricsEvent != nil {
 		var countv4_dup int
 		var countv6_dup int
@@ -516,6 +538,7 @@ type state struct {
 	userAgent     string
 	etags         map[string]string
 	enableEtags   bool
+	useSerial     int
 
 	server *rtr.Server
 
@@ -655,6 +678,14 @@ func main() {
 		lockJson:     &sync.RWMutex{},
 	}
 
+	if serialId, ok := serialToId[*UseSerial]; ok {
+		s.useSerial = serialId
+	} else {
+		log.Fatalf("Serial configuration %s is unknown", *UseSerial)
+	}
+
+	server.SetManualSerial(s.useSerial == USE_SERIAL_FULL)
+
 	if *ExportSign != "" {
 		keyFile, err := os.Open(*ExportSign)
 		if err != nil {
@@ -683,6 +714,38 @@ func main() {
 		log.Fatalf("Specify at least a bind address")
 	}
 
+	err := s.updateFile(*CacheBin)
+	if err != nil {
+		switch err.(type) {
+		case HttpNotModified:
+			log.Info(err)
+		case IdenticalFile:
+			log.Info(err)
+		case IdenticalEtag:
+			log.Info(err)
+		default:
+			log.Errorf("Error updating: %v", err)
+		}
+	}
+
+	slurmFile := *Slurm
+	if slurmFile != "" {
+		err := s.updateSlurm(slurmFile)
+		if err != nil {
+			switch err.(type) {
+			case HttpNotModified:
+				log.Info(err)
+			case IdenticalEtag:
+				log.Info(err)
+			default:
+				log.Errorf("Slurm: %v", err)
+			}
+		}
+		if !*SlurmRefresh {
+			slurmFile = ""
+		}
+	}
+
 	if *Bind != "" {
 		go func() {
 			sessid, _ := server.GetSessionId(nil)
@@ -795,37 +858,6 @@ func main() {
 		}()
 	}
 
-	slurmFile := *Slurm
-	if slurmFile != "" {
-		err := s.updateSlurm(slurmFile)
-		if err != nil {
-			switch err.(type) {
-			case HttpNotModified:
-				log.Info(err)
-			case IdenticalEtag:
-				log.Info(err)
-			default:
-				log.Errorf("Slurm: %v", err)
-			}
-		}
-		if !*SlurmRefresh {
-			slurmFile = ""
-		}
-	}
-
-	err := s.updateFile(*CacheBin)
-	if err != nil {
-		switch err.(type) {
-		case HttpNotModified:
-			log.Info(err)
-		case IdenticalFile:
-			log.Info(err)
-		case IdenticalEtag:
-			log.Info(err)
-		default:
-			log.Errorf("Error updating: %v", err)
-		}
-	}
 	s.routineUpdate(*CacheBin, *RefreshInterval, slurmFile)
 
 }

cmd/rtrdump/rtrdump.go

@@ -33,6 +33,10 @@ var (
 	Connect = flag.String("connect", "127.0.0.1:8282", "Connection address")
 	OutFile = flag.String("file", "output.json", "Output file")
 
+	InitSerial = flag.Bool("serial", false, "Send serial query instead of reset")
+	Serial     = flag.Int("serial.value", 0, "Serial number")
+	Session    = flag.Int("session.id", 0, "Session ID")
+
 	ConnType     = flag.String("type", "plain", "Type of connection: plain, tls or ssh")
 	ValidateCert = flag.Bool("tls.validate", true, "Validate TLS")
 
@@ -62,6 +66,10 @@ var (
 
 type Client struct {
 	Data prefixfile.ROAList
+
+	InitSerial bool
+	Serial     uint32
+	SessionID  uint16
 }
 
 func (c *Client) HandlePDU(cs *rtr.ClientSession, pdu rtr.PDU) {
@@ -86,7 +94,8 @@ func (c *Client) HandlePDU(cs *rtr.ClientSession, pdu rtr.PDU) {
 	case *rtr.PDUEndOfData:
 		t := time.Now().UTC().UnixNano() / 1000000000
 		c.Data.Metadata.Generated = int(t)
-		c.Data.Metadata.Valid = int(pdu.SerialNumber)
+		c.Data.Metadata.Valid = int(t) + int(pdu.RefreshInterval)
+		c.Data.Metadata.Serial = int(pdu.SerialNumber)
 		cs.Disconnect()
 	case *rtr.PDUCacheResponse:
 	default:
@@ -95,7 +104,11 @@ func (c *Client) HandlePDU(cs *rtr.ClientSession, pdu rtr.PDU) {
 }
 
 func (c *Client) ClientConnected(cs *rtr.ClientSession) {
-	cs.SendResetQuery()
+	if c.InitSerial {
+		cs.SendSerialQuery(c.SessionID, c.Serial)
+	} else {
+		cs.SendResetQuery()
+	}
 }
 
 func (c *Client) ClientDisconnected(cs *rtr.ClientSession) {
@@ -124,6 +137,9 @@ func main() {
 			Metadata: prefixfile.MetaData{},
 			Data:     make([]prefixfile.ROAJson, 0),
 		},
+		InitSerial: *InitSerial,
+		Serial:     uint32(*Serial),
+		SessionID:  uint16(*Session),
 	}
 
 	clientSession := rtr.NewClientSession(cc, client)

lib/client.go

@@ -62,9 +62,10 @@ func (c *ClientSession) SendResetQuery() {
 	c.SendPDU(pdu)
 }
 
-func (c *ClientSession) SendSerialQuery(serial uint32) {
+func (c *ClientSession) SendSerialQuery(sessionid uint16, serial uint32) {
 	pdu := &PDUSerialQuery{
-		// to fill
+		SessionId:    sessionid,
+		SerialNumber: serial,
 	}
 	c.SendPDU(pdu)
 }