5.1. Sniffer原理及驱动定制

sniffer默认使用的是基于bro流量分析的数据源驱动，但也提供了如：redis、kafka、rabbitmq、logstash、UDPServer、syslog、file等其他数据源驱动的demo供大家参考（无法直接使用，需要根据源数据的格式进行调整）

工作方式

工作方式.png

生成消息

  1. **此部分需根据自己业务进行修改定制**

以Nginx + Logstash为例（Nginx输出日志格式配置仅供参考）

  1. 1) 行格式
  2. log_format log_line '"$remote_addr" "$remote_port" "$server_addr" "$server_port" "$request_length" \
  3. "$content_length" "$body_bytes_sent" "$request_uri" "$host" "$http_user_agent" "$status" "$http_cookie" \
  4. "$request_method" "$http_referer" "$http_x_forwarded_for" "$request_time" "$sent_http_set_cookie" \
  5. "$content_type" "$upstream_http_content_type" "$request_data"';
  7. 例：
  8. <14>Jun 26 17:13:21 10-10-92-198 NGINX[26113]: "114.242.250.233" "65033" "10.10.92.198" "80" "726" "134" \
  9. "9443" "/gateway/shop/getStroeForDistance" "m.lechebang.com" "Mozilla/5.0 (iPhone; CPU iPhone OS 8_1_3
  10. like Mac OS X) AppleWebKit/600.1.4 (KHTML, like Gecko) Mobile/12B466 MicroMessenger/6.1.4 NetType/3G+" "200" \
  11. "-" "POST" "http://m.lechebang.com/webapp/shop/list?cityId=10101&locationId=0&brandTypeId=\
  12. 6454&maintenancePlanId=227223&oilInfoId=3906" "0.114"
  14. 2) json格式
  15. log_format log_json '{ "@timestamp": "$time_local", '
  16. '"remote_addr": "$remote_addr", '
  17. '"referer": "$http_referer", '
  18. '"request": "$request", '
  19. '"status": $status, '
  20. '"bytes": $body_bytes_sent, '
  21. '"agent": "$http_user_agent", '
  22. '"x_forwarded": "$http_x_forwarded_for", '
  23. '"up_addr": "$upstream_addr",'
  24. '"up_host": "$upstream_http_host",'
  25. '"up_resp_time": "$upstream_response_time",'
  26. '"request_time": "$request_time"'
  27. ' }';
  29. 例：
  30. { "@timestamp": "12/Dec/2017:14:30:40 +0800", "remote_addr": "10.88.122.108", "referer": "-", "request": "GET / HTTP/1.1", "status": 304, "bytes":0, "agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.84 Safari/537.36", "x_forwarded": "-", "up_addr": "-","up_host": "-","up_resp_time": "-","request_time": "0.000" }
  33. Nginx日志配置请参考：
  34.     http://nginx.org/en/docs/http/ngx_http_log_module.html#log_format
  36. 注：
  37.     日志输出的详细字段、字段顺序、字段名，需要与sniffer中对应驱动的消息格式化处理逻辑一致

部署Logstash客户端，抽取Nginx日志输出:

  1. #logstash配置示例:
  2. input {
  3.     file {
  4.         path => ['/data/nginx/logs/access.log']
  5.         start_position => "beginning"
  6.         codec => "json"
  7.         tags => ['user']
  8.         type => "nginx"
  9.     }
  10. }
  11. output {
  12.     if [type] == "nginx" {
  13.         #Sniffer logstash驱动模式
  14.         tcp {
  15.             host => "127.0.0.1"
  16.             port => "5044"
  17.             key => "nginx"
  18.             db => "10"
  19.             data_type => "list"
  20.             codec => "line"
  21.         },
  22.         #Sniffer redislist驱动模式
  23.         redis {
  24.             host => "127.0.0.1"
  25.             port => "6379"
  26.             key => "nginx_access_log"
  27.             db => "0"
  28.             data_type => "list"
  29.             codec => "line"
  30.         }
  31.     }
  32. }

读取数据

  1. **此部分需根据自己业务进行修改定制**

