【FlowTarget】流量检测系统的实现记录

FlowTarget是MercyCloud研发的流量检测系统，采用GO+PHP语言编写，经现网环境长期运行测试而成的一款自用网络流量分析系统

MercyCloud FlowTarget

代码全部不开源，系统也不对外出售，本文章只做部分技术实现分析记录。（处理代码有删减 只保留大致内容）

0.类型定义

type LogPacket struct {
	SrcIP, DstIP     string
	SrcPort, DstPort string
	SrcMAC, DstMAC   string
	Protocol, Type   []string //7层应用识别、TCP、UDP识别、传输协议方法识别
	Host, Result     string
	Time             string
}


var (
	device       string        = "enp7s0"
	snapshot_len int32         = 1024
	promiscuous  bool          = true // 网卡混杂模式
	timeout      time.Duration = -1 * time.Second
	logStatus    bool          = false
)


	var (
		handle    *pcap.Handle
		err       error
		eth       layers.Ethernet
		ip4       layers.IPv4
		ip6       layers.IPv6
		tcp       layers.TCP
		udp       layers.UDP
		dns       layers.DNS
		logPacket LogPacket
	)

1.获取流量

本项目采用由Google编写GoPacket包来获取流量

此处代码为使用PACP采集流量，如果为了性能以及大流量可改为使用pfring/afpacket，将采集到的流量计入到

//启用数据库连接
    client := influxdb2.NewClient(conf.InfluxdbUrl, conf.InfluxdbToken)
    defer client.Close()

//开始使用PCAP抓数据包
	handle, err = pcap.OpenLive(device, snapshot_len, promiscuous, timeout)
	if err != nil {
		log.Fatal(err)
	}
	defer handle.Close()

	packetSource := gopacket.NewPacketSource(handle, handle.LinkType())

2.初步分析流量

for packet := range packetSource.Packets() {

		logPacket.Time = time.Now().Format("2006-01-02 15:04:05")
		parser := gopacket.NewDecodingLayerParser(
			layers.LayerTypeEthernet,
			&eth,
			&ip4,
			&ip6,
			&tcp,
			&udp,
			&dns,
		)

		foundLayerTypes := []gopacket.LayerType{}
		// ——对于无法通过传输层协议或应用层协议报文头内容区分会话特征的数据流量，ISMS 应以数据 流(源 IP、目的 IP、源端口、目的端口均相同，速率大于 1 帧/s 且持续时间大于 10s 的数据流量)为 单位记录访问日志，记录信息至少应包括源 IP、目的 IP、源端口、目的端口、访问时间(起始时间，

		err := parser.DecodeLayers(packet.Data(), &foundLayerTypes)
		if err != nil {
			//解析到无法识别到数据包(如果是ARP等不支持等报文则记录下就离开)
		}

		writeAPI := client.WriteAPI(conf.InfluxdbOrg, conf.InfluxdbBucket)
		defer client.Close()

		for _, layerType := range foundLayerTypes {
			switch layerType {
			//处理二层信息 VRRP...
			case layers.LayerTypeEthernet:
				logPacket.SrcMAC = eth.SrcMAC.String()
				logPacket.DstMAC = eth.DstMAC.String()
				//处理部分特殊协议
			// case layers.LayerTypeOSPF:
			// case layers.LayerTypeBFD:
			// case layers.LayerTypeCiscoDiscovery:
			// case layers.LayerTypeDot1Q:
			// case layers.LayerTypePPP:
			// case layers.LayerTypeSTP:
			case layers.LayerTypeARP: //处理ARP报文
				logPacket.Type = append(logPacket.Type, layerType.String())
				continue //如果遇到以上报文记录即可，不需要DPI分析
				//TODO 检测并预警ARP泛洪
			case layers.LayerTypeIPv4, layers.LayerTypeICMPv4: //记录IP
				
			case layers.LayerTypeIPv6, layers.LayerTypeICMPv6:
				
			case layers.LayerTypeDNS: //处理DNS报文

				p := influxdb2.NewPoint("DNS",
					map[string]string{},
					map[string]interface{}{
						"Questions":   dns.Questions,
						"Authorities": dns.Authorities,
						"Answers":     dns.Answers,
						"SrcIP":       logPacket.SrcIP,
						"DstIP":       logPacket.DstIP,
						"SrcMAC":      logPacket.SrcMAC,
						"DstMAC":      logPacket.DstMAC,
					},
					time.Now())
				writeAPI.WritePoint(p)
			case layers.LayerTypeTCP, layers.LayerTypeUDP: // 处理TCP和UDP
				
			}

			writeAPI.Flush()

		}

 到这里，恭喜你按照七层模型，前四层你已经处理到差不多了，剩下你就可以处理四层以上到内容了

以上部分代码建议处理1G以内带宽使用，1G以上请升级你到设备配置及使用pfring/afpacket

3. 流量分析

在这里你可以将流量扔给nDPI/libprotoident或其他开源DPI服务进行分析，一开始咱们是使用goDPI这个开源包抓的

这个是goDPI的案例代码，

package main

import (
	"fmt"
	"github.com/mushorg/go-dpi"
	"github.com/mushorg/go-dpi/types"
	"github.com/mushorg/go-dpi/utils"
)

func main() {
	godpi.Initialize()
	defer godpi.Destroy()
	packets, err := utils.ReadDumpFile("/tmp/http.cap")
	if err != nil {
		fmt.Println(err)
	} else {
		for packet := range packets {
             #可并入上面代码 Start ，packet和上面是一样的意思

			flow, _ := godpi.GetPacketFlow(packet)
			result := godpi.ClassifyFlow(flow)
			if result.Protocol != types.Unknown {
				fmt.Println(result.Source, "detected protocol", result.Protocol)
			} else {
				fmt.Println("No detection was made")
			}
           
            #可并入上面代码结束
		}
	}
}

不出以外的话，走到这里，你已经实现了基础到流量记录以最基础的分析

-1. 记录流量

package packet

import (
	"os"
	"time"

	"github.com/google/gopacket"
	"github.com/google/gopacket/layers"
	"github.com/google/gopacket/pcapgo"
)

//导出PCAP文件
func ExportPack(logPacket LogPacket, packet gopacket.Packet) {
	f, _ := os.Create("dump/" + device + "-" + logPacket.SrcIP + "-" + time.Now().Format("2006-01-02-15:04:05") + ".pcap")
	w := pcapgo.NewWriter(f)
	w.WriteFileHeader(1024, layers.LinkTypeEthernet)
	defer f.Close()
	w.WritePacket(packet.Metadata().CaptureInfo, packet.Data())
}

-2. 阻断流量

当检测到违法或奇怪到流量的时候咱们需要阻断流量，旁路部署时的阻断流量的方法

原理参考

这段代码来自案例代码

err = handle.WritePacketData(outgoingPacket)
    if err != nil {
        log.Fatal(err)
    }
 
    // This time lets fill out some information
    ipLayer := &layers.IPv4{
        SrcIP: 起源IP,
        DstIP: 目标IP,
    }
    ethernetLayer := &layers.Ethernet{
        SrcMAC: 起源MAC,
        DstMAC: 目标MAC,
    }
    tcpLayer := &layers.TCP{
        SrcPort: 起源端口,
        DstPort: 目标端口,
    }
    // And create the packet with the layers
    buffer = gopacket.NewSerializeBuffer()
    gopacket.SerializeLayers(buffer, options,
        ethernetLayer,
        ipLayer,
        tcpLayer,
        gopacket.Payload(#TCP RST包或其他数据包/例如HTTP未备案页面返回),
    )
    outgoingPacket = buffer.Bytes()