Add metrics

cmd/crowdsec-cli/metrics.go

@@ -63,6 +63,8 @@ func FormatPrometheusMetrics(out io.Writer, url string, formatType string) error
 	lapi_machine_stats := map[string]map[string]map[string]int{}
 	lapi_bouncer_stats := map[string]map[string]map[string]int{}
 	decisions_stats := map[string]map[string]map[string]int{}
+	waap_engine_stats := map[string]map[string]int{}
+	waap_rule_stats := map[string]map[string]map[string]int{}
 	alerts_stats := map[string]int{}
 	stash_stats := map[string]struct {
 		Type  string
@@ -226,10 +228,30 @@ func FormatPrometheusMetrics(out io.Writer, url string, formatType string) error
 					Type  string
 					Count int
 				}{Type: mtype, Count: ival}
+			case "cs_waf_reqs_total":
+				if _, ok := waap_engine_stats[metric.Labels["waap_engine"]]; !ok {
+					waap_engine_stats[metric.Labels["waap_engine"]] = make(map[string]int, 0)
+				}
+				waap_engine_stats[metric.Labels["waap_engine"]]["processed"] = ival
+			case "cs_waf_block_total":
+				if _, ok := waap_engine_stats[metric.Labels["waap_engine"]]; !ok {
+					waap_engine_stats[metric.Labels["waap_engine"]] = make(map[string]int, 0)
+				}
+				waap_engine_stats[metric.Labels["waap_engine"]]["blocked"] = ival
+			case "cs_waf_rule_hits":
+				waapEngine := metric.Labels["waap_engine"]
+				ruleID := metric.Labels["rule_id"]
+				if _, ok := waap_rule_stats[waapEngine]; !ok {
+					waap_rule_stats[waapEngine] = make(map[string]map[string]int, 0)
+				}
+				if _, ok := waap_rule_stats[waapEngine][ruleID]; !ok {
+					waap_rule_stats[waapEngine][ruleID] = make(map[string]int, 0)
+				}
+				waap_rule_stats[waapEngine][ruleID]["processed"] = ival
 			default:
+				log.Infof("unknown: %+v", fam.Name)
 				continue
 			}
-
 		}
 	}
 
@@ -244,6 +266,8 @@ func FormatPrometheusMetrics(out io.Writer, url string, formatType string) error
 		decisionStatsTable(out, decisions_stats)
 		alertStatsTable(out, alerts_stats)
 		stashStatsTable(out, stash_stats)
+		waapMetricsToTable(out, waap_engine_stats)
+		waapRulesToTable(out, waap_rule_stats)
 		return nil
 	}
 
@@ -282,7 +306,6 @@ func FormatPrometheusMetrics(out io.Writer, url string, formatType string) error
 
 var noUnit bool
 
