linux 获取公网流量 tcpdump + python + C++

前言

需求为，统计linux上得上下行公网流量，常规得命令如iftop 、sar、ifstat、nload等只能获取流量得大小，不能区分公私网，所以需要通过抓取网络包并排除私网段才能拿到公网流量。下面提供了一些有效得解决思路，提供了部分得代码片段，但不提供整个代码内容。

方案

方案一：在Linux服务器上安装服务，抓取到数据包后发送到中央计算服务器

优点：①由于计算是在中央，并不会占用linux业务服务器得资源性能

缺点：①通过scp或其他传输手段，会占用一定得网络资源，特别当数据量大时

②中央计算服务器得资源易达到瓶颈，需要开多个计算节点

方案二：在Linux服务器上安装服务，抓取到数据包后本地处理，直接将结果发至中央

优点：①中央节点压力减小，数据实时性提升

缺点：①每个linux服务器都需要抓包并且处理，需要占用一定的计算资源，且单位包越大消耗越高

方案三：在同层交换机上放置旁路监控服务器，抓取广播域内所有包，逐个分析每个linux服务器上的流量情况

优点：

①独立部署，不会占用linux服务器资源

②维护相对容易，linux业务服务器与监控节点解耦，只需要维护监控节点而不需要维护linux服务器

缺点：

①需要跟linux服务器处在一个广播域，并且要求性能足够，当linux业务服务器数据过多时，易有瓶颈

②可能存在漏抓包的情况

③需要监听所有得vlan，并分别部署监控节点

方案四：加入流控设备，旁路到核心网关

优点：

①无需自研，有成熟的抓取功能，且有更多丰富的功能

缺点：

①成本较高

②缺乏控制能力，比如流量限速，阻断等

下面主要介绍第一和第二种方法得实现

方案一技术实现

先来看技术拓扑图，流程为：

linux服务器开启tcpdump抓包
linux服务器将pcap包发送到流量分析服务器
流量服务器进行公私网流量拆分，公网白名单过滤
获取步骤三得结果并传输到中央

下面讲解每一步得实现细节

1.linux服务器开启tcpdump抓包

核心命令为"tcpdump", "-i", interface , f"host {target_ip}", '-s 96',"-w", output_file ,'-p'

-i 接网卡名，host接希望抓取得ip（可不填），-s 96 表示只抓取包头部（可能有得包头超过这个长度，但96依然是个比较适中得值，-w 保存为pcap包，-p 关闭混杂模式（避免监听到广播得其他包使流量过多）

#执行tcpdump抓包
def run_tcpdump(target_ip):INTERVAL = 60interface = getInterface() #获取网卡名timestamp = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") #获取开始时间output_file = f"{SAVEPATH}/xxx_{timestamp}.pcap"tcpdump_command = ["tcpdump", "-i", interface , f"host {target_ip}", '-s 96',"-w", output_file ,'-p']try:process = subprocess.Popen(tcpdump_command) #使用subprocess执行shell命令time.sleep(INTERVAL) #定义休眠时间process.terminate()  #结束shell命令process.wait()       #等待子进程退出，确保子进程清理完成except Exception as e:logger.error(f"xxx")

2.linux服务器将pcap包发送到流量分析服务器

这一个步骤没有太多可讲得，使用socket 或requests 或 subprocess(scp)都可以，目的就是将抓取到得包发送到流量分析服务器。

3.流量服务器进行公私网流量拆分，公网白名单过滤

这一步是关键步骤，公私网流量，公网白名单过滤拆分需要一定得计算性能，所以核心代码推荐用C++语言编写，博主测试过同代码情况下，python C++ java得表现能力，C++（或C）性能要远超其他语言，如同体积包，假设占用30%得单核心性能，C++可以做到1%以内。

代码要实现两点：

1.公私网彻底拆分

2.过滤公网白名单内的（不希望统计得）数据

公私网流量拆分参考以下代码片段，首先剔除三类子网地址（0xFF000000 是掩码（255.0.0.0 的二进制形式），只保留 IP 地址的最高 8 位。0x0A000000 是 10.0.0.0 的二进制形式。）

注意：现在得大型网络多采用overlay得方式，有些ip虽然在私网段，但由于走了vxlan隧道占用了公网带宽，其实也是数据公网得。后续得处理需读者自行处理。