修改sniffer.conf

  1. #支持同时多源
  2. sources: [logstash,redislist]
  4. #对应redislistdriver.py
  5. redislist:
  6.     driver: redislist
  7.     host: 127.0.0.1
  8.     port: 6379
  9.     interface: any
  10.     instances: 1
  11.     parser:
  12.         name: test
  13.         module: testparser
  15. #对应logstashdriver.py
  16. #其实是TCPServer
  17. logstash:
  18.     driver: logstash
  19.     port: 5044
  20.     instances: 1
  21.     interface: any
  22.     parser:
  23.         name: test
  24.         module: testparser
  26. #对应rabbitmqdriver.py
  27. rabbitmq:
  28.     driver: rabbitmq
  29.     amqp_url: redis://localhost:6379/
  30.     queue_name: test_queue
  31.     exchange_name: test_queue
  32.     exchange_type: direct
  33.     durable: true
  34.     routing_key: test
  35.     instances: 1
  36.     interface: any
  37.     parser:
  38.         name: test
  39.         module: testparser
  41. #对应brohttpdriver.py
  42. default:
  43.     driver: bro
  44.     interface: eth0
  45.     ports: [80, 81, 1080, 3128, 8000, 8080, 8888, 9001, 8081]
  46.     start_port: 48880
  47.     instances: 1
  48.     parser:
  49.         name: test
  50.         module: testparser
  52. #对应syslogdriver.py
  53. syslog:
  54.     driver: syslog
  55.     interface: eth0
  56.     port: 9514
  57.     parser:
  58.         name: test
  59.         module: testparser
  61. #对应kafkadriver.py
  62. kafka:
  63.     driver: kafka
  64.     interface: any
  65.     bootstrap_servers: 127.0.0.1:9092
  66.     parser:
  67.         name: test
  68.         module: testparser
  70. ...其他省略...

驱动定制(消息格式化)

所有驱动位于:

  1. 目录：
  2.     sniffer/nebula_sniffer/nebula_sniffer/drivers/
  3.     #bro驱动
  4.         brohttpdriver.py
  5.     #基于文件的驱动
  6.         filedriver.py
  7.     #kafka驱动
  8.         kafkadriver.py  
  9.     #logstash驱动
  10.         logstashdriver.py
  11.     #UDPServer驱动
  12.         pktdriver.py
  13.     #rabbitmq驱动
  14.         rabbitmqdriver.py
  15.     #redis驱动
  16.         redislistdriver.py
  17.     #syslog驱动
  18.         syslogdriver.py
  19.         syslogtextdriver.py
  20.     #tshark驱动
  21.         tsharkhttpsdriver.py

