求助：如何用Python+osquery实现实时网络信息获取及动态更新

要实现网络状态变化时同步输出更新，你可以通过定时轮询+状态对比的方式，基于现有代码扩展。以下是完整实现：

import osquery
import time
from typing import Set, Tuple

def get_socket_info(instance: osquery.SpawnInstance) -> Set[Tuple]:
    """查询当前活跃的网络套接字，返回可哈希的集合用于对比"""
    query = """
        SELECT 
            (CASE family WHEN 2 THEN 'IP4' WHEN 10 THEN 'IP6' ELSE family END) AS family,
            (CASE protocol WHEN 6 THEN 'TCP' WHEN 17 THEN 'UDP' ELSE protocol END) AS protocol,
            local_address, local_port, remote_address, remote_port
        FROM process_open_sockets 
        WHERE family IN (2, 10) AND protocol IN (6, 17)
    """
    result = instance.client.query(query)
    # 将每条结果转为元组，方便存入集合做对比
    return {
        (
            item["family"],
            item["protocol"],
            item["local_address"],
            item["local_port"],
            item["remote_address"],
            item["remote_port"]
        )
        for item in result.response
    }

if __name__ == "__main__":
    # 初始化osquery实例
    instance = osquery.SpawnInstance()
    instance.open()
    
    # 获取初始套接字状态
    previous_sockets = get_socket_info(instance)
    print("=== 初始网络套接字状态 ===")
    for sock in previous_sockets:
        print(f"Family: {sock[0]}, Protocol: {sock[1]}, Local: {sock[2]}:{sock[3]}, Remote: {sock[4]}:{sock[5]}")
    
    try:
        print("\n=== 开始监控网络变化 ===")
        while True:
            time.sleep(2)  # 每隔2秒查询一次，可根据需求调整
            current_sockets = get_socket_info(instance)
            
            # 找出新增的套接字
            added = current_sockets - previous_sockets
            if added:
                print("\n[+] 新增连接:")
                for sock in added:
                    print(f"Family: {sock[0]}, Protocol: {sock[1]}, Local: {sock[2]}:{sock[3]}, Remote: {sock[4]}:{sock[5]}")
            
            # 找出关闭的套接字
            removed = previous_sockets - current_sockets
            if removed:
                print("\n[-] 断开连接:")
                for sock in removed:
                    print(f"Family: {sock[0]}, Protocol: {sock[1]}, Local: {sock[2]}:{sock[3]}, Remote: {sock[4]}:{sock[5]}")
            
            # 更新状态
            previous_sockets = current_sockets
    except KeyboardInterrupt:
        print("\n监控已停止")
    finally:
        instance.close()

关键说明： #

• get_socket_info函数封装查询逻辑，将返回的字典结果转为元组集合，方便后续对比差异
• 主循环通过time.sleep()控制查询间隔，可根据需求调整时间（比如1秒或5秒）
• 每次查询后对比当前状态与上次状态，输出新增/移除的连接信息
• 捕获KeyboardInterrupt（Ctrl+C）来优雅停止监控，最后关闭osquery实例

进阶优化方向： #

如果需要更高效的实时监控（避免轮询开销），可以使用osquery的事件订阅功能，通过注册process_open_sockets表的变更事件来触发通知。不过这需要基于osquery扩展机制实现，Python端可以通过osquery的Thrift API来订阅事件，复杂度相对较高，适合对性能有要求的场景。

内容的提问来源于stack exchange，提问作者Prince Amle