diff --git a/README.md b/README.md index cbac19a..44d3a4d 100644 --- a/README.md +++ b/README.md @@ -507,3 +507,50 @@ GeoIP работает локально: DB-IP Country Lite за октябрь DASHBOARD_TRAFFIC_PREVIEW=1 go test -run '^TestTrafficPreview$' -timeout 30m # http://127.0.0.1:19105/traffic ``` + +## Сетевая активность TCP/UDP/ICMP (v0.21.0) + +В **Настройки → Карта трафика** включите дополнительный сбор TCP/UDP/ICMP. +Он выполняется только выбранным агентом v0.21.0+; на странице карты появится +кнопка **TCP / UDP**, открывающая `/network`. Основной выключатель карты +отключает оба сборщика. Сетевой режим по умолчанию выключен и не создаёт +общих алертов/писем. + +Для Debian/Ubuntu на хосте агента (не внутри контейнера NPM): + +```sh +sudo apt-get update +sudo apt-get install -y conntrack nftables +``` + +Агенту нужны root/CAP_NET_ADMIN в сетевом namespace хоста. Установщик агента +запускает службу от root. Чтение: `/proc/net/nf_conntrack`, если доступен, +иначе `conntrack -L -f ipv4/ipv6 -o extended`. Ничего устанавливать на хост +самого Dashboard не требуется. Firewall, NAT и sysctl сборщик не меняет. + +Страница показывает состояния TCP, UDP/ICMP-потоки, обе пары адресов до/после +NAT (в том числе Docker), локальную GeoIP, топ инициаторов и портов, фильтры, +поиск и детали потока. Список листается по страницам; переход на следующую +фиксирует снимок, «К свежим» возвращает live. Для телефона — вкладки. + +Это **снимки таблицы примерно каждые 5 секунд**, не полный журнал соединений: +короткие потоки между снимками могут быть пропущены. До 5 000 записей на +снимок, 8 MiB вывода команды и таймаут 4 секунды; ограничения/ошибки видны в +диагностике. Сохраняется только последний снимок каждого агента. Инициатор +не обязательно внешний клиент: NPM → upstream — отдельный поток. Порт не +доказывает протокол приложения; UDP ASSURED не означает TCP handshake. + +Байты/пакеты доступны только при включённом учёте conntrack для конкретных +потоков. Это накопленные значения за жизнь потока, а не трафик за 24 часа. +При отсутствии учёта интерфейс показывает «нет учёта». При необходимости +администратор может включить `net.netfilter.nf_conntrack_acct=1`; изменение +применяется к новым потокам и автоматически не выполняется. + +Firewall читается раз в 30 секунд командой `nft -j list ruleset`. Поддержаны +анонимные счётчики непосредственно перед финальным `drop/reject` в таблице +или цепочке с `crowdsec` в имени. Остальные счётчики не выдаются за блокировки. +Это сумма пакетов/байтов выбранных правил с момента их создания, независимо +от фильтров потоков; при пересоздании правил она может уменьшиться. +Для iptables/ipset-only, именованных счётчиков и правил без счётчиков выводится +«недоступно», а не ноль. Отброшенный пакет до подтверждения conntrack может +отсутствовать в таблице потоков. Содержимое пакетов не перехватывается. diff --git a/agent.go b/agent.go index de38f9f..edc57ee 100644 --- a/agent.go +++ b/agent.go @@ -145,6 +145,7 @@ type AgentUpdateManifest struct { type agentReportResponse struct { Accepted bool `json:"accepted"` + NetworkEnabled bool `json:"networkEnabled"` TrafficEnabled bool `json:"trafficEnabled"` Update *AgentUpdateManifest `json:"update,omitempty"` Commands []AgentCommand `json:"commands,omitempty"` @@ -340,6 +341,14 @@ func runGuestAgent(opts agentOptions) error { return err } } + var networkStop func() + stopNetwork := func() { + if networkStop != nil { + networkStop() + networkStop = nil + } + } + defer stopNetwork() var trafficStop func() stopTraffic := func() { if trafficStop != nil { @@ -367,6 +376,7 @@ func runGuestAgent(opts agentOptions) error { fresh, enrollmentErr := enrollRemote(config, opts.EnrollmentToken) if enrollmentErr == nil { stopTraffic() + stopNetwork() config = fresh if saveErr := saveAgentConfig(opts.ConfigPath, config); saveErr != nil { return saveErr @@ -383,6 +393,11 @@ func runGuestAgent(opts agentOptions) error { if response.StatusCode == http.StatusOK { var result agentReportResponse if json.Unmarshal(responseData, &result) == nil { + if !result.NetworkEnabled { + stopNetwork() + } else if networkStop == nil { + networkStop = startAgentNetwork(config) + } if !result.TrafficEnabled { stopTraffic() } else if trafficStop == nil { diff --git a/agent_network.go b/agent_network.go new file mode 100644 index 0000000..e11fc4d --- /dev/null +++ b/agent_network.go @@ -0,0 +1,266 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "os" + "os/exec" + "runtime" + "strconv" + "strings" + "time" +) + +// Bounded command output and deadline keep large conntrack tables from exhausting the agent. +type networkBuffer struct { + bytes.Buffer + limit int +} + +func (b *networkBuffer) Write(p []byte) (int, error) { + if len(p) > b.limit-b.Len() { + return 0, errors.New("output limit exceeded") + } + return b.Buffer.Write(p) +} +func networkCommand(ctx context.Context, name string, args ...string) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, 4*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, name, args...) + out := &networkBuffer{limit: 8 << 20} + stderr := &networkBuffer{limit: 4096} + cmd.Stdout = out + cmd.Stderr = stderr + if err := cmd.Run(); err != nil { + return nil, fmt.Errorf("%s: %v %s", name, err, trafficText(stderr.String(), 240)) + } + return out.Bytes(), nil +} +func parseNetworkFlow(line string) (NetworkFlow, bool) { + var f NetworkFlow + fields := strings.Fields(line) + protoIndex := -1 + for i, v := range fields { + if v == "tcp" || v == "udp" || v == "icmp" || v == "icmpv6" { + f.Protocol = v + protoIndex = i + break + } + } + if protoIndex < 0 { + return f, false + } + values := map[string][]string{} + for _, v := range fields[protoIndex+1:] { + if k, val, ok := strings.Cut(v, "="); ok { + values[k] = append(values[k], val) + } + } + get := func(k string, i int) string { + if len(values[k]) > i { + return values[k][i] + } + return "" + } + integer := func(k string, i int) int { n, _ := strconv.Atoi(get(k, i)); return n } + number := func(k string, i int) uint64 { n, _ := strconv.ParseUint(get(k, i), 10, 64); return n } + f.Source = get("src", 0) + f.Destination = get("dst", 0) + f.ReplySource = get("src", 1) + f.ReplyDestination = get("dst", 1) + if net.ParseIP(f.Source) == nil || net.ParseIP(f.Destination) == nil || net.ParseIP(f.ReplySource) == nil || net.ParseIP(f.ReplyDestination) == nil { + return f, false + } + f.SourcePort = integer("sport", 0) + f.DestinationPort = integer("dport", 0) + f.ReplySourcePort = integer("sport", 1) + f.ReplyDestinationPort = integer("dport", 1) + f.State = "TRACKED" + if f.Protocol == "tcp" && len(fields) > protoIndex+3 && !strings.Contains(fields[protoIndex+3], "=") { + f.State = fields[protoIndex+3] + } + if strings.Contains(line, "[UNREPLIED]") { + f.State = "UNREPLIED" + if f.Protocol == "tcp" && len(fields) > protoIndex+3 { + f.State = fields[protoIndex+3] + } + } + if f.Protocol == "udp" && strings.Contains(line, "[ASSURED]") { + f.State = "ASSURED" + } + f.Accounting = len(values["bytes"]) == 2 && len(values["packets"]) == 2 + f.Bytes = number("bytes", 0) + f.ReplyBytes = number("bytes", 1) + f.Packets = number("packets", 0) + f.ReplyPackets = number("packets", 1) + f.Zone = get("zone", 0) + return f, true +} +func appendNetworkFlows(snapshot *NetworkSnapshot, raw []byte) { + for _, line := range strings.Split(string(raw), "\n") { + if strings.TrimSpace(line) == "" { + continue + } + f, ok := parseNetworkFlow(line) + if !ok { + if strings.Contains(line, "src=") { + snapshot.ParseErrors++ + } + continue + } + if len(snapshot.Flows) >= 5000 { + snapshot.Limited = true + continue + } + snapshot.Flows = append(snapshot.Flows, f) + } +} + +// Only counters immediately before a terminal drop/reject in a CrowdSec table/chain +// are reported as blocked packets. General chain and set counters are not summed. +func parseFirewallCounters(raw []byte) ([]FirewallCounter, error) { + var doc struct { + NFTables []struct { + Rule *struct { + Family string `json:"family"` + Table string `json:"table"` + Chain string `json:"chain"` + Handle int `json:"handle"` + Expr []map[string]json.RawMessage `json:"expr"` + } `json:"rule"` + } `json:"nftables"` + } + if err := json.Unmarshal(raw, &doc); err != nil { + return nil, err + } + result := []FirewallCounter{} + for _, item := range doc.NFTables { + r := item.Rule + if r == nil || !strings.Contains(strings.ToLower(r.Table+" "+r.Chain), "crowdsec") { + continue + } + if len(r.Expr) < 2 { + continue + } + last := r.Expr[len(r.Expr)-1] + _, drop := last["drop"] + _, reject := last["reject"] + if !drop && !reject { + continue + } + counter, ok := r.Expr[len(r.Expr)-2]["counter"] + if !ok { + continue + } + var c struct { + Packets uint64 `json:"packets"` + Bytes uint64 `json:"bytes"` + } + if json.Unmarshal(counter, &c) != nil || len(counter) == 0 || counter[0] != '{' { + continue + } + result = append(result, FirewallCounter{Name: fmt.Sprintf("%s/%s/%s #%d", r.Family, r.Table, r.Chain, r.Handle), Packets: c.Packets, Bytes: c.Bytes}) + if len(result) == 128 { + break + } + } + return result, nil +} +func collectNetwork(ctx context.Context) NetworkSnapshot { + s := NetworkSnapshot{At: time.Now().Unix(), Flows: []NetworkFlow{}} + if runtime.GOOS != "linux" { + s.Error = "Сбор conntrack поддерживается только на Linux" + return s + } + // Proc contains both address families when available. Modern kernels often expose only netlink. + for _, path := range []string{"/proc/net/nf_conntrack", "/proc/net/ip_conntrack"} { + f, err := os.Open(path) + if err != nil { + continue + } + raw, err := io.ReadAll(io.LimitReader(f, (8<<20)+1)) + f.Close() + if err != nil { + s.Error = err.Error() + return s + } + if len(raw) > 8<<20 { + s.Error = "Таблица conntrack превышает лимит чтения 8 MiB" + return s + } + appendNetworkFlows(&s, raw) + return s + } + for _, family := range []string{"ipv4", "ipv6"} { + raw, err := networkCommand(ctx, "conntrack", "-L", "-f", family, "-o", "extended") + if err != nil { + s.Error += "Conntrack " + family + ": нужен пакет conntrack и права root/CAP_NET_ADMIN. " + err.Error() + "; " + continue + } + appendNetworkFlows(&s, raw) + } + return s +} +func startAgentNetwork(config agentConfig) func() { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + client := &http.Client{Timeout: 10 * time.Second} + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + var firewall []FirewallCounter + var firewallAt int64 + var firewallErr string + lastFirewall := time.Time{} + for { + if ctx.Err() != nil { + return + } + snapshot := collectNetwork(ctx) + if time.Since(lastFirewall) >= 30*time.Second { + lastFirewall = time.Now() + raw, err := networkCommand(ctx, "nft", "-j", "list", "ruleset") + firewall = nil + firewallAt = 0 + firewallErr = "" + if err == nil { + firewall, err = parseFirewallCounters(raw) + } + if err != nil { + firewallErr = "Счётчики nftables недоступны: " + err.Error() + } else if len(firewall) == 0 { + firewallErr = "В правилах CrowdSec нет поддерживаемых счётчиков перед drop/reject. Это не означает отсутствие блокировок." + } else { + firewallAt = time.Now().Unix() + } + } + snapshot.Firewall = firewall + snapshot.FirewallAt = firewallAt + snapshot.FirewallError = firewallErr + data, _ := json.Marshal(snapshot) + req, err := http.NewRequestWithContext(ctx, "POST", config.Server+"/api/agent/network", 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() + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() + return func() { cancel(); <-done } +} diff --git a/assets/network.html b/assets/network.html new file mode 100644 index 0000000..8d4e519 --- /dev/null +++ b/assets/network.html @@ -0,0 +1,39 @@ +Сетевая активность · Proxmox Dashboard +
PROXMOX / NETWORK OBSERVATORY

