捕獲 DNS 服務器請求，並對來源加以分析

有一天有留意到 DNS 服務器負載變高。查看了一下日誌，發現有大量的請求湧入。本著並沒有對我的設施造成嚴重影響的事實，並暫時沒有作出限制性操作，畢竟我的 DNS 服務器除了用來處理自己網絡內的私有域名解析，也順帶是讓閒置的資源作一點公益用途。

根據手冊對 DNS 服務器作出性能調整後，響應速度快了一半。而後請求數量下降了不少後，我便開始思考對請求來源與內容作一些監控，以便後來對可能出現的濫用情況有知情手段。

一開始我打算讓 netfilter 代勞，直接從內核打日誌。可搞了一天防火牆日誌，發現 ulogd2 好像不是很適合實時分析的場景，便作罷轉而使用傳統的抓包分析手段了。

tcpdump & tshark

tcpdump 是底層的數據包捕獲工具，直接從網絡接口讀取原始數據包，輸出量大且格式基礎。tshark 是 Wireshark 的命令行版本，提供高層協議解析和格式化輸出。

使用場景：

直接用 tshark：需要協議級別解析（如提取 DNS 查詢名稱、HTTP 請求頭），或輸出需要結構化格式（JSON、CSV）時。
分開用 tcpdump + 後期處理：捕獲量大需要高性能、複雜過濾邏輯、或使用 Lua 腳本自定義數據處理時。

我是分開使用，先用 tcpdump 在生產服務器上獲得數據，再將採樣所得拿回本地分析。從 53 端口截獲 udp 數據：

1	tcpdump -i ens192 'udp and port 53' -w "dns-capture_$(date +%Y%m%d-%H%M).pcap" -v

ens192 是我的外網網卡。為了方便區分，我在文件名上增加了日期。且通過 -v 輸出捕獲狀態（多少個包），讓等待過程沒那麼無聊。如果流量不大，可以用 -vv 來看粗略的所捕獲的包的信息。

用 tshark 處理捕獲到的數據：

1	tshark -r dns-capture_20260426-1440.pcap -T ek

-T ek 將輸出格式轉換為 ElasticSearch 的批插入 JSON 格式。我使用這個格式只是因為 json 方便用工具處理。它會在每一條紀錄前先輸出對應該插入的索引，而後才會輸出 JSON 文檔內容。

後面需要使用 jq 命令配合來過濾於重新格式化。

使用 jq 來預加工數據

jq 是個很方便的 json 處理工具。可以直接通過其類似管道的處理語句來將每行的 json 對象加工成期望的樣子：

tshark -r dns-capture_20260426-1440.pcap -T ek \
| jq '.layers | select ( . != null ) | select ( .dns != null ) | {
ip : .ip.ip_ip_addr, 
dns: .dns
}' \
> dns_requests.jsonl

這裡的 jq 語句是：

從每行中取字段 layers
僅取非空的結果，也就是說空結果在這裡就被丟掉並繼續下一行 JSON
僅取 dns 字段非空的結果，效果同上
重新構建一個 JSON 對象，將字段 ip.ip_ip_addr 賦值到新對象的 ip、將字段 dns 賦值到新對象的 dns

這就能僅取出需要的內容了。如果不確定有什麼內容，可以直接 tshark -r file.pcap -t ek | jq | less 先預覽一下。

用 lua 來查詢補充、歸組與統計

數據處理的語言我用了 lua，簡簡單單。

其中，為了方便對數據作處理，我在 lua 腳本中使用了 rxi/json.lua 來處理 json，以及 darunimator/mmdblua 來查詢 MaxMind GeoLite 數據庫。

MaxMind 的數據庫可以去其網站註冊下載，或者不想註冊的話，也應該會有一些渠道直接獲取，不過可能就沒有官網的那麼新。mmdblua 使用的是二進制版的數據庫。我下載的是 GeoLite2-City，包含了一部分城市信息。

這是一個從 stdin 讀取每一行 json 字符串並查詢字符串 IP 信息的示例。

#!/usr/bin/env lua

local mmdb = require "mmdb"
local json = require "json"

local geodb = assert(mmdb.read("GeoLite2-City.mmdb"))

for line in io.stdin:lines() do
    if not line or line == '' then goto continue; end

    local ip = json.decode(line)
    local result = geodb:search_ipv4(ip)
    if not result then goto continue; end

    io.stdout:write(json.encode(result),'\n')
    io.stdout:flush()
    ::continue::    
end

mmdblua 便是簡單地查出一個 ip 所對應的 table。這段代碼相當於一個 mapper，將 IP 轉換為 JSON 對象。

這是對這些數據作一個統計輸出的腳本，僅是作了區域，沒有到達城市，因為 mmdblua 並沒有實現到這一步。城市級別的歸總，等以後有空了再細化。

