1 Star 0 Fork 0

zhangjungang/beats

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
文件
克隆/下载
icmp.go 8.33 KB
一键复制 编辑 原始数据 按行查看 历史
Steffen Siering 提交于 2016-11-14 14:50 . More Packetbeat cleanups (#2972)
package icmp
import (
"expvar"
"net"
"time"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/logp"
"github.com/elastic/beats/packetbeat/flows"
"github.com/elastic/beats/packetbeat/protos"
"github.com/elastic/beats/packetbeat/publish"
"github.com/tsg/gopacket/layers"
)
type icmpPlugin struct {
sendRequest bool
sendResponse bool
localIps []net.IP
// Active ICMP transactions.
// The map key is the hashableIcmpTuple associated with the request.
transactions *common.Cache
transactionTimeout time.Duration
results publish.Transactions
}
type ICMPv4Processor interface {
ProcessICMPv4(flowID *flows.FlowID, hdr *layers.ICMPv4, pkt *protos.Packet)
}
type ICMPv6Processor interface {
ProcessICMPv6(flowID *flows.FlowID, hdr *layers.ICMPv6, pkt *protos.Packet)
}
const (
directionLocalOnly = iota
directionFromInside
directionFromOutside
)
// Notes that are added to messages during exceptional conditions.
const (
duplicateRequestMsg = "Another request with the same Id and Seq was received so this request was closed without receiving a response."
orphanedRequestMsg = "Request was received without an associated response."
orphanedResponseMsg = "Response was received without an associated request."
)
var (
unmatchedRequests = expvar.NewInt("icmp.unmatched_requests")
unmatchedResponses = expvar.NewInt("icmp.unmatched_responses")
duplicateRequests = expvar.NewInt("icmp.duplicate_requests")
)
func New(testMode bool, results publish.Transactions, cfg *common.Config) (*icmpPlugin, error) {
p := &icmpPlugin{}
config := defaultConfig
if !testMode {
if err := cfg.Unpack(&config); err != nil {
return nil, err
}
}
if err := p.init(results, &config); err != nil {
return nil, err
}
return p, nil
}
func (icmp *icmpPlugin) init(results publish.Transactions, config *icmpConfig) error {
icmp.setFromConfig(config)
var err error
icmp.localIps, err = common.LocalIPAddrs()
if err != nil {
logp.Err("icmp", "Error getting local IP addresses: %s", err)
icmp.localIps = []net.IP{}
}
logp.Debug("icmp", "Local IP addresses: %s", icmp.localIps)
var removalListener = func(k common.Key, v common.Value) {
icmp.expireTransaction(k.(hashableIcmpTuple), v.(*icmpTransaction))
}
icmp.transactions = common.NewCacheWithRemovalListener(
icmp.transactionTimeout,
protos.DefaultTransactionHashSize,
removalListener)
icmp.transactions.StartJanitor(icmp.transactionTimeout)
icmp.results = results
return nil
}
func (icmp *icmpPlugin) setFromConfig(config *icmpConfig) {
icmp.sendRequest = config.SendRequest
icmp.sendResponse = config.SendResponse
icmp.transactionTimeout = config.TransactionTimeout
}
func (icmp *icmpPlugin) ProcessICMPv4(
flowID *flows.FlowID,
icmp4 *layers.ICMPv4,
pkt *protos.Packet,
) {
typ := uint8(icmp4.TypeCode >> 8)
code := uint8(icmp4.TypeCode)
id, seq := extractTrackingData(4, typ, &icmp4.BaseLayer)
tuple := &icmpTuple{
icmpVersion: 4,
srcIP: pkt.Tuple.SrcIP,
dstIP: pkt.Tuple.DstIP,
id: id,
seq: seq,
}
msg := &icmpMessage{
ts: pkt.Ts,
Type: typ,
code: code,
length: len(icmp4.BaseLayer.Payload),
}
if isRequest(tuple, msg) {
if flowID != nil {
flowID.AddICMPv4Request(id)
}
icmp.processRequest(tuple, msg)
} else {
if flowID != nil {
flowID.AddICMPv4Response(id)
}
icmp.processResponse(tuple, msg)
}
}
func (icmp *icmpPlugin) ProcessICMPv6(
flowID *flows.FlowID,
icmp6 *layers.ICMPv6,
pkt *protos.Packet,
) {
typ := uint8(icmp6.TypeCode >> 8)
code := uint8(icmp6.TypeCode)
id, seq := extractTrackingData(6, typ, &icmp6.BaseLayer)
tuple := &icmpTuple{
icmpVersion: 6,
srcIP: pkt.Tuple.SrcIP,
dstIP: pkt.Tuple.DstIP,
id: id,
seq: seq,
}
msg := &icmpMessage{
ts: pkt.Ts,
Type: typ,
code: code,
length: len(icmp6.BaseLayer.Payload),
}
if isRequest(tuple, msg) {
if flowID != nil {
flowID.AddICMPv6Request(id)
}
icmp.processRequest(tuple, msg)
} else {
if flowID != nil {
flowID.AddICMPv6Response(id)
}
icmp.processResponse(tuple, msg)
}
}
func (icmp *icmpPlugin) processRequest(tuple *icmpTuple, msg *icmpMessage) {
logp.Debug("icmp", "Processing request. %s", tuple)
trans := icmp.deleteTransaction(tuple.Hashable())
if trans != nil {
trans.notes = append(trans.notes, duplicateRequestMsg)
logp.Debug("icmp", duplicateRequestMsg+" %s", tuple)
duplicateRequests.Add(1)
icmp.publishTransaction(trans)
}
trans = &icmpTransaction{ts: msg.ts, tuple: *tuple}
trans.request = msg
if requiresCounterpart(tuple, msg) {
icmp.transactions.Put(tuple.Hashable(), trans)
} else {
icmp.publishTransaction(trans)
}
}
func (icmp *icmpPlugin) processResponse(tuple *icmpTuple, msg *icmpMessage) {
logp.Debug("icmp", "Processing response. %s", tuple)
revTuple := tuple.Reverse()
trans := icmp.deleteTransaction(revTuple.Hashable())
if trans == nil {
trans = &icmpTransaction{ts: msg.ts, tuple: revTuple}
trans.notes = append(trans.notes, orphanedResponseMsg)
logp.Debug("icmp", orphanedResponseMsg+" %s", tuple)
unmatchedResponses.Add(1)
}
trans.response = msg
icmp.publishTransaction(trans)
}
func (icmp *icmpPlugin) direction(t *icmpTransaction) uint8 {
if !icmp.isLocalIP(t.tuple.srcIP) {
return directionFromOutside
}
if !icmp.isLocalIP(t.tuple.dstIP) {
return directionFromInside
}
return directionLocalOnly
}
func (icmp *icmpPlugin) isLocalIP(ip net.IP) bool {
if ip.IsLoopback() {
return true
}
for _, localIP := range icmp.localIps {
if ip.Equal(localIP) {
return true
}
}
return false
}
func (icmp *icmpPlugin) getTransaction(k hashableIcmpTuple) *icmpTransaction {
v := icmp.transactions.Get(k)
if v != nil {
return v.(*icmpTransaction)
}
return nil
}
func (icmp *icmpPlugin) deleteTransaction(k hashableIcmpTuple) *icmpTransaction {
v := icmp.transactions.Delete(k)
if v != nil {
return v.(*icmpTransaction)
}
return nil
}
func (icmp *icmpPlugin) expireTransaction(tuple hashableIcmpTuple, trans *icmpTransaction) {
trans.notes = append(trans.notes, orphanedRequestMsg)
logp.Debug("icmp", orphanedRequestMsg+" %s", &trans.tuple)
unmatchedRequests.Add(1)
icmp.publishTransaction(trans)
}
func (icmp *icmpPlugin) publishTransaction(trans *icmpTransaction) {
if icmp.results == nil {
return
}
logp.Debug("icmp", "Publishing transaction. %s", &trans.tuple)
event := common.MapStr{}
// common fields - group "env"
event["client_ip"] = trans.tuple.srcIP
event["ip"] = trans.tuple.dstIP
// common fields - group "event"
event["@timestamp"] = common.Time(trans.ts) // timestamp of the first packet
event["type"] = "icmp" // protocol name
event["path"] = trans.tuple.dstIP // what is requested (dst ip)
if trans.HasError() {
event["status"] = common.ERROR_STATUS
} else {
event["status"] = common.OK_STATUS
}
if len(trans.notes) > 0 {
event["notes"] = trans.notes
}
// common fields - group "measurements"
responsetime, hasResponseTime := trans.ResponseTimeMillis()
if hasResponseTime {
event["responsetime"] = responsetime
}
switch icmp.direction(trans) {
case directionFromInside:
if trans.request != nil {
event["bytes_out"] = trans.request.length
}
if trans.response != nil {
event["bytes_in"] = trans.response.length
}
case directionFromOutside:
if trans.request != nil {
event["bytes_in"] = trans.request.length
}
if trans.response != nil {
event["bytes_out"] = trans.response.length
}
}
// event fields - group "icmp"
icmpEvent := common.MapStr{}
event["icmp"] = icmpEvent
icmpEvent["version"] = trans.tuple.icmpVersion
if trans.request != nil {
request := common.MapStr{}
icmpEvent["request"] = request
request["message"] = humanReadable(&trans.tuple, trans.request)
request["type"] = trans.request.Type
request["code"] = trans.request.code
// TODO: Add more info. The IPv4/IPv6 payload could be interesting.
// if icmp.SendRequest {
// request["payload"] = ""
// }
}
if trans.response != nil {
response := common.MapStr{}
icmpEvent["response"] = response
response["message"] = humanReadable(&trans.tuple, trans.response)
response["type"] = trans.response.Type
response["code"] = trans.response.code
// TODO: Add more info. The IPv4/IPv6 payload could be interesting.
// if icmp.SendResponse {
// response["payload"] = ""
// }
}
icmp.results.PublishTransaction(event)
}
Loading...
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/zhangjungang/beats.git
git@gitee.com:zhangjungang/beats.git
zhangjungang
beats
beats
v5.6.13

搜索帮助

0d507c66 1850385 C8b1a773 1850385