sniffer会根据sniffer配置加载对应驱动：

  1. def get_driver(config, interface, parser, idx):
  2.     """ global c """
  4.     from complexconfig.configcontainer import configcontainer
  6.     #不同driver，初始化方式不同
  7.     name = config['driver']
  8.     if name == "bro":
  9.         from nebula_sniffer.drivers.brohttpdriver import BroHttpDriver
  10.         embedded = config.get("embedded", True)
  11.         ports = config['ports']
  12.         from nebula_sniffer.utils import expand_ports
  13.         ports = expand_ports(ports)  # extend it
  14.         start_port = int(config['start_port'])
  15.         bpf_filter = config.get("bpf_filter", "")
  17.         home = configcontainer.get_config("sniffer").get_string("sniffer.bro.home")
  19.         if ports and home:
  20.             driver = BroHttpDriver(interface=interface, embedded_bro=embedded, idx=idx, ports=ports, bro_home=home,
  21.                                    start_port=start_port, bpf_filter=bpf_filter)
  22.         elif ports:
  23.             driver = BroHttpDriver(interface=interface, embedded_bro=embedded, idx=idx, ports=ports,
  24.                                    start_port=start_port, bpf_filter=bpf_filter)
  25.         elif home:
  26.             driver = BroHttpDriver(interface=interface, embedded_bro=embedded, idx=idx, bro_home=home,
  27.                                    start_port=start_port, bpf_filter=bpf_filter)
  28.         else:
  29.             driver = BroHttpDriver(interface=interface, embedded_bro=embedded, idx=idx,
  30.                                    start_port=start_port, bpf_filter=bpf_filter)
  31.         return driver
  33.     if name == "tshark":
  34.         from nebula_sniffer.drivers.tsharkhttpsdriver import TsharkHttpsDriver
  35.         interface = interface
  36.         ports = config["ports"]
  37.         bpf_filter = config.get("bpf_filter", "")
  38.         if ports:
  39.             driver = TsharkHttpsDriver(interface=interface, ports=ports, bpf_filter=bpf_filter)
  40.         else:
  41.             driver = TsharkHttpsDriver(interface=interface, bpf_filter=bpf_filter)
  42.         return driver
  44.     if name == "syslog":
  45.         from nebula_sniffer.drivers.syslogdriver import SyslogDriver
  46.         port = int(config["port"])
  47.         driver = SyslogDriver(port)
  48.         return driver
  50.     if name == "packetbeat":
  51.         from nebula_sniffer.drivers.pktdriver import PacketbeatDriver
  52.         port = int(config["port"])
  53.         driver = PacketbeatDriver(port)
  54.         return driver
  56.     if name == "redislist":
  57.         from nebula_sniffer.drivers.redislistdriver import RedisListDriver
  58.         host = config["host"]
  59.         port = int(config['port'])
  60.         password = config.get('password', '')
  61.         driver = RedisListDriver(host, port, password)
  62.         return driver
  64.     if name == "logstash":
  65.         from nebula_sniffer.drivers.logstashdriver import LogstashDriver
  66.         port = int(config['port'])
  67.         driver = LogstashDriver(port)
  68.         return driver
  70.     if name == "rabbitmq":
  71.         from nebula_sniffer.drivers.rabbitmqdriver import RabbitmqDriver
  72.         amqp_url = config['amqp_url']
  73.         queue_name = config['queue_name']
  74.         exchange_name = config['exchange_name']
  75.         exchange_type = config['exchange_type']
  76.         durable = bool(config['durable'])
  77.         routing_key = config['routing_key']
  78.         driver = RabbitmqDriver(amqp_url, queue_name, exchange_name, exchange_type, durable, routing_key)
  79.         return driver
  81.     if name == "kafka":
  82.         from nebula_sniffer.drivers.kafkadriver import KafkaDriver
  83.         topics = config['topics']
  84.         #config['bootstrap_servers']
  85.         #kafka支持的配置参数
  86.         #请参考python kafka库的使用方法
  87.         driver = KafkaDriver(topics, **config)
  88.         return driver
  90.     return None

