Add opt-in NPM and CrowdSec traffic map with selectable agent source
This commit is contained in:
Maxim
387
agent_traffic.go
Normal file
387
agent_traffic.go
Normal file
@@ -0,0 +1,387 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
var npmTrafficPattern = regexp.MustCompile(`^\[([^]]+)\] .*? (\d{3}) - ([A-Z]+) (?:https?|HTTP) (\S+) "([^"]*)" \[Client ([^]]+)\]`)
|
||||
|
||||
func parseNPMTraffic(line string) (TrafficEvent, bool) {
|
||||
m := npmTrafficPattern.FindStringSubmatch(line)
|
||||
if m == nil {
|
||||
return TrafficEvent{}, false
|
||||
}
|
||||
ip := net.ParseIP(m[6])
|
||||
if ip == nil {
|
||||
return TrafficEvent{}, false
|
||||
}
|
||||
ts, err := time.Parse("02/Jan/2006:15:04:05 -0700", m[1])
|
||||
if err != nil {
|
||||
return TrafficEvent{}, false
|
||||
}
|
||||
status, _ := strconv.Atoi(m[2])
|
||||
return TrafficEvent{Time: ts.Unix(), Kind: "request", IP: ip.String(), Domain: trafficText(m[4], 253), Method: m[3], Path: trafficPath(m[5]), Status: status}, true
|
||||
}
|
||||
|
||||
type trafficCursor struct {
|
||||
Offset int64 `json:"offset"`
|
||||
Generation int64 `json:"generation"`
|
||||
Seen int64 `json:"seen"`
|
||||
}
|
||||
type trafficAgentState struct {
|
||||
BacklogBytes int64 `json:"backlogBytes"`
|
||||
Initialized bool `json:"initialized"`
|
||||
StartedAt int64 `json:"startedAt"`
|
||||
Cursors map[string]trafficCursor `json:"cursors"`
|
||||
Pending []TrafficEvent `json:"pending"`
|
||||
ParseErrors uint64 `json:"parseErrors"`
|
||||
CrowdSeen map[string]int64 `json:"crowdSeen"`
|
||||
}
|
||||
|
||||
func (state *trafficAgentState) readLogs(dir string) (int, error) {
|
||||
state.BacklogBytes = 0
|
||||
paths, err := filepath.Glob(filepath.Join(dir, "*_access.log*"))
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
sort.Strings(paths)
|
||||
files := 0
|
||||
budget := int64(1 << 20)
|
||||
now := time.Now().Unix()
|
||||
var firstErr error
|
||||
for _, path := range paths {
|
||||
if strings.HasSuffix(path, ".gz") {
|
||||
continue
|
||||
}
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
firstErr = err
|
||||
continue
|
||||
}
|
||||
info, err := f.Stat()
|
||||
if err != nil || !info.Mode().IsRegular() {
|
||||
f.Close()
|
||||
continue
|
||||
}
|
||||
st, ok := info.Sys().(*syscall.Stat_t)
|
||||
if !ok {
|
||||
f.Close()
|
||||
continue
|
||||
}
|
||||
active := strings.HasSuffix(path, "_access.log")
|
||||
if active {
|
||||
files++
|
||||
}
|
||||
key := fmt.Sprintf("%d:%d", st.Dev, st.Ino)
|
||||
cursor, known := state.Cursors[key]
|
||||
if !known {
|
||||
if !state.Initialized || !active {
|
||||
cursor.Offset = info.Size()
|
||||
}
|
||||
cursor.Generation = now
|
||||
}
|
||||
cursor.Seen = now
|
||||
if info.Size() < cursor.Offset {
|
||||
cursor.Offset = 0
|
||||
cursor.Generation++
|
||||
}
|
||||
if len(state.Pending) < 400 && budget > 0 {
|
||||
_, err = f.Seek(cursor.Offset, io.SeekStart)
|
||||
if err == nil {
|
||||
reader := bufio.NewReaderSize(io.LimitReader(f, budget), 64<<10)
|
||||
for len(state.Pending) < 400 && budget > 0 {
|
||||
line, readErr := reader.ReadString('\n')
|
||||
// Keep an unfinished line for the next poll. Bound malformed huge lines.
|
||||
if readErr != nil {
|
||||
if len(line) >= 64<<10 {
|
||||
cursor.Offset += int64(len(line))
|
||||
state.ParseErrors++
|
||||
}
|
||||
break
|
||||
}
|
||||
offset := cursor.Offset
|
||||
cursor.Offset += int64(len(line))
|
||||
budget -= int64(len(line))
|
||||
if len(line) > 64<<10 {
|
||||
state.ParseErrors++
|
||||
continue
|
||||
}
|
||||
e, ok := parseNPMTraffic(line)
|
||||
if !ok {
|
||||
state.ParseErrors++
|
||||
continue
|
||||
}
|
||||
if e.Time < now-86400 {
|
||||
continue
|
||||
}
|
||||
e.ID = fmt.Sprintf("npm:%s:%d:%d", key, cursor.Generation, offset)
|
||||
state.Pending = append(state.Pending, e)
|
||||
}
|
||||
} else {
|
||||
firstErr = err
|
||||
}
|
||||
}
|
||||
if info.Size() > cursor.Offset {
|
||||
state.BacklogBytes += info.Size() - cursor.Offset
|
||||
}
|
||||
state.Cursors[key] = cursor
|
||||
f.Close()
|
||||
}
|
||||
for key, cursor := range state.Cursors {
|
||||
if cursor.Seen < now-7*86400 {
|
||||
delete(state.Cursors, key)
|
||||
}
|
||||
}
|
||||
if files > 0 {
|
||||
state.Initialized = true
|
||||
}
|
||||
if firstErr != nil {
|
||||
return files, errors.New("Не удалось прочитать часть access-логов NPM")
|
||||
}
|
||||
if files == 0 {
|
||||
return files, errors.New("Access-логи NPM не найдены")
|
||||
}
|
||||
return files, nil
|
||||
}
|
||||
|
||||
type crowdTrafficAlert struct {
|
||||
ID int64 `json:"id"`
|
||||
UUID string `json:"uuid"`
|
||||
CreatedAt string `json:"created_at"`
|
||||
Scenario string `json:"scenario"`
|
||||
Simulated bool `json:"simulated"`
|
||||
Source struct {
|
||||
IP string `json:"ip"`
|
||||
} `json:"source"`
|
||||
Events []struct {
|
||||
Meta []struct {
|
||||
Key string `json:"key"`
|
||||
Value string `json:"value"`
|
||||
} `json:"meta"`
|
||||
} `json:"events"`
|
||||
Decisions []struct {
|
||||
ID int64 `json:"id"`
|
||||
Type string `json:"type"`
|
||||
Value string `json:"value"`
|
||||
Simulated bool `json:"simulated"`
|
||||
} `json:"decisions"`
|
||||
}
|
||||
|
||||
func parseCrowdTraffic(data []byte) ([]TrafficEvent, error) {
|
||||
var alerts []crowdTrafficAlert
|
||||
if err := json.Unmarshal(data, &alerts); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := []TrafficEvent{}
|
||||
for _, a := range alerts {
|
||||
if a.Simulated || net.ParseIP(a.Source.IP) == nil {
|
||||
continue
|
||||
}
|
||||
ts, err := time.Parse(time.RFC3339Nano, a.CreatedAt)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
id := a.UUID
|
||||
if id == "" {
|
||||
id = fmt.Sprintf("%d:%d", a.ID, ts.Unix())
|
||||
}
|
||||
// An alert may span multiple domains. Do not invent a one-to-one link.
|
||||
domains := map[string]bool{}
|
||||
for _, event := range a.Events {
|
||||
for _, m := range event.Meta {
|
||||
if m.Key == "target_fqdn" {
|
||||
domains[m.Value] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
domain := ""
|
||||
if len(domains) == 1 {
|
||||
for d := range domains {
|
||||
domain = d
|
||||
}
|
||||
}
|
||||
out = append(out, TrafficEvent{ID: "cs-alert:" + id, Time: ts.Unix(), Kind: "alert", IP: a.Source.IP, Domain: domain, Reason: a.Scenario})
|
||||
for _, d := range a.Decisions {
|
||||
if !d.Simulated && net.ParseIP(d.Value) != nil {
|
||||
out = append(out, TrafficEvent{ID: fmt.Sprintf("cs-decision:%s:%d", id, d.ID), Time: ts.Unix(), Kind: "decision", IP: d.Value, Domain: domain, Reason: d.Type + ": " + a.Scenario})
|
||||
}
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// cscli uses the existing local machine credentials. No new bouncer key or
|
||||
// public LAPI port is needed, and only the read-only alerts list is executed.
|
||||
func collectCrowdTraffic(ctx context.Context) ([]TrafficEvent, bool, error) {
|
||||
ctx, cancel := context.WithTimeout(ctx, 6*time.Second)
|
||||
defer cancel()
|
||||
cmd := exec.CommandContext(ctx, "cscli", "alerts", "list", "--since", "24h", "--limit", "50", "-o", "json")
|
||||
var output limitedTrafficOutput
|
||||
cmd.Stdout = &output
|
||||
if err := cmd.Run(); err != nil {
|
||||
return nil, false, errors.New("cscli alerts list недоступен; проверьте права агента и локальную LAPI")
|
||||
}
|
||||
events, err := parseCrowdTraffic(output.Bytes())
|
||||
if err != nil {
|
||||
return nil, false, errors.New("Не удалось разобрать JSON CrowdSec")
|
||||
}
|
||||
var raw []json.RawMessage
|
||||
json.Unmarshal(output.Bytes(), &raw)
|
||||
return events, len(raw) >= 50, nil
|
||||
}
|
||||
|
||||
type limitedTrafficOutput struct{ bytes.Buffer }
|
||||
|
||||
func (b *limitedTrafficOutput) Write(p []byte) (int, error) {
|
||||
if b.Len()+len(p) > 4<<20 {
|
||||
return 0, errors.New("CrowdSec response too large")
|
||||
}
|
||||
return b.Buffer.Write(p)
|
||||
}
|
||||
|
||||
func saveTrafficAgentState(path string, state trafficAgentState) error {
|
||||
data, err := json.Marshal(state)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
f, err := os.CreateTemp(filepath.Dir(path), ".traffic-")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
name := f.Name()
|
||||
defer os.Remove(name)
|
||||
if err = f.Chmod(0600); err == nil {
|
||||
_, err = f.Write(data)
|
||||
}
|
||||
if err == nil {
|
||||
err = f.Sync()
|
||||
}
|
||||
closeErr := f.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if closeErr != nil {
|
||||
return closeErr
|
||||
}
|
||||
return os.Rename(name, path)
|
||||
}
|
||||
func runAgentTraffic(ctx context.Context, config agentConfig, stateDir string) {
|
||||
dir := os.Getenv("DASHBOARD_NPM_LOG_DIR")
|
||||
if dir == "off" {
|
||||
return
|
||||
}
|
||||
if dir == "" {
|
||||
dir = "/opt/npm/data/logs"
|
||||
}
|
||||
|
||||
statePath := filepath.Join(stateDir, "traffic-state.json")
|
||||
state := trafficAgentState{StartedAt: time.Now().Unix(), Cursors: map[string]trafficCursor{}, CrowdSeen: map[string]int64{}}
|
||||
if data, err := os.ReadFile(statePath); err == nil {
|
||||
if json.Unmarshal(data, &state) != nil {
|
||||
fmt.Fprintln(os.Stderr, "traffic: invalid checkpoint; collector stopped")
|
||||
return
|
||||
}
|
||||
} else if !os.IsNotExist(err) {
|
||||
fmt.Fprintln(os.Stderr, "traffic: checkpoint unreadable")
|
||||
return
|
||||
}
|
||||
if state.Cursors == nil {
|
||||
state.Cursors = map[string]trafficCursor{}
|
||||
}
|
||||
if state.CrowdSeen == nil {
|
||||
state.CrowdSeen = map[string]int64{}
|
||||
}
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
lastCrowd := time.Time{}
|
||||
crowdAt := int64(0)
|
||||
crowdError := ""
|
||||
crowdLimited := false
|
||||
ticker := time.NewTicker(2 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
files, logErr := state.readLogs(dir)
|
||||
if len(state.Pending) < 400 && time.Since(lastCrowd) >= 15*time.Second {
|
||||
events, limited, err := collectCrowdTraffic(ctx)
|
||||
crowdLimited = limited
|
||||
lastCrowd = time.Now()
|
||||
if err != nil {
|
||||
crowdError = err.Error()
|
||||
} else {
|
||||
crowdError = ""
|
||||
crowdAt = time.Now().Unix()
|
||||
for _, e := range events {
|
||||
if _, seen := state.CrowdSeen[e.ID]; !seen && len(state.Pending) < 500 {
|
||||
state.Pending = append(state.Pending, e)
|
||||
state.CrowdSeen[e.ID] = e.Time
|
||||
}
|
||||
}
|
||||
for id, ts := range state.CrowdSeen {
|
||||
if ts < time.Now().Unix()-86400 {
|
||||
delete(state.CrowdSeen, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Persist the pending batch AND offsets before sending. A retry/restart may
|
||||
// resend events; the server's (agent,event_id) unique key makes it idempotent.
|
||||
if err := saveTrafficAgentState(statePath, state); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "traffic: checkpoint write failed; collector stopped")
|
||||
return
|
||||
}
|
||||
batch := TrafficBatch{BacklogBytes: state.BacklogBytes, CrowdSecLimited: crowdLimited, Events: state.Pending, Files: files, ParseErrors: state.ParseErrors, CrowdSecAt: crowdAt, CrowdSecError: crowdError, StartedAt: state.StartedAt}
|
||||
if logErr != nil {
|
||||
batch.LogError = logErr.Error()
|
||||
}
|
||||
data, _ := json.Marshal(batch)
|
||||
req, err := http.NewRequestWithContext(ctx, "POST", config.Server+"/api/agent/traffic", bytes.NewReader(data))
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+config.ID+"."+config.Secret)
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if res, err := client.Do(req); err == nil {
|
||||
io.Copy(io.Discard, io.LimitReader(res.Body, 4096))
|
||||
res.Body.Close()
|
||||
if res.StatusCode == 200 {
|
||||
state.Pending = nil
|
||||
if saveTrafficAgentState(statePath, state) != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func startAgentTraffic(config agentConfig, stateDir string) func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() { defer close(done); runAgentTraffic(ctx, config, stateDir) }()
|
||||
return func() { cancel(); <-done }
|
||||
}
|
||||
Reference in New Issue
Block a user