limit usable

README.md

@@ -18,30 +18,32 @@ Under the hood, it works as a pipeline of data processing stages that slice and
 
 ## Install and usage
 
+See [bgpipe releases](https://github.com/bgpfix/bgpipe/releases/) on GitHub, or compile from source:
+
 ```
 # install golang, eg. https://go.dev/dl/
 $ go version
-go version go1.21.1 linux/amd64
+go version go1.21.3 linux/amd64
 
 # install bgpipe
 $ go install github.com/bgpfix/bgpipe@latest
 
 # bgpipe has built-in docs
 $ bgpipe -h
-Usage: bgpipe [OPTIONS] [--] STAGE [STAGE-OPTIONS] [STAGE-ARGUMENTS...] [--] ...
+Usage: bgpipe [OPTIONS] [--] STAGE [STAGE-OPTIONS] [ARGUMENTS...] [--] STAGE...
 
 Options:
   -l, --log string       log level (debug/info/warn/error/disabled) (default "info")
-  -D, --debug            alias for --log debug
   -e, --events strings   log given events ("all" means all events) (default [PARSE,ESTABLISHED])
+  -k, --kill strings     kill session on given events
   -i, --stdin            read stdin after session is established (unless explicitly configured)
   -s, --silent           do not write stdout (unless explicitly configured)
-  -r, --reverse          reverse the pipe direction
   -2, --short-asn        use 2-byte ASN numbers
 
 Supported stages (run stage -h to get its help)
   connect                connect to a TCP endpoint
   exec                   pass through a background JSON processor
+  limit                  limit prefix lengths and counts
   listen                 wait for a TCP client to connect
   mrt                    read MRT file with BGP4MP messages (uncompress if needed)
   speaker                run a simple local BGP speaker
@@ -54,14 +56,16 @@ Stage usage: connect [OPTIONS] ADDR
 
 connect to a TCP endpoint
 
-Options:
+Stage Options:
+      --timeout duration   connect timeout (0 means none)
+      --md5 string         TCP MD5 password
+
+Global Options:
   -L, --left               operate in L direction
   -R, --right              operate in R direction
   -W, --wait strings       wait for given event before starting
   -S, --stop strings       stop after given event is handled
   -I, --in string          where to inject new messages (default "next")
-      --timeout duration   connect timeout (0 means none)
-      --md5 string         TCP MD5 password
 ```
 
 ## Examples
@@ -88,15 +92,22 @@ $ bgpipe \
 # 2nd stage: MRT file reader, starting when the BGP session is established
 # 3rd stage: listen on TCP *:179 for new connection
 $ bgpipe \
-    -- speaker --active --asn 65055 \
-    -- mrt --wait ESTABLISHED updates.20230301.0000.bz2 \
-    -- listen :179
+  -- speaker --active --asn 65055 \
+  -- mrt --wait ESTABLISHED updates.20230301.0000.bz2 \
+  -- listen :179
 
 # a BGP sed-in-the-middle proxy rewriting ASNs in OPEN messages
 $ bgpipe \
-    -- connect 1.2.3.4 \
-    -- exec -LR -c sed -ure '/"OPEN"/{ s/65055/65001/g; s/57355/65055/g }' \
-    -- connect 85.232.240.179
+  -- connect 1.2.3.4 \
+  -- exec -LR -c sed -ure '/"OPEN"/{ s/65055/65001/g; s/57355/65055/g }' \
+  -- connect 85.232.240.179
+
+# filter prefixes lengths and add max-prefix session limits
+$ bgpipe --kill limit/session \
+  -- connect 1.2.3.4 \
+  -- limit -LR --ipv4 --min-length  8 --max-length 24 --session 1000000 \
+  -- limit -LR --ipv6 --min-length 16 --max-length 48 --session 250000 \
+  -- connect 5.6.7.8
 ```
 
 ## Author

core/attach.go

@@ -83,6 +83,15 @@ func (b *Bgpipe) Attach() error {
 		})
 	}
 
+	// kill events?
+	if evs := b.parseEvents(k, "kill", "STOP"); len(evs) > 0 {
+		p.Options.AddHandler(b.KillEvent, &pipe.Handler{
+			Pre:   true,
+			Order: math.MinInt + 1,
+			Types: evs,
+		})
+	}
+
 	return nil
 }
 

core/bgpipe.go

@@ -154,21 +154,26 @@ func (b *Bgpipe) LogEvent(ev *pipe.Event) bool {
 		l = l.Uint64("seq", ev.Seq)
 	}
 
-	if vals, ok := ev.Value.([]any); ok {
-		for i, val := range vals {
-			switch v := val.(type) {
-			case *StageBase:
-				l = l.Stringer("stage", v)
-			default:
-				l = l.Interface(fmt.Sprintf("%T[%d]", v, i), v)
-			}
+	if vals, ok := ev.Value.([]any); ok && len(vals) > 0 {
+		is := len(vals) - 1
+		if s, _ := vals[is].(*StageBase); s != nil {
+			l = l.Stringer("stage", s)
+			vals = vals[:is]
 		}
+		l = l.Interface("vals", vals)
 	}
 
 	l.Msgf("event %s", ev.Type)
 	return true
 }
 
+// KillEvent kills session because of given event ev
+func (b *Bgpipe) KillEvent(ev *pipe.Event) bool {
+	b.Cancel(fmt.Errorf("%w: %s", ErrKill, ev))
+	// TODO: why not pipe.Stop()?
+	return false
+}
+
 // AddRepo adds mapping between stage commands and their NewStageFunc
 func (b *Bgpipe) AddRepo(cmds map[string]NewStage) {
 	for cmd, newfunc := range cmds {

core/config.go

@@ -37,6 +37,7 @@ func (b *Bgpipe) addFlags() {
 	f.SetInterspersed(false)
 	f.StringP("log", "l", "info", "log level (debug/info/warn/error/disabled)")
 	f.StringSliceP("events", "e", []string{"PARSE", "ESTABLISHED"}, "log given events (\"all\" means all events)")
+	f.StringSliceP("kill", "k", []string{}, "kill session on given events")
 	f.BoolP("stdin", "i", false, "read stdin after session is established (unless explicitly configured)")
 	f.BoolP("silent", "s", false, "do not write stdout (unless explicitly configured)")
 	f.BoolP("short-asn", "2", false, "use 2-byte ASN numbers")

core/errors.go

@@ -9,4 +9,5 @@ var (
 	ErrFirstOrLast  = errors.New("must be either the first or the last stage")
 	ErrInject       = errors.New("invalid --in option value")
 	ErrLR           = errors.New("select either --left or --right, not both")
+	ErrKill         = errors.New("session killed by an event")
 )

core/util.go

@@ -80,7 +80,7 @@ func (b *Bgpipe) parseEvents(k *koanf.Koanf, key string, sds ...string) []string
 		case has_dot:
 			et = fmt.Sprintf("bgpfix/%s.%s", dot, et_upper)
 		case has_slash:
-			et = fmt.Sprintf("%s/%s", slash, et_upper) // stage event
+			et = fmt.Sprintf("%s/%s", slash, et) // stage event
 		default:
 			// stage name + stage defaults?
 			if et == et_lower && len(sds) > 0 {

go.mod

@@ -5,10 +5,11 @@ go 1.21.0
 require github.com/spf13/pflag v1.0.5
 
 require (
-	github.com/RoaringBitmap/roaring v1.6.0 // indirect
-	github.com/bgpfix/bgpfix v0.1.4-0.20231030120927-ee68b7c36a96 // indirect
-	github.com/bits-and-blooms/bitset v1.2.0 // indirect
-	github.com/mschoch/smat v0.2.0 // indirect
+	github.com/RoaringBitmap/roaring v1.6.0
+	github.com/bgpfix/bgpfix v0.1.4
+	github.com/puzpuzpuz/xsync/v3 v3.0.0
+	github.com/rs/zerolog v1.31.0
+	golang.org/x/sys v0.13.0
 )
 
 require (
@@ -21,9 +22,8 @@ require (
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
-	github.com/puzpuzpuz/xsync/v3 v3.0.0 // indirect
-	github.com/rs/zerolog v1.31.0
-	golang.org/x/sys v0.13.0
+	github.com/bits-and-blooms/bitset v1.2.0 // indirect
+	github.com/mschoch/smat v0.2.0 // indirect
 )
 
 // replace github.com/bgpfix/bgpfix => ../bgpfix

go.sum

@@ -1,9 +1,7 @@
 github.com/RoaringBitmap/roaring v1.6.0 h1:dc7kRiroETgJcHhWX6BerXkZz2b3JgLGg9nTURJL/og=
 github.com/RoaringBitmap/roaring v1.6.0/go.mod h1:plvDsJQpxOC5bw8LRteu/MLWHsHez/3y6cubLI4/1yE=
-github.com/bgpfix/bgpfix v0.1.3 h1:SJKS2MqWBwxJJGPxTjx364cCj9h2QS+dXH7HF1bwSt4=
-github.com/bgpfix/bgpfix v0.1.3/go.mod h1:Cz+WLqKyf4nTNnmBZwtvrd3gib+VzU6moK5KzviCPrk=
-github.com/bgpfix/bgpfix v0.1.4-0.20231030120927-ee68b7c36a96 h1:gtmiyM0AHv+lXkvwUWkOIyDivNQ1+HFp4uHnk99J7Jw=
-github.com/bgpfix/bgpfix v0.1.4-0.20231030120927-ee68b7c36a96/go.mod h1:Cz+WLqKyf4nTNnmBZwtvrd3gib+VzU6moK5KzviCPrk=
+github.com/bgpfix/bgpfix v0.1.4 h1:mRaGgX4CGWyuV7VeF3zE4OCRbNjYiYSZvdiKILr80bg=
+github.com/bgpfix/bgpfix v0.1.4/go.mod h1:Cz+WLqKyf4nTNnmBZwtvrd3gib+VzU6moK5KzviCPrk=
 github.com/bits-and-blooms/bitset v1.2.0 h1:Kn4yilvwNtMACtf1eYDlG8H77R07mZSPbMjLyS07ChA=
 github.com/bits-and-blooms/bitset v1.2.0/go.mod h1:gIdJ4wp64HaoK2YrL1Q5/N7Y16edYb8uY+O0FJTyyDA=
 github.com/buger/jsonparser v1.1.1 h1:2PnMjfWD7wBILjqQbt530v576A/cAbQvEW9gGIpYMUs=