Сетевая активность.

+
ОЖИДАНИЕПодключение к агенту…
+
TCP · ESTABLISHED—
TCP · SYN_SENT / SYN_RECV—
UDP · отслеживаемые потоки—
ICMP / ICMPv6 · записи—
+
+ +

Текущая таблица conntrack

—
ИнициаторНазначение / NATПротоколСостояниеБайты → / ←
Ожидание первого снимка…
—
+
+ +

 + diff --git a/assets/traffic.html b/assets/traffic.html index 74cf241..31fc10b 100644 --- a/assets/traffic.html +++ b/assets/traffic.html @@ -10,7 +10,7 @@ @media(prefers-reduced-motion:reduce){*{scroll-behavior:auto!important}}
PROXMOX / LIVE OBSERVATORY

Traffic space.

-
+
Уникальные IP · 24 ч—
HTTP-запросы · 24 ч—
Срабатывания · 24 ч—
Решения CrowdSec · 24 ч—
Глобальная активность
HTTPСрабатываниеРешение
Без геолокации: —
@@ -35,7 +35,7 @@ function drawMap(now){const canvas=el('world'),w=Math.max(1,Math.floor(canvas.cl } function ranking(id,items,label){el(id).innerHTML=items.length?items.slice(0,7).map((c,i)=>'
'+String(i+1).padStart(2,'0')+''+safe(label(c.name))+''+num(c.count)+'
').join(''):'
Пока нет событий
'} function drawChart(){const c=el('chart'),w=c.clientWidth,h=Math.max(35,c.clientHeight),dpr=devicePixelRatio||1;c.width=w*dpr;c.height=h*dpr;const ctx=c.getContext('2d');ctx.scale(dpr,dpr);if(!view)return;const buckets=new Map(view.series.map(v=>[Number(v.name),v.count])),end=Math.floor(view.now/60)*60,values=Array.from({length:60},(_,i)=>buckets.get(end-(59-i)*60)||0),max=Math.max(1,...values);ctx.strokeStyle='#26314b';ctx.beginPath();ctx.moveTo(0,h-18);ctx.lineTo(w,h-18);ctx.stroke();ctx.fillStyle='#55c9ff';values.forEach((v,i)=>{const x=i*w/60,height=v/max*(h-32);ctx.fillRect(x,h-18-height,Math.max(1,w/60-3),height)});ctx.fillStyle='#8490ab';ctx.font='10px system-ui';ctx.fillText('−60 мин',0,h-2);ctx.textAlign='right';ctx.fillText('сейчас',w,h-2);ctx.fillText('макс. '+num(max)+'/мин',w,12)} -function render(data){view=data;el('unique').textContent=num(data.uniqueIps);el('requests').textContent=num(data.requests);el('alerts').textContent=num(data.alerts);el('decisions').textContent=num(data.decisions);el('qps').textContent=Number(data.qps).toFixed(2);el('geoDate').textContent=data.geoDate;ranking('countries',data.countries,countryName);ranking('ips',data.ips,v=>v); +function render(data){view=data;el('networkLink').hidden=!data.settings.networkEnabled;el('unique').textContent=num(data.uniqueIps);el('requests').textContent=num(data.requests);el('alerts').textContent=num(data.alerts);el('decisions').textContent=num(data.decisions);el('qps').textContent=Number(data.qps).toFixed(2);el('geoDate').textContent=data.geoDate;ranking('countries',data.countries,countryName);ranking('ips',data.ips,v=>v); const src=data.sources[0],fresh=src&&data.now-src.lastSeen<20;el('live').textContent=paused?'ПАУЗА':fresh?'● В ЭФИРЕ':'ОЖИДАНИЕ АГЕНТА';el('live').classList.toggle('stale',!fresh||paused);el('sourceName').textContent=src?src.name:'Источник ещё не прислал данные'; const messages=[];if(!src)messages.push('Выбранный агент ещё не прислал данные. Нужен агент v0.20.0 или новее.');else{messages.push(src.name+' · '+src.status.files+' логов · сбор с '+new Date(src.status.startedAt*1000).toLocaleString('ru-RU'));if(!fresh)messages.push('Источник не обновляется; показана сохранённая история');if(src.status.logError)messages.push(src.status.logError);if(src.status.crowdsecError)messages.push('CrowdSec: '+src.status.crowdsecError);if(src.status.backlogBytes)messages.push('Ожидает чтения: '+num(src.status.backlogBytes)+' байт');if(src.status.crowdsecLimited)messages.push('CrowdSec: получены последние 50 срабатываний; более ранняя история может быть неполной');if(src.status.parseErrors)messages.push('Нераспознанных строк: '+src.status.parseErrors)}if(!data.geoReady)messages.push('База GeoIP недоступна');if(data.limited)messages.push('Достигнут лимит 200 000 событий: часть суточной истории удалена');statusMessages=messages;el('status').textContent=fresh?'● на связи':'● нет связи';el('sourceDetails').title=messages.join(' · '); el('status').classList.toggle('error',!fresh||!!src?.status.logError||!!src?.status.crowdsecError||data.limited); diff --git a/network_flows.go b/network_flows.go new file mode 100644 index 0000000..16d3532 --- /dev/null +++ b/network_flows.go @@ -0,0 +1,170 @@ +package main + +import ( + "database/sql" + _ "embed" + "encoding/json" + "errors" + "io" + "net" + "net/http" + "time" +) + +//go:embed assets/network.html +var networkHTML string + +// Original and reply tuples describe one conntrack entry, not two connections. +type NetworkFlow struct { + External bool `json:"external"` + Protocol string `json:"protocol"` + Source string `json:"source"` + Destination string `json:"destination"` + SourcePort int `json:"sourcePort"` + DestinationPort int `json:"destinationPort"` + ReplySource string `json:"replySource"` + ReplyDestination string `json:"replyDestination"` + ReplySourcePort int `json:"replySourcePort"` + ReplyDestinationPort int `json:"replyDestinationPort"` + State string `json:"state"` + Country string `json:"country"` + DestinationCountry string `json:"destinationCountry"` + Packets uint64 `json:"packets"` + ReplyPackets uint64 `json:"replyPackets"` + Bytes uint64 `json:"bytes"` + ReplyBytes uint64 `json:"replyBytes"` + Accounting bool `json:"accounting"` + Zone string `json:"zone,omitempty"` +} +type FirewallCounter struct { + Name string `json:"name"` + Packets uint64 `json:"packets"` + Bytes uint64 `json:"bytes"` +} +type NetworkSnapshot struct { + At int64 `json:"at"` + ReceivedAt int64 `json:"receivedAt"` + Flows []NetworkFlow `json:"flows"` + Error string `json:"error,omitempty"` + Limited bool `json:"limited"` + ParseErrors int `json:"parseErrors"` + Firewall []FirewallCounter `json:"firewall"` + FirewallAt int64 `json:"firewallAt"` + FirewallError string `json:"firewallError,omitempty"` + AgentName string `json:"agentName"` +} + +func networkExternalIP(value string) bool { + ip := net.ParseIP(value) + if ip == nil || !ip.IsGlobalUnicast() || ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() { + return false + } + if v4 := ip.To4(); v4 != nil && v4[0] == 100 && v4[1] >= 64 && v4[1] <= 127 { + return false + } + return true +} +func (s *Store) SaveNetwork(agent string, snapshot NetworkSnapshot) error { + if len(snapshot.Flows) > 5000 || len(snapshot.Firewall) > 128 { + return errors.New("network snapshot too large") + } + if snapshot.At < time.Now().Unix()-120 || snapshot.At > time.Now().Unix()+60 { + return errors.New("invalid snapshot time") + } + for i := range snapshot.Flows { + f := &snapshot.Flows[i] + if net.ParseIP(f.Source) == nil || net.ParseIP(f.Destination) == nil || net.ParseIP(f.ReplySource) == nil || net.ParseIP(f.ReplyDestination) == nil { + return errors.New("invalid network address") + } + for _, p := range []int{f.SourcePort, f.DestinationPort, f.ReplySourcePort, f.ReplyDestinationPort} { + if p < 0 || p > 65535 { + return errors.New("invalid network port") + } + } + if f.Protocol != "tcp" && f.Protocol != "udp" && f.Protocol != "icmp" && f.Protocol != "icmpv6" { + return errors.New("invalid protocol") + } + f.State = trafficText(f.State, 32) + f.Zone = trafficText(f.Zone, 16) + f.External = networkExternalIP(f.Source) || networkExternalIP(f.Destination) + f.Country = lookupTrafficCountry(f.Source) + f.DestinationCountry = lookupTrafficCountry(f.Destination) + } + snapshot.Error = trafficText(snapshot.Error, 600) + snapshot.FirewallError = trafficText(snapshot.FirewallError, 600) + for i := range snapshot.Firewall { + snapshot.Firewall[i].Name = trafficText(snapshot.Firewall[i].Name, 200) + } + snapshot.ReceivedAt = time.Now().Unix() + snapshot.AgentName = "" + raw, err := json.Marshal(snapshot) + if err != nil { + return err + } + _, err = s.db.Exec(`INSERT INTO network_sources(agent,snapshot) VALUES(?,?) ON CONFLICT(agent) DO UPDATE SET snapshot=excluded.snapshot`, agent, string(raw)) + return err +} +func networkRoutes(mux *http.ServeMux, s *Store) { + enabled := func() bool { c := s.TrafficSettings(); return c.Enabled && c.NetworkEnabled } + mux.HandleFunc("/network", func(w http.ResponseWriter, r *http.Request) { + if !enabled() { + http.Error(w, "Включите TCP/UDP в настройках карты трафика", 404) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Write([]byte(networkHTML)) + }) + mux.HandleFunc("/api/network", func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + w.WriteHeader(405) + return + } + if !enabled() { + http.Error(w, "Сбор сети отключён", 404) + return + } + var raw, name string + err := s.db.QueryRow(`SELECT n.snapshot,a.name FROM network_sources n JOIN agents a ON a.id=n.agent WHERE n.agent=?`, s.TrafficSettings().AgentID).Scan(&raw, &name) + if err != nil && err != sql.ErrNoRows { + writeJSON(w, nil, err) + return + } + if err != nil { + writeJSON(w, NetworkSnapshot{Flows: []NetworkFlow{}, Error: "Ожидание агента v0.21.0 или новее"}, nil) + return + } + var snapshot NetworkSnapshot + if err = json.Unmarshal([]byte(raw), &snapshot); err != nil { + writeJSON(w, nil, err) + return + } + snapshot.AgentName = name + writeJSON(w, snapshot, nil) + }) + mux.HandleFunc("/api/agent/network", func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + w.WriteHeader(405) + return + } + id, secret, ok := agentCredentials(r) + if !ok || !s.AuthenticateAgent(id, secret) { + http.Error(w, "Unauthorized", 401) + return + } + c := s.TrafficSettings() + if !c.Enabled || !c.NetworkEnabled || c.AgentID != id { + http.Error(w, "Источник отключён", 403) + return + } + var snapshot NetworkSnapshot + if json.NewDecoder(io.LimitReader(r.Body, 4<<20)).Decode(&snapshot) != nil { + http.Error(w, "Invalid snapshot", 400) + return + } + if err := s.SaveNetwork(id, snapshot); err != nil { + http.Error(w, err.Error(), 400) + return + } + writeJSON(w, map[string]bool{"accepted": true}, nil) + }) +} diff --git a/network_flows_test.go b/network_flows_test.go new file mode 100644 index 0000000..31ff4d0 --- /dev/null +++ b/network_flows_test.go @@ -0,0 +1,137 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +const networkNATLine = "ipv4 2 tcp 6 431999 ESTABLISHED src=8.8.8.8 dst=192.168.0.12 sport=51000 dport=443 packets=12 bytes=1500 src=172.18.0.2 dst=8.8.8.8 sport=443 dport=51000 packets=16 bytes=2500 [ASSURED] mark=0 zone=2 use=1" + +func TestNetworkConntrackTuples(t *testing.T) { + f, ok := parseNetworkFlow(networkNATLine) + if !ok || f.Source != "8.8.8.8" || f.Destination != "192.168.0.12" || f.ReplySource != "172.18.0.2" || f.SourcePort != 51000 || f.ReplySourcePort != 443 || f.State != "ESTABLISHED" || !f.Accounting || f.Bytes != 1500 || f.ReplyBytes != 2500 || f.Zone != "2" { + t.Fatalf("NAT tuple corrupted: %+v", f) + } + udp, ok := parseNetworkFlow("ipv6 10 udp 17 29 src=2001:4860:4860::8888 dst=fd00::2 sport=51000 dport=53 [UNREPLIED] src=fd00::2 dst=2001:4860:4860::8888 sport=53 dport=51000 mark=0 use=1") + if !ok || udp.State != "UNREPLIED" || udp.Accounting || udp.Protocol != "udp" { + t.Fatalf("UDP: %+v", udp) + } + icmp, ok := parseNetworkFlow("icmp 1 29 src=8.8.8.8 dst=192.168.0.12 type=8 code=0 id=3 src=192.168.0.12 dst=8.8.8.8 type=0 code=0 id=3 use=1") + if !ok || icmp.SourcePort != 0 || icmp.State != "TRACKED" { + t.Fatalf("ICMP: %+v", icmp) + } + if _, ok := parseNetworkFlow("tcp 6 20 SYN_SENT src=not-an-ip dst=192.168.0.12"); ok { + t.Fatal("invalid accepted") + } +} +func TestNetworkSnapshotLimit(t *testing.T) { + var s NetworkSnapshot + appendNetworkFlows(&s, []byte(strings.Repeat(networkNATLine+"\n", 5001))) + if len(s.Flows) != 5000 || !s.Limited { + t.Fatalf("limit not enforced %d", len(s.Flows)) + } + b := networkBuffer{limit: 2} + if _, err := b.Write([]byte("123")); err == nil || b.Len() != 0 { + t.Fatal("output unbounded") + } +} +func TestFirewallOnlyConfirmedDropCounters(t *testing.T) { + raw := `{"nftables":[{"rule":{"family":"ip","table":"crowdsec","chain":"crowdsec-chain","handle":3,"expr":[{"match":{}},{"counter":{"packets":10,"bytes":1000}},{"drop":null}]}},{"rule":{"table":"crowdsec","expr":[{"counter":{"packets":99}},{"accept":null}]}},{"rule":{"table":"other","expr":[{"counter":{"packets":99}},{"drop":null}]}},{"rule":{"table":"crowdsec","expr":[{"counter":{"packets":99}},{"match":{}},{"drop":null}]}},{"rule":{"table":"crowdsec","expr":[{"counter":"named-counter"},{"drop":null}]}}]}` + counters, err := parseFirewallCounters([]byte(raw)) + if err != nil || len(counters) != 1 || counters[0].Packets != 10 || counters[0].Bytes != 1000 { + t.Fatalf("invented blocks: %+v %v", counters, err) + } + if _, err := parseFirewallCounters([]byte("invalid")); err == nil { + t.Fatal("invalid nft JSON accepted") + } +} +func TestNetworkRouteAuthorizationAndSourceSelection(t *testing.T) { + s, a := trafficFixture(t) + s.SaveTrafficSettings(TrafficSettings{Enabled: true, AgentID: a.ID}) + mux := http.NewServeMux() + trafficRoutes(mux, s) + flow, _ := parseNetworkFlow(networkNATLine) + snapshot := NetworkSnapshot{At: time.Now().Unix(), Flows: []NetworkFlow{flow}} + raw, _ := json.Marshal(snapshot) + post := func(token string) int { + r := httptest.NewRequest("POST", "/api/agent/network", bytes.NewReader(raw)) + r.Header.Set("Authorization", "Bearer "+token) + w := httptest.NewRecorder() + mux.ServeHTTP(w, r) + return w.Code + } + if got := post(a.ID + "." + a.Secret); got != 403 { + t.Fatalf("disabled accepts data: %d", got) + } + s.SaveTrafficSettings(TrafficSettings{Enabled: true, NetworkEnabled: true, AgentID: a.ID}) + if got := post("wrong.token"); got != 401 { + t.Fatalf("unauthenticated: %d", got) + } + if got := post(a.ID + "." + a.Secret); got != 200 { + t.Fatalf("valid: %d", got) + } + w := httptest.NewRecorder() + mux.ServeHTTP(w, httptest.NewRequest("GET", "/api/network", nil)) + var got NetworkSnapshot + json.Unmarshal(w.Body.Bytes(), &got) + if w.Code != 200 || len(got.Flows) != 1 || got.Flows[0].Country == "" || got.ReceivedAt == 0 { + t.Fatalf("bad snapshot: %s", w.Body.String()) + } + token, _ := s.CreateAgentEnrollmentToken() + other, _ := s.EnrollAgent(agentEnrollmentRequest{Token: token, Name: "other"}) + s.SaveTrafficSettings(TrafficSettings{Enabled: true, NetworkEnabled: true, AgentID: other.ID}) + if code := post(a.ID + "." + a.Secret); code != 403 { + t.Fatalf("old source accepted: %d", code) + } + w = httptest.NewRecorder() + mux.ServeHTTP(w, httptest.NewRequest("GET", "/api/network", nil)) + json.Unmarshal(w.Body.Bytes(), &got) + if len(got.Flows) != 0 { + t.Fatal("old source data leaked") + } + s.SaveTrafficSettings(TrafficSettings{Enabled: false, NetworkEnabled: true, AgentID: a.ID}) + for _, path := range []string{"/network", "/api/network"} { + w = httptest.NewRecorder() + mux.ServeHTTP(w, httptest.NewRequest("GET", path, nil)) + if w.Code != 404 { + t.Fatalf("disabled page: %s", path) + } + } +} +func TestNetworkRejectBadSnapshots(t *testing.T) { + s, a := trafficFixture(t) + if err := s.SaveNetwork(a.ID, NetworkSnapshot{At: time.Now().Unix() - 300}); err == nil { + t.Fatal("stale timestamp") + } + f, _ := parseNetworkFlow(networkNATLine) + f.SourcePort = 99999 + if err := s.SaveNetwork(a.ID, NetworkSnapshot{At: time.Now().Unix(), Flows: []NetworkFlow{f}}); err == nil { + t.Fatal("invalid port") + } +} +func TestNetworkCommandCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := networkCommand(ctx, "conntrack", "-L"); err == nil { + t.Fatal("cancelled command ran") + } +} + +func TestNetworkExternalAddresses(t *testing.T) { + for _, ip := range []string{"192.168.1.1", "127.0.0.1", "100.64.0.1", "224.0.0.1", "ff02::1", "fe80::1", "fd00::1", "::"} { + if networkExternalIP(ip) { + t.Fatalf("not external: %s", ip) + } + } + for _, ip := range []string{"8.8.8.8", "2001:4860:4860::8888"} { + if !networkExternalIP(ip) { + t.Fatalf("external: %s", ip) + } + } +} diff --git a/traffic.go b/traffic.go index bdac5b2..5440a13 100644 --- a/traffic.go +++ b/traffic.go @@ -76,6 +76,7 @@ const trafficSchema = `CREATE TABLE IF NOT EXISTS traffic_events ( country TEXT NOT NULL, UNIQUE(agent,event_id)); CREATE INDEX IF NOT EXISTS traffic_ts ON traffic_events(ts); CREATE INDEX IF NOT EXISTS traffic_filter ON traffic_events(domain,kind,ts); + CREATE TABLE IF NOT EXISTS network_sources(agent TEXT PRIMARY KEY, snapshot TEXT NOT NULL); CREATE TABLE IF NOT EXISTS traffic_sources(agent TEXT PRIMARY KEY, last_seen INTEGER NOT NULL, status_json TEXT NOT NULL);` func (s *Store) SaveTraffic(agent string, batch TrafficBatch) error { @@ -244,6 +245,7 @@ func (s *Store) TrafficView(domain, kind string, external bool) (TrafficView, er func trafficRoutes(mux *http.ServeMux, s *Store) { trafficSettingsRoute(mux, s) + networkRoutes(mux, s) mux.HandleFunc("/traffic", func(w http.ResponseWriter, r *http.Request) { if !s.TrafficSettings().Enabled { http.Error(w, "Карта трафика отключена. Включите её в настройках Dashboard.", 404) diff --git a/traffic_preview_test.go b/traffic_preview_test.go index 0d7960b..995745a 100644 --- a/traffic_preview_test.go +++ b/traffic_preview_test.go @@ -15,7 +15,7 @@ func TestTrafficPreview(t *testing.T) { t.Skip("manual visual preview") } s, a := trafficFixture(t) - s.SaveTrafficSettings(TrafficSettings{Enabled: true, AgentID: a.ID, ServerCountry: "RU"}) + s.SaveTrafficSettings(TrafficSettings{Enabled: true, NetworkEnabled: true, AgentID: a.ID, ServerCountry: "RU"}) mux := http.NewServeMux() trafficRoutes(mux, s) mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { @@ -38,11 +38,35 @@ func TestTrafficPreview(t *testing.T) { } events = append(events, TrafficEvent{ID: fmt.Sprintf("preview-%d-%d", i, j), Time: time.Now().Unix(), Kind: kind, IP: ip, Domain: "demo.example", Method: "GET", Path: "/demo", Status: 200, Reason: "ДЕМО · crowdsecurity/http-probing"}) } + flows := []NetworkFlow{} + for k := 0; k < 32; k++ { + f, _ := parseNetworkFlow(networkNATLine) + f.SourcePort = 51000 + k + f.ReplyDestinationPort = f.SourcePort + f.Source = []string{"8.8.8.8", "1.1.1.1", "20.194.96.114", "192.168.0.10", "51.158.145.102", "2001:4860:4860::8888"}[k%6] + f.ReplyDestination = f.Source + if k%3 == 0 { + f.Protocol = "udp" + f.State = "ASSURED" + f.DestinationPort = 53 + f.ReplySourcePort = 53 + } + if k%5 == 0 && f.Protocol == "tcp" { + f.State = "SYN_SENT" + } + flows = append(flows, f) + } + s.SaveNetwork(a.ID, NetworkSnapshot{At: time.Now().Unix(), Flows: flows, FirewallAt: time.Now().Unix(), Firewall: []FirewallCounter{{Name: "DEMO ip/crowdsec/drop #3", Packets: 12345, Bytes: 654321}}}) s.SaveTraffic(a.ID, TrafficBatch{Events: events, Files: 22, StartedAt: time.Now().Add(-time.Hour).Unix(), CrowdSecAt: time.Now().Unix()}) time.Sleep(3 * time.Second) } }() handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/network" { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.Write([]byte(strings.Replace(networkHTML, "PROXMOX / NETWORK OBSERVATORY", "ДЕМО / ИСКУССТВЕННЫЕ ДАННЫЕ", 1))) + return + } if r.URL.Path == "/traffic" { w.Header().Set("Content-Type", "text/html; charset=utf-8") w.Write([]byte(strings.Replace(trafficHTML, "PROXMOX / LIVE OBSERVATORY", "ДЕМО / ИСКУССТВЕННЫЕ ДАННЫЕ", 1))) diff --git a/traffic_settings.go b/traffic_settings.go index 6f557c5..4a846d7 100644 --- a/traffic_settings.go +++ b/traffic_settings.go @@ -9,9 +9,10 @@ import ( ) type TrafficSettings struct { - Enabled bool `json:"enabled"` - AgentID string `json:"agentId"` - ServerCountry string `json:"serverCountry"` + NetworkEnabled bool `json:"networkEnabled"` + Enabled bool `json:"enabled"` + AgentID string `json:"agentId"` + ServerCountry string `json:"serverCountry"` } func (s *Store) TrafficSettings() TrafficSettings { diff --git a/web.go b/web.go index 3f562c0..bcfd83c 100644 --- a/web.go +++ b/web.go @@ -284,7 +284,7 @@ func routes(collector *cpuCollector, store *Store) http.Handler { return } trafficSettings := store.TrafficSettings() - result := agentReportResponse{Accepted: true, TrafficEnabled: trafficSettings.Enabled && trafficSettings.AgentID == id} + result := agentReportResponse{Accepted: true, NetworkEnabled: trafficSettings.Enabled && trafficSettings.NetworkEnabled && trafficSettings.AgentID == id, TrafficEnabled: trafficSettings.Enabled && trafficSettings.AgentID == id} result.Commands, _ = store.DispatchAgentActions(id) if report.Version != version && version != "dev" { manifest, _, manifestErr := currentAgentUpdate() @@ -882,6 +882,8 @@ footer{color:#657581;font-size:12px;margin-top:17px}