388 lines
10 KiB
Go
388 lines
10 KiB
Go
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 }
|
||
}
|