#!/usr/bin/env lua

local mmdb = require "mmdb"
local json = require "json"

local geodb = assert(mmdb.read("GeoLite2-City.mmdb"))

local summary = {}
-- 填寫本機公網 IP，用來忽略主機向外方向的包
local host_ip = '127.0.0.1'

for line in io.stdin:lines() do
    if not line or line == '' then goto continue; end

    local entry = json.decode(line)
    local ip, dns = entry.ip, entry.dns
    if not ip or not dns then goto continue; end

    local ip_src = table.unpack(ip)
    if ip_src == host_ip then goto continue; end

    local result = geodb:search_ipv4(ip_src)
    if not result then goto continue; end
    -- Get location entry
    local pos, cty = result.location, result.country
    local ctnt = result.continent
    if not ctnt then goto continue; end
    local key = tostring(ctnt.geoname_id) .. '/' .. tostring(pos.latitude) .. '/' .. tostring(pos.longitude)
    local entry = summary[key]
    if not entry then
        entry = { location=pos; country=cty.names["zh-CN"]; queries={}; remote_addrs={} }
        summary[key] = entry
    end
    
    -- Get query entry
    local queries = entry.queries
    local _, query = table.unpack(dns.text or {})
    if not query then goto continue; end
    queries[query] = (queries[query] or 0) + 1
    
    -- Get IP entry
    local remote_addrs = entry.remote_addrs
    remote_addrs[ip_src] = (remote_addrs[ip_src] or 0) + 1
    ::continue::    
end

io.stdout:write(json.encode(summary),'\n')

最後整理出來一個 map 大概結構如下：

{
  // 以每一個位置作為 key
  'geoname_id/latitude/longitude' : {
    'location': { /* location dict */ },
    'country' : '地區名稱',
    'queries' : { '查詢請求': 1 /* 次數 */ },
    'remote_addrs' : { '來源 IP': 1 /* 次數 */}
  }
}

我接下來便是將通過上面腳本處理的數據，用 jq 再處理一次變成 GeoJSON 後可視化：

# 先處理原數據，得出聚合結果
cat dns_requests.jsonl \
| lua inspect-query-locations.lua | tee inspected-query.json | jq

# 然後再將聚合結果編織成一個單一的 GeoJSON
cat inspected-query.json \
| jq 'keys[] as $k | .[$k] | {type: "Feature", geometry: {
type: "Point",
coordinates: [.location.longitude, .location.latitude]
},
properties: {
name: .country
, remotes: .remote_addrs
, requests: .queries
}
}' -cM \
| jq -s '{ type: "FeatureCollection", features: . }' -c \
| tee inspected-GeoJSON.geojson

以上使用了兩個 jq 命令來拼接最後的 GeoJSON。-cM 是 -c -M 的縮寫， -c 讓 jq 不要將 JSON 格式化成好看的格式，否則下一步會因為這些換行符而解析錯誤。-M 可有可無，就是關掉終端顏色顯示。 -s 意思是將輸出的每一行都存到一個數組裡，再進行後續的處理。其實最後一句 jq 處理可以看成是一個 JSON 模板。

可視化請求紀錄

對數據進行捕獲、格式化於補充之後，接下來便是讓數據直觀明瞭。

GeoJSON

Wiki 條目

將請求信息通過對 mmdb 的查詢整理好後，轉換成 GeoJSON 格式可以很方便地在線顯示。

GeoJSON 主要格式，這裡以比較獨特的排版來直觀地預覽其結構：

{
  "type": "FeatureCollection",
  "features": [
    // 一個單獨的點
    { "type": "Feature", 
      "geometry": { 
        "type": "Point", 
        "coordinates": [102.0, 0.5] 
      },
      // 放置附加在點上的其他數據的字典
      "properties": { "prop0": "value0" }
    },

    // 多個點連成的線段
    { "type": "Feature",
      "geometry": { 
        "type": "LineString", 
        "coordinates": [ 
          [102.0, 0.0],
          [103.0, 1.0],
          [104.0, 0.0],
          [105.0, 1.0]
        ] 
      },
      // 屬性字典不一定是字符串作為內容，可以是數字
      "properties": { 
        "prop0": "value0",
        "prop1": 0.0
      }
    },

    // 多條線段與點連成的多邊形
    { "type": "Feature",
      "geometry": { 
        "type": "Polygon", 
        "coordinates": [
          [ 
            [100.0, 0.0], 
            [101.0, 0.0], 
            [101.0, 1.0], 
            [100.0, 1.0], 
            [100.0, 0.0] 
          ]
        ]
      },
      // 也不一定只是字面量，可以是對象甚至數組
      "properties": {
        "prop0": "value0",
        "prop1": { "this": "that" }
      }
    }
  ]
}