crowdsec/pkg/parser/runtime.go

362 lines
10 KiB
Go
Raw Normal View History

2020-05-15 09:39:16 +00:00
package parser
/*
This file contains
- the runtime parsing routines
*/
import (
"errors"
"fmt"
"reflect"
"strings"
"sync"
"time"
2020-05-15 09:39:16 +00:00
"github.com/crowdsecurity/crowdsec/pkg/types"
"strconv"
"github.com/mohae/deepcopy"
2020-05-15 09:39:16 +00:00
"github.com/prometheus/client_golang/prometheus"
log "github.com/sirupsen/logrus"
"github.com/antonmedv/expr"
)
/* ok, this is kinda experimental, I don't know how bad of an idea it is .. */
func SetTargetByName(target string, value string, evt *types.Event) bool {
if evt == nil {
return false
}
2020-05-20 09:00:25 +00:00
2020-05-15 09:39:16 +00:00
//it's a hack, we do it for the user
2020-05-20 09:00:25 +00:00
target = strings.TrimPrefix(target, "evt.")
2020-05-15 09:39:16 +00:00
log.Debugf("setting target %s to %s", target, value)
defer func() {
if r := recover(); r != nil {
log.Errorf("Runtime error while trying to set '%s': %+v", target, r)
2020-05-15 09:39:16 +00:00
return
}
}()
iter := reflect.ValueOf(evt).Elem()
if (iter == reflect.Value{}) || iter.IsZero() {
log.Tracef("event is nill")
//event is nill
return false
}
for _, f := range strings.Split(target, ".") {
/*
** According to current Event layout we only have to handle struct and map
*/
switch iter.Kind() {
case reflect.Map:
tmp := iter.MapIndex(reflect.ValueOf(f))
/*if we're in a map and the field doesn't exist, the user wants to add it :) */
if (tmp == reflect.Value{}) || tmp.IsZero() {
log.Debugf("map entry is zero in '%s'", target)
}
iter.SetMapIndex(reflect.ValueOf(f), reflect.ValueOf(value))
return true
case reflect.Struct:
tmp := iter.FieldByName(f)
if !tmp.IsValid() {
log.Debugf("'%s' is not a valid target because '%s' is not valid", target, f)
2020-05-15 09:39:16 +00:00
return false
}
if tmp.Kind() == reflect.Ptr {
tmp = reflect.Indirect(tmp)
}
2020-05-15 09:39:16 +00:00
iter = tmp
//nolint: gosimple
2020-05-15 09:39:16 +00:00
break
case reflect.Ptr:
tmp := iter.Elem()
iter = reflect.Indirect(tmp.FieldByName(f))
2020-05-15 09:39:16 +00:00
default:
log.Errorf("unexpected type %s in '%s'", iter.Kind(), target)
return false
}
}
//now we should have the final member :)
if !iter.CanSet() {
log.Errorf("'%s' can't be set", target)
return false
}
if iter.Kind() != reflect.String {
log.Errorf("Expected string, got %v when handling '%s'", iter.Kind(), target)
return false
}
iter.Set(reflect.ValueOf(value))
return true
}
func printStaticTarget(static types.ExtraField) string {
if static.Method != "" {
return static.Method
} else if static.Parsed != "" {
return fmt.Sprintf(".Parsed[%s]", static.Parsed)
} else if static.Meta != "" {
return fmt.Sprintf(".Meta[%s]", static.Meta)
} else if static.Enriched != "" {
return fmt.Sprintf(".Enriched[%s]", static.Enriched)
} else if static.TargetByName != "" {
return static.TargetByName
} else {
return "?"
}
}
func (n *Node) ProcessStatics(statics []types.ExtraField, event *types.Event) error {
2020-05-15 09:39:16 +00:00
//we have a few cases :
//(meta||key) + (static||reference||expr)
var value string
2020-12-14 13:12:22 +00:00
clog := n.Logger
2020-05-15 09:39:16 +00:00
for _, static := range statics {
value = ""
if static.Value != "" {
value = static.Value
} else if static.RunTimeValue != nil {
output, err := expr.Run(static.RunTimeValue, map[string]interface{}{"evt": event})
2020-05-15 09:39:16 +00:00
if err != nil {
clog.Warningf("failed to run RunTimeValue : %v", err)
continue
}
2020-05-20 09:26:21 +00:00
switch out := output.(type) {
2020-05-15 09:39:16 +00:00
case string:
2020-05-20 09:26:21 +00:00
value = out
2020-05-15 09:39:16 +00:00
case int:
2020-05-20 09:26:21 +00:00
value = strconv.Itoa(out)
2022-06-08 10:15:29 +00:00
case map[string]interface{}:
clog.Warnf("Expression returned a map, please use ToJsonString() to convert it to string if you want to keep it as is, or refine your expression to extract a string")
case []interface{}:
clog.Warnf("Expression returned a map, please use ToJsonString() to convert it to string if you want to keep it as is, or refine your expression to extract a string")
case nil:
clog.Debugf("Expression returned nil, skipping")
2020-05-15 09:39:16 +00:00
default:
2022-06-08 10:15:29 +00:00
clog.Errorf("unexpected return type for RunTimeValue : %T", output)
2020-05-15 09:39:16 +00:00
return errors.New("unexpected return type for RunTimeValue")
}
}
if value == "" {
//allow ParseDate to have empty input
if static.Method != "ParseDate" {
clog.Debugf("Empty value for %s, skip.", printStaticTarget(static))
continue
}
2020-05-15 09:39:16 +00:00
}
if static.Method != "" {
processed := false
/*still way too hackish, but : inject all the results in enriched, and */
if enricherPlugin, ok := n.EnrichFunctions.Registered[static.Method]; ok {
clog.Tracef("Found method '%s'", static.Method)
ret, err := enricherPlugin.EnrichFunc(value, event, enricherPlugin.Ctx, n.Logger)
if err != nil {
clog.Errorf("method '%s' returned an error : %v", static.Method, err)
2020-05-15 09:39:16 +00:00
}
processed = true
clog.Debugf("+ Method %s('%s') returned %d entries to merge in .Enriched\n", static.Method, value, len(ret))
//Hackish check, but those methods do not return any data by design
if len(ret) == 0 && static.Method != "UnmarshalJSON" {
clog.Debugf("+ Method '%s' empty response on '%s'", static.Method, value)
}
for k, v := range ret {
clog.Debugf("\t.Enriched[%s] = '%s'\n", k, v)
event.Enriched[k] = v
}
} else {
clog.Debugf("method '%s' doesn't exist or plugin not initialized", static.Method)
2020-05-15 09:39:16 +00:00
}
if !processed {
clog.Debugf("method '%s' doesn't exist", static.Method)
2020-05-15 09:39:16 +00:00
}
} else if static.Parsed != "" {
clog.Debugf(".Parsed[%s] = '%s'", static.Parsed, value)
event.Parsed[static.Parsed] = value
2020-05-15 09:39:16 +00:00
} else if static.Meta != "" {
clog.Debugf(".Meta[%s] = '%s'", static.Meta, value)
event.Meta[static.Meta] = value
2020-05-15 09:39:16 +00:00
} else if static.Enriched != "" {
clog.Debugf(".Enriched[%s] = '%s'", static.Enriched, value)
event.Enriched[static.Enriched] = value
2020-05-15 09:39:16 +00:00
} else if static.TargetByName != "" {
if !SetTargetByName(static.TargetByName, value, event) {
2020-05-15 09:39:16 +00:00
clog.Errorf("Unable to set value of '%s'", static.TargetByName)
} else {
clog.Debugf("%s = '%s'", static.TargetByName, value)
}
} else {
clog.Fatal("unable to process static : unknown target")
2020-05-15 09:39:16 +00:00
}
}
return nil
}
var NodesHits = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_node_hits_total",
Help: "Total events entered node.",
2020-05-15 09:39:16 +00:00
},
[]string{"source", "type", "name"},
2020-05-15 09:39:16 +00:00
)
var NodesHitsOk = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_node_hits_ok_total",
2021-12-06 16:29:23 +00:00
Help: "Total events successfully exited node.",
2020-05-15 09:39:16 +00:00
},
[]string{"source", "type", "name"},
2020-05-15 09:39:16 +00:00
)
var NodesHitsKo = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "cs_node_hits_ko_total",
2021-12-06 16:29:23 +00:00
Help: "Total events unsuccessfully exited node.",
2020-05-15 09:39:16 +00:00
},
[]string{"source", "type", "name"},
2020-05-15 09:39:16 +00:00
)
func stageidx(stage string, stages []string) int {
for i, v := range stages {
if stage == v {
return i
}
}
return -1
}
type ParserResult struct {
Evt types.Event
Success bool
}
2020-06-10 10:14:27 +00:00
var ParseDump bool
var DumpFolder string
var StageParseCache map[string]map[string][]ParserResult
var StageParseMutex sync.Mutex
2020-06-10 10:14:27 +00:00
func Parse(ctx UnixParserCtx, xp types.Event, nodes []Node) (types.Event, error) {
var event = xp
2020-05-20 15:50:56 +00:00
2020-05-15 09:39:16 +00:00
/* the stage is undefined, probably line is freshly acquired, set to first stage !*/
if event.Stage == "" && len(ctx.Stages) > 0 {
event.Stage = ctx.Stages[0]
log.Tracef("no stage, set to : %s", event.Stage)
2020-05-15 09:39:16 +00:00
}
event.Process = false
if event.Time.IsZero() {
event.Time = time.Now().UTC()
}
2020-05-15 09:39:16 +00:00
if event.Parsed == nil {
event.Parsed = make(map[string]string)
}
if event.Enriched == nil {
event.Enriched = make(map[string]string)
}
if event.Meta == nil {
event.Meta = make(map[string]string)
}
if event.Type == types.LOG {
log.Tracef("INPUT '%s'", event.Line.Raw)
}
2020-06-10 10:14:27 +00:00
if ParseDump {
if StageParseCache == nil {
StageParseMutex.Lock()
StageParseCache = make(map[string]map[string][]ParserResult)
StageParseCache["success"] = make(map[string][]ParserResult)
StageParseCache["success"][""] = make([]ParserResult, 0)
StageParseMutex.Unlock()
}
2020-06-10 10:14:27 +00:00
}
2020-05-15 09:39:16 +00:00
for _, stage := range ctx.Stages {
2020-06-10 10:14:27 +00:00
if ParseDump {
StageParseMutex.Lock()
if _, ok := StageParseCache[stage]; !ok {
StageParseCache[stage] = make(map[string][]ParserResult)
}
StageParseMutex.Unlock()
2020-06-10 10:14:27 +00:00
}
2022-03-16 08:40:00 +00:00
/* if the node is forward in stages, seek to this stage */
2020-05-15 09:39:16 +00:00
/* this is for example used by testing system to inject logs in post-syslog-parsing phase*/
if stageidx(event.Stage, ctx.Stages) > stageidx(stage, ctx.Stages) {
log.Tracef("skipping stage, we are already at [%s] expecting [%s]", event.Stage, stage)
continue
}
log.Tracef("node stage : %s, current stage : %s", event.Stage, stage)
/* if the stage is wrong, it means that the log didn't manage "pass" a stage with a onsuccess: next_stage tag */
if event.Stage != stage {
log.Debugf("Event not parsed, expected stage '%s' got '%s', abort", stage, event.Stage)
2022-03-16 08:40:00 +00:00
event.Process = false
return event, nil
2020-05-15 09:39:16 +00:00
}
isStageOK := false
for idx, node := range nodes {
//Only process current stage's nodes
if event.Stage != node.Stage {
continue
}
2020-06-10 10:14:27 +00:00
clog := log.WithFields(log.Fields{
"node-name": node.rn,
"stage": event.Stage,
})
2020-05-15 09:39:16 +00:00
clog.Tracef("Processing node %d/%d -> %s", idx, len(nodes), node.rn)
if ctx.Profiling {
2020-05-15 09:39:16 +00:00
node.Profiling = true
}
ret, err := node.process(&event, ctx, map[string]interface{}{"evt": &event})
2020-05-15 09:39:16 +00:00
if err != nil {
2022-06-08 10:15:29 +00:00
clog.Errorf("Error while processing node : %v", err)
return event, err
2020-05-15 09:39:16 +00:00
}
clog.Tracef("node (%s) ret : %v", node.rn, ret)
if ParseDump {
if len(StageParseCache[stage][node.Name]) == 0 {
StageParseMutex.Lock()
StageParseCache[stage][node.Name] = make([]ParserResult, 0)
StageParseMutex.Unlock()
}
evtcopy := deepcopy.Copy(event)
parserInfo := ParserResult{Evt: evtcopy.(types.Event), Success: ret}
StageParseMutex.Lock()
StageParseCache[stage][node.Name] = append(StageParseCache[stage][node.Name], parserInfo)
StageParseMutex.Unlock()
}
if ret {
2020-05-15 09:39:16 +00:00
isStageOK = true
}
if ret && node.OnSuccess == "next_stage" {
2020-05-15 09:39:16 +00:00
clog.Debugf("node successful, stop end stage %s", stage)
break
}
//the parsed object moved onto the next phase
if event.Stage != stage {
clog.Tracef("node moved stage, break and redo")
break
}
}
if !isStageOK {
2020-05-15 09:39:16 +00:00
log.Debugf("Log didn't finish stage %s", event.Stage)
event.Process = false
return event, nil
}
}
event.Process = true
return event, nil
}