bool is_private_ip(const std::string &ip) {struct in_addr addr;inet_pton(AF_INET, ip.c_str(), &addr);uint32_t ip_addr = ntohl(addr.s_addr);// 10.0.0.0/8if ((ip_addr & 0xFF000000) == 0x0A000000) return true;// 172.16.0.0/12if ((ip_addr & 0xFFF00000) == 0xAC100000) return true;// 192.168.0.0/16if ((ip_addr & 0xFFFF0000) == 0xC0A80000) return true;return false;
}

定义4个变量分别存公网下行，公网上行，私网下行，私网上行

    uint64_t public_in= 0;uint64_t public_out = 0;uint64_t private_in = 0;uint64_t private_out = 0;  pcap_t *handle = pcap_open_offline(pcap_file.c_str(), errbuf); #获取pcap包pcap_next_ex(handle, &header, &data)  #调用libpcap 捕获库 const struct ip *ip_header = (struct ip *)(data + sizeof(struct ether_header)); #获取包得headstd::string src_ip = inet_ntoa(ip_header->ip_src); #获取src ipstd::string dst_ip = inet_ntoa(ip_header->ip_dst); #获取dst ip#判断是私网还是公网if (dst_ip == target_ip) {if (is_private_ip(src_ip)) {private_in += pkt_len;} else {public_in += pkt_len;}} else if (src_ip == target_ip) {if (is_private_ip(dst_ip)) {private_out += pkt_len;} else {public_out += pkt_len;}}

接下来是过滤不希望统计的ip名单，在上面代码的基础上再做一层判断即可，

    #判断是私网还是公网if (dst_ip == target_ip) {if (is_private_ip(src_ip)) {private_in += pkt_len;} else {if (filter_write(src_ip)){  #return true则公网ip不在白名单内，需要加和public_in += pkt_len;}}} else if (src_ip == target_ip) {if (is_private_ip(dst_ip)) {private_out += pkt_len;} else {if (filter_write(src_ip)){public_out += pkt_len;}}}

4.获取步骤三得结果上报并持久化

拿到公网流量后，传输给用于持久化的程序进行入库操作（对于此类数据，建议用clickhouse进行建库，对体量小的用mysql也可以接受）.关于传输方式，可以选择使用生产消费者方式，使用消息中间件缓冲数据（如Kafka）。数据量小通过http/https协议传输后插入也可以。

kafka模板

def push_kafka(data, retries=5, backoff_factor=1):logger.info(f"上传 {data}")# 配置 Kafka 生产者kafka_config = {'bootstrap.servers': KAFKA_URL,  # 替换为你的 Kafka 地址'client.id': 'my-producer',}producer = Producer(kafka_config)topic = KAFKA_TOPIC  # 替换为你的 Kafka 主题名# 重试逻辑for attempt in range(retries):try:# 将数据转换为 JSON 格式并发送到 Kafkaproducer.produce(topic=topic, value=json.dumps(data), callback=delivery_report)producer.flush()  # 强制刷新缓冲区logger.info("数据上传成功")return  # 成功后返回except KafkaException as e:logger.error(f"Kafka error occurred: {e}")except Exception as e:logger.error(f"Unexpected error: {e}")# 等待后重试wait_time = backoff_factor * (2 ** attempt)logger.info(f"等待 {wait_time} 秒后重试...")time.sleep(wait_time)logger.error("所有重试均失败")

http/https直传

def post_with_retries_center(data, retries=5, backoff_factor=1):URL_CENTER = "192.168.10.10" #填自己的地址logger.info(f"上传{data}")session = requests.Session()headers = {'Content-Type': 'application/json',}#  定义重试策略retry_strategy = Retry(total=retries,status_forcelist=[404, 500, 502, 503, 504],allowed_methods=["POST"],backoff_factor=backoff_factor)#  创建适配器并将其安装到会话对象中adapter = HTTPAdapter(max_retries=retry_strategy)session.mount("http://", adapter)session.mount("https://", adapter)try:response = requests.post(URL_CENTER, json=data, headers=headers)response.raise_for_status()  # 如果响应状态码不是200，则抛出异常except requests.exceptions.ConnectionError as conn_err:logger.error(f"Connection error occurred: {conn_err}")except requests.exceptions.Timeout as timeout_err:logger.error(f"Timeout error occurred: {timeout_err}")except requests.exceptions.RequestException as req_err:logger.error(f"An error occurred: {req_err}")except Exception as e:logger.error(f"Unexpected error: {e}")return None