【技术干货】使用 gopacket 从网络捕获及重组数据包

$ mkdir gopacket-tcp-reassembly-test$ cd gopacket-tcp-reassembly-test/$ go mod init gopacket-tcp-reassembly-test$ go get -u github.com/google/[email protected]

package main
import (  "bufio"  "bytes"  "flag"  "github.com/google/gopacket"  "github.com/google/gopacket/layers"  "github.com/google/gopacket/pcap"  "github.com/google/gopacket/tcpassembly"  "github.com/google/gopacket/tcpassembly/tcpreader"  "io"  "log"  "net/http"  "sync"  "time")
// Cache 用于检索响应对应的请求。这样做的原因包括：// 1. 解析 HTTP 响应时，需要相应的请求对象// 2. 将响应和请求配对type Cache struct {  mtx sync.RWMutex  // 格式为：{网络层哈希值: {传输层哈希值: 请求对象, ...}, ...}  m map[uint64]map[uint64]*http.Request  // 用于判断 Flow 是请求还是响应  requests map[string]struct{}}
func NewCache() *Cache {  return &Cache{    m:        make(map[uint64]map[uint64]*http.Request),    requests: make(map[string]struct{}),  }}
// Get 用于在解析响应前，从 Cache 中获取请求func (c *Cache) Get(netHash, transportHash uint64) *http.Request {  c.mtx.RLock()  defer c.mtx.RUnlock()  if transportMap, found := c.m[netHash]; found {    if r, found := transportMap[transportHash]; found {      return r    }  }  return nil}
// Set 用于在解析完请求后，将其保存到 Cache 中func (c *Cache) Set(net, transport gopacket.Flow, r *http.Request) {  c.mtx.Lock()  defer c.mtx.Unlock()  transportMap, found := c.m[net.FastHash()]  if found {    transportMap[transport.FastHash()] = r  } else {    c.m[net.FastHash()] = map[uint64]*http.Request{transport.FastHash(): r}  }  c.requests[net.Src().String()+":"+transport.Src().String()] = struct{}{}}
// Delete 用于从 Cache 中删除条目func (c *Cache) Delete(net, transport gopacket.Flow) {  c.mtx.Lock()  defer c.mtx.Unlock()  // 如果是请求，那么删除 requests 中的条目  key := net.Src().String() + ":" + transport.Src().String()  if _, found := c.requests[key]; found {    delete(c.requests, key)    return  }  // 如果是响应，那么删除缓存的请求对象  transportMap, found := c.m[net.FastHash()]  if !found {    return  }  delete(transportMap, transport.FastHash())  if len(transportMap) == 0 {    delete(c.m, net.FastHash())  }}
var cache = NewCache()
// httpStream 真正地处理 HTTP 请求的解码type httpStream struct {  net, transport gopacket.Flow  r              tcpreader.ReaderStream}
func (h *httpStream) run() {  buf := bufio.NewReader(&h.r)  for {    var err error    // Magic Number 用于确定协议。比如 HTTP 请求报文、HTTP 响应报文    var magicNumber []byte    var resp *http.Response    var req *http.Request    isResponse := false    // PEEK（返回下 N 个字节，但不推进 Reader）出前 2 个字节，以确定 Magic Number    magicNumber, err = buf.Peek(2)    if err != nil {      // 如果遇到其它错误，那么先打印错误信息，再返回      if err != io.EOF {        log.Printf("Error peeking magic number, %sn", err)      } else {        // 如果遇到 EOF，那么清理缓存        cache.Delete(h.net, h.transport)      }      return    }    // 如果前 2 个字节是 HT，那么认为该 Flow 是 HTTP 响应    if bytes.Equal(magicNumber, []byte{'H', 'T'}) {      isResponse = true    }    if isResponse {      resp, err = http.ReadResponse(buf, cache.Get(h.net.FastHash(), h.transport.FastHash()))    } else {      if req, err = http.ReadRequest(buf); err == nil {        cache.Set(h.net, h.transport, req)      }    }    if err == io.EOF {      // 必须读到 EOF...非常重要！      cache.Delete(h.net, h.transport)      return    } else if err != nil {      log.Printf("Error reading stream, %s, %s, %s", h.net, h.transport, err)      continue    }    if req != nil {      body, _ := io.ReadAll(req.Body)      _ = req.Body.Close()      log.Printf("http request: %#vn", *req)      log.Println(string(body))    } else {      body, _ := io.ReadAll(resp.Body)      // 读完 resp.Body 时，必须调用 resp.Body.Close      _ = resp.Body.Close()      log.Printf("http response: %#vn", *resp)      log.Println(string(body))    }  }}
// 使用 tcpassembly.StreamFactory 和 tcpassembly.Stream 接口，构建 HTTP 请求解析器。// httpStreamFactory 实现 tcpassembly.StreamFactorytype httpStreamFactory struct{}
func (h *httpStreamFactory) New(net, transport gopacket.Flow) tcpassembly.Stream {  hs := &httpStream{    net:       net,    transport: transport,    r:         tcpreader.NewReaderStream(),  }  // 重要...必须保证读取 reader 流的数据  go hs.run()  // ReaderStream 实现 tcpassembly.Stream, 因此可以返回指向它的指针  return &hs.r}
func main() {  // 解析命令行参数  var iface = flag.String("i", "eth0", "Interface to get packets from")  var snaplen = flag.Int("s", 1600, "SnapLen for pcap packet capture")  flag.Parse()
  var handle *pcap.Handle  var err error  // pcap.OpenLive 打开设备，返回 *pcap.Handle。  // 它接受设备名称（eth0），每个数据包的最大大小（snaplen），是否将接口设置为混杂模式（即是否接收目的地址不为本机的包），以及超时作为参数。  // 警告：该函数仅接受毫秒时间戳。对于纳秒解决方案，使用 pcap.InactiveHandle。  // 如果将 timeout 设置为 30s，那么每 30s 刷新一次数据包；设置成负数，将立刻刷新数据包  handle, err = pcap.OpenLive(*iface, int32(*snaplen), true, time.Second)  if err != nil {    log.Panic(err)  }  // 务必释放 Handle  defer handle.Close()
  // 设置 TCP 流重组  // 1. 创建 httpStreamFactory 结构体，实现 tcpassembly.StreamFactory 接口  streamFactory := &httpStreamFactory{}  // 2. 创建连接池  streamPool := tcpassembly.NewStreamPool(streamFactory)  // 3. 创建重组器  assembler := tcpassembly.NewAssembler(streamPool)  // 创建数据包源。  // handle.LinkType() 表示从 2 层以太网链路上抓取数据包  packetSource := gopacket.NewPacketSource(handle, handle.LinkType())  // 从该 Channel 中，持续地读取数据包  packets := packetSource.Packets()  ticker := time.Tick(time.Minute)  for {    select {    case packet := <-packets:      if packet.NetworkLayer() == nil || packet.TransportLayer() == nil ||        packet.TransportLayer().LayerType() != layers.LayerTypeTCP {        log.Println("Unusable packet")        continue      }      tcp := packet.TransportLayer().(*layers.TCP)      // 4. 将数据包添加到重组器中      assembler.AssembleWithTimestamp(packet.NetworkLayer().NetworkFlow(), tcp, packet.Metadata().Timestamp)    case <-ticker:      // 每隔 1 分钟，刷新之前 2 分钟内不活跃的连接      assembler.FlushOlderThan(time.Now().Add(time.Minute * -2))    }  }}

$ sudo apt install -y nginx# 测试 Nginx 是否成功启动$ curl http://127.0.0.1

# 在项目目录下执行$ sudo go run main.go -i lo

$ curl http://127.0.0.1

package main
import (  "fmt"  "github.com/google/gopacket/pcap"  "log")
func main() {  // 尝试迭代当前机器上的所有接口  devices, err := pcap.FindAllDevs()  if err != nil {    log.Panic(err)  }  // 打印设备信息  fmt.Println("Devices found:")  for _, device := range devices {    fmt.Println("nName: ", device.Name)    fmt.Println("Description: ", device.Description)    fmt.Println("Devices addresses:")    for _, address := range device.Addresses {      fmt.Println("- IP address: ", address.IP)      fmt.Println("- Subnet mask: ", address.Netmask)    }  }}