下面是logstash驱动示例：

  1. #其他部分代码略过，仅贴出核心代码
  2. #此处示例是对nginx json格式log的处理
  4. class LogstashDriver(Driver):
  6.     ...其他省略...
  8.     #对logstash客户端发送过来的消息进行格式化
  9.     def _recv_msg_fn_in(self, msg, addr):
  10.         """
  11.         log-line：
  12.         36.7.130.69 - [16/Jul/2017:23:58:42 +0800] ffp.hnair.com 1 "GET /FFPClub/upload/index/e9b1bb4a-e1dd-47e1-8699-9828685004b4.jpg HTTP/1.1" 200 487752 - "http://ffp.hnair.com/FFPClub/cn/index.html" "Mozilla/5.0 (Windows NT 6.1; Trident/7.0; rv:11.0; NetworkBench/8.0.1.309-4992258-2148837) like Gecko" "-"
  14.         log-json：
  15.         {"@timestamp_scs": "2017-09-21T12:18:17+08:00", "scs_request_uri": "/site-wap/my/transferin.htm",
  16.          "scs_status": "200", "scs_bytes_sent": "26829", "scs_upstream_cache_status": "-", "scs_request_time": "0.570",
  17.          "scs_upstream_response_time": "0.570", "scs_host": "pay.autohome.com.cn", "scs_remote_addr": "10.20.2.23",
  18.          "scs_server_addr": "10.20.252.33", "scs_upstream_addr": "10.20.252.20:8253", "scs_upstream_status": "200",
  19.          "scs_http_referer": "https://pay.autohome.com.cn/site-wap/activity/upin.htm?__sub_from=A2002027782510100",
  20.          "scs_http_user_agent": "Mozilla/5.0 (Linux; Android 5.1; OPPO A59m Build/LMY47I; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/43.0.2357.121 Mobile Safari/537.36 autohomeapp/1.0+%28auto_android%3B8.4.0%3BOi-CR_rnyywoODk5jv0ve5luKLNxl7AfnEsGsBBPNdWdDtP8ZDdRHA3ePFiDlWOr%3B5.1%3BOPPO%2BA59m%29 auto_android/8.4.0 nettype/wifi",
  21.          "scs_http_X_Forwarded_For": "220.166.199.167"}
  23.         """
  25.         try:
  26.             if not msg:
  27.                 return
  28.             msg = msg.strip()
  29.             if not msg:
  30.                 return
  31.             self.logger.debug("get log msg %s from address %s", msg, addr)
  33.             #收到消息后,根据消息格式进行解析处理
  34.             try:
  35.                 msg = json.loads(msg)
  36.             except Exception as e:
  37.                 return
  39.             #从消息中提取字段数据
  40.             c_ip = msg.get('scs_http_X_Forwarded_For', '')
  41.             if c_ip:
  42.                 c_ip_group = c_ip.split(',')
  43.                 if c_ip_group:
  44.                     c_ip = c_ip_group[-1]
  45.             c_port = 0
  46.             s_port = 80
  47.             c_bytes = 0
  48.             s_bytes = msg.get('scs_bytes_sent', 0)
  49.             if s_bytes == '-':
  50.                 s_bytes = 0
  51.             else:
  52.                 s_bytes = int(s_bytes)
  53.             status = int(msg.get('scs_status', 0))
  54.             req_body = ''
  56.             args = dict()
  57.             args["method"] = 'GET'
  58.             args["host"] = msg.get('scs_host', '').lower()
  59.             args["uri"] = msg.get('scs_request_uri', '').lower()
  60.             args["referer"] = msg.get('scs_http_referer', '').lower()
  61.             args["user_agent"] = msg.get('scs_http_user_agent', '').lower()
  62.             args["status_code"] = status
  63.             args["status_msg"] = ""
  64.             args["source_ip"] = c_ip
  65.             args["source_port"] = c_port
  66.             args["dest_ip"] = ''
  67.             args["dest_port"] = s_port
  69.             request_time = 0.0
  70.             try:
  71.                 ctime = msg['@timestamp_scs']
  72.                 ctime = ctime.replace('T', ' ').replace('+08:00', '')
  73.                 time_array = time.strptime(ctime, "%Y-%m-%d %H:%M:%S")
  74.                 # 转换成时间戳
  75.                 request_time = time.mktime(time_array)
  76.             except Exception as e:
  77.                 pass
  79.             args["req_time"] = int(request_time * 1000)
  81.             # headers
  82.             args["req_headers"] = {}
  84.             args["resp_headers"] = {}
  86.             # no body for logstash
  87.             args["log_body"] = False
  88.             args["req_body"] = ""
  89.             args["resp_body"] = ""
  90.             args["req_body_len"] = c_bytes
  91.             args["resp_body_len"] = s_bytes
  92.             args["req_content_type"] = ''
  93.             args["resp_content_type"] = ''
  94.             args["req_body"] = req_body
  96.             args["debug_processing"] = False
  98.             self.logger.debug("get http data from logstash: %s", args)
  100.             try:
  101.                 #最终格式化为Httpmsg格式
  102.                 new_msg = HttpMsg(**args)
  103.             except BeFilteredException as bfe:
  104.                 return
  105.             except Exception as err:
  106.                 self.logger.debug("fail to parse: %s", args)
  107.                 return
  109.             self.logger.debug("get http msg from logstash: %s", new_msg)
  111.             #丢到队列，进行事件提取
  112.             self.put_msg(new_msg)
  113.             self.count += 1
  114.             if self.count % 1000 == 0:
  115.                 print "has put {}".format(self.count)
  116.             return new_msg
  118.         except Exception as ex:
  119.             self.logger.error("fail to parse logstash data: %s", ex)
  121.     ...其他省略...

事件提取

  1. **通用模块，无需修改定制**
  1. #源码：
  2. #   sniffer/nebula_sniffer/nebula_sniffer/main.py
  4. class Main(object):
  5.         def event_processor(self):
  6.             ...其他省略...
  8.             events = []
  9.             if isinstance(msg, HttpMsg):
  10.                 # 对http信息进行处理，返回一个events（事件列表）
  11.                 events = self.parser.get_events_from_http_msg(msg)
  13.             ...其他省略...

输出事件

  1. **通用模块，无需修改定制**