-
 func runMetrics(cmd *cobra.Command, args []string) error {
 	flags := cmd.Flags()
 
@@ -314,7 +337,6 @@ func runMetrics(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
-
 func NewMetricsCmd() *cobra.Command {
 	cmdMetrics := &cobra.Command{
 		Use:               "metrics",
@@ -322,7 +344,7 @@ func NewMetricsCmd() *cobra.Command {
 		Long:              `Fetch metrics from the prometheus server and display them in a human-friendly way`,
 		Args:              cobra.ExactArgs(0),
 		DisableAutoGenTag: true,
-		RunE: runMetrics,
+		RunE:              runMetrics,
 	}
 
 	flags := cmdMetrics.PersistentFlags()

cmd/crowdsec-cli/metrics_table.go

@@ -90,7 +90,7 @@ func bucketStatsTable(out io.Writer, stats map[string]map[string]int) {
 	keys := []string{"curr_count", "overflow", "instantiation", "pour", "underflow"}
 
 	if numRows, err := metricsToTable(t, stats, keys); err != nil {
-		log.Warningf("while collecting acquis stats: %s", err)
+		log.Warningf("while collecting bucket stats: %s", err)
 	} else if numRows > 0 {
 		renderTableTitle(out, "\nBucket Metrics:")
 		t.Render()
@@ -113,6 +113,37 @@ func acquisStatsTable(out io.Writer, stats map[string]map[string]int) {
 	}
 }
 
+func waapMetricsToTable(out io.Writer, metrics map[string]map[string]int) {
+	t := newTable(out)
+	t.SetRowLines(false)
+	t.SetHeaders("WAF Engine", "Processed", "Blocked")
+	t.SetAlignment(table.AlignLeft, table.AlignLeft)
+	keys := []string{"processed", "blocked"}
+	if numRows, err := metricsToTable(t, metrics, keys); err != nil {
+		log.Warningf("while collecting waap stats: %s", err)
+	} else if numRows > 0 {
+		renderTableTitle(out, "\nWaap Metrics:")
+		t.Render()
+	}
+}
+
+func waapRulesToTable(out io.Writer, metrics map[string]map[string]map[string]int) {
+	for waapEngine, waapEngineRulesStats := range metrics {
+		t := newTable(out)
+		t.SetRowLines(false)
+		t.SetHeaders("Rule ID", "Processed")
+		t.SetAlignment(table.AlignLeft, table.AlignLeft)
+		keys := []string{"processed"}
+		if numRows, err := metricsToTable(t, waapEngineRulesStats, keys); err != nil {
+			log.Warningf("while collecting waap stats: %s", err)
+		} else if numRows > 0 {
+			renderTableTitle(out, fmt.Sprintf("\nWaap '%s' Rules Metrics:", waapEngine))
+			t.Render()
+		}
+	}
+
+}
+
 func parserStatsTable(out io.Writer, stats map[string]map[string]int) {
 	t := newTable(out)
 	t.SetRowLines(false)
@@ -122,7 +153,7 @@ func parserStatsTable(out io.Writer, stats map[string]map[string]int) {
 	keys := []string{"hits", "parsed", "unparsed"}
 
 	if numRows, err := metricsToTable(t, stats, keys); err != nil {
-		log.Warningf("while collecting acquis stats: %s", err)
+		log.Warningf("while collecting parsers stats: %s", err)
 	} else if numRows > 0 {
 		renderTableTitle(out, "\nParser Metrics:")
 		t.Render()

cmd/crowdsec/metrics.go

@@ -164,6 +164,7 @@ func registerPrometheus(config *csconfig.PrometheusCfg) {
 			leaky.BucketsCurrentCount,
 			cache.CacheMetrics, exprhelpers.RegexpCacheMetrics,
 			waap.WafGlobalParsingHistogram, waap.WafReqCounter, waap.WafRuleHits,
+			waap.WafBlockCounter,
 		)
 	} else {
 		log.Infof("Loading prometheus collectors")
@@ -174,7 +175,7 @@ func registerPrometheus(config *csconfig.PrometheusCfg) {
 			leaky.BucketsPour, leaky.BucketsUnderflow, leaky.BucketsCanceled, leaky.BucketsInstantiation, leaky.BucketsOverflow, leaky.BucketsCurrentCount,
 			globalActiveDecisions, globalAlerts,
 			cache.CacheMetrics, exprhelpers.RegexpCacheMetrics,
-			waap.WafGlobalParsingHistogram, waap.WafInbandParsingHistogram, waap.WafOutbandParsingHistogram, waap.WafReqCounter, waap.WafRuleHits,
+			waap.WafGlobalParsingHistogram, waap.WafInbandParsingHistogram, waap.WafOutbandParsingHistogram, waap.WafReqCounter, waap.WafRuleHits, waap.WafBlockCounter,
 		)
 
 	}

pkg/acquisition/modules/waap/metrics.go

@@ -34,7 +34,15 @@ var WafReqCounter = prometheus.NewCounterVec(
 		Name: "cs_waf_reqs_total",
 		Help: "Total events processed by the WAF.",
 	},
-	[]string{"source"},
+	[]string{"source", "waap_engine"},
+)
+
+var WafBlockCounter = prometheus.NewCounterVec(
+	prometheus.CounterOpts{
+		Name: "cs_waf_block_total",
+		Help: "Total events blocked by the WAF.",
+	},
+	[]string{"source", "waap_engine"},
 )
 
 var WafRuleHits = prometheus.NewCounterVec(
@@ -42,5 +50,5 @@ var WafRuleHits = prometheus.NewCounterVec(
 		Name: "cs_waf_rule_hits",
 		Help: "Count of triggered rule, by rule_id and type (inband/outofband).",
 	},
-	[]string{"rule_id", "type"},
+	[]string{"rule_id", "type", "waap_engine", "source"},
 )

pkg/acquisition/modules/waap/utils.go

@@ -201,7 +201,8 @@ func (r *WaapRunner) AccumulateTxToEvent(evt *types.Event, req waf.ParsedRequest
 			evt.Waap.HasOutBandMatches = true
 		}
 
-		WafRuleHits.With(prometheus.Labels{"rule_id": fmt.Sprintf("%d", rule.Rule().ID()), "type": kind}).Inc()
+		// TODO: Fetch the Name of the rule when possible
+		WafRuleHits.With(prometheus.Labels{"rule_id": fmt.Sprintf("%d", rule.Rule().ID()), "type": kind, "source": req.RemoteAddrNormalized, "waap_engine": req.WaapEngine}).Inc()
 
 		name := "NOT_SET"
 		version := "NOT_SET"

pkg/acquisition/modules/waap/waap.go

@@ -133,6 +133,10 @@ func (wc *WaapSource) UnmarshalConfig(yamlConfig []byte) error {
 		return fmt.Errorf("waap_config or waap_config_path must be set")
 	}
 
+	if wc.config.Name == "" {
+		wc.config.Name = fmt.Sprintf("%s:%d%s", wc.config.ListenAddr, wc.config.ListenPort, wc.config.Path)
+	}
+
 	csConfig := csconfig.GetConfig()
 	wc.lapiURL = fmt.Sprintf("%sv1/decisions/stream", csConfig.API.Client.Credentials.URL)
 	wc.AuthCache = NewAuthCache()
@@ -349,10 +353,16 @@ func (w *WaapSource) waapHandler(rw http.ResponseWriter, r *http.Request) {
 		rw.WriteHeader(http.StatusInternalServerError)
 		return
 	}
+	parsedRequest.WaapEngine = w.config.Name
+
+	WafReqCounter.With(prometheus.Labels{"source": parsedRequest.RemoteAddrNormalized, "waap_engine": parsedRequest.WaapEngine}).Inc()
 
 	w.InChan <- parsedRequest
 
 	response := <-parsedRequest.ResponseChannel
+	if response.InBandInterrupt {
+		WafBlockCounter.With(prometheus.Labels{"source": parsedRequest.RemoteAddrNormalized, "waap_engine": parsedRequest.WaapEngine}).Inc()
+	}
 
 	waapResponse := w.WaapRuntime.GenerateResponse(response.InBandInterrupt)
 

pkg/acquisition/modules/waap/waap_runner.go

@@ -184,7 +184,6 @@ func (r *WaapRunner) Run(t *tomb.Tomb) error {
 			request.IsInBand = true
 			request.IsOutBand = false
 
-			WafReqCounter.With(prometheus.Labels{"source": request.RemoteAddr}).Inc()
 			//to measure the time spent in the WAF
 			startParsing := time.Now()
 

pkg/waf/request.go

@@ -3,10 +3,13 @@ package waf
 import (
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/url"
 
+
 	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
 )
 
 const (
@@ -60,23 +63,25 @@ const (
 // }
 
 type ParsedRequest struct {
-	RemoteAddr       string
-	Host             string
-	ClientIP         string
-	URI              string
-	Args             url.Values
-	ClientHost       string
-	Headers          http.Header
-	URL              *url.URL
-	Method           string
-	Proto            string
-	Body             []byte
-	TransferEncoding []string
-	UUID             string
-	Tx               ExtendedTransaction
-	ResponseChannel  chan WaapTempResponse
-	IsInBand         bool
-	IsOutBand        bool
+	RemoteAddr           string
+	Host                 string
+	ClientIP             string
+	URI                  string
+	Args                 url.Values
+	ClientHost           string
+	Headers              http.Header
+	URL                  *url.URL
+	Method               string
+	Proto                string
+	Body                 []byte
+	TransferEncoding     []string
+	UUID                 string
+	Tx                   ExtendedTransaction
+	ResponseChannel      chan WaapTempResponse
+	IsInBand             bool
+	IsOutBand            bool
+	WaapEngine           string
+	RemoteAddrNormalized string
 }
 
 // Generate a ParsedRequest from a http.Request. ParsedRequest can be consumed by the Waap Engine
@@ -123,20 +128,35 @@ func NewParsedRequestFromRequest(r *http.Request) (ParsedRequest, error) {
 		return ParsedRequest{}, fmt.Errorf("unable to parse url '%s': %s", clientURI, err)
 	}
 
+	RemoteAddrNormalized := ""
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		log.Errorf("Invalid waap remote IP source %v: %s", r.RemoteAddr, err.Error())
+		RemoteAddrNormalized = r.RemoteAddr
+	} else {
+		ip := net.ParseIP(host)
+		if ip == nil {
+			log.Errorf("Invalid waap remote IP address source %v: %s", r.RemoteAddr, err.Error())
+			RemoteAddrNormalized = r.RemoteAddr
+		}
+		RemoteAddrNormalized = ip.String()
+	}
+
 	return ParsedRequest{
-		RemoteAddr:       r.RemoteAddr,
-		UUID:             uuid.New().String(),
-		ClientHost:       clientHost,
-		ClientIP:         clientIP,
-		URI:              parsedURL.Path,
-		Method:           clientMethod,
-		Host:             r.Host,
-		Headers:          r.Header,
-		URL:              r.URL,
-		Proto:            r.Proto,
-		Body:             body,
-		Args:             parsedURL.Query(), //TODO: Check if there's not potential bypass as it excludes malformed args
-		TransferEncoding: r.TransferEncoding,
-		ResponseChannel:  make(chan WaapTempResponse),
+		RemoteAddr:           r.RemoteAddr,
+		UUID:                 uuid.New().String(),
+		ClientHost:           clientHost,
+		ClientIP:             clientIP,
+		URI:                  parsedURL.Path,
+		Method:               clientMethod,
+		Host:                 r.Host,
+		Headers:              r.Header,
+		URL:                  r.URL,
+		Proto:                r.Proto,
+		Body:                 body,
+		Args:                 parsedURL.Query(), //TODO: Check if there's not potential bypass as it excludes malformed args
+		TransferEncoding:     r.TransferEncoding,
+		ResponseChannel:      make(chan WaapTempResponse),
+		RemoteAddrNormalized: RemoteAddrNormalized,
 	}, nil
 }