问题标签 [netflow]

我是 scapy 的新手，正在寻找确认以下命令是挖掘（解析）UDP/NetflowV9 数据包的有效方法。

我问的原因是该方法对我来说似乎有效并且适用于我尝试过的许多数据包。但是，我遇到的至少一个数据包失败了。失败的解码/解析针对具有 4 个 Flowset 的 UDP/Netflow 数据包，如下所示：

• Flowset 1：具有 2 个流记录的数据流集
• 流程集 2：选项模板流程集
• Flowset 3：具有 1 个流记录的数据流集
• Flowset 4：数据模板流程集

方法

上面的命令似乎可以显示有关前 2 个流集的详细信息，但最后一个命令的输出对于流集 3 有错误的数据（与 Wireshark 显示的不匹配）。特别是，首先，对于流集 3 flowsetID 和长度错误。我什至还没有尝试查看 flowset 4，因为 flowset 3 的数据是错误的。

所以，我只是在寻找确认我的方法是正确的。如果我的方法是正确的，那么我将尝试找到一种方法来匿名化 pcap 文件（单帧）并在单独的 stackoverflow 帖子中询问。我不想在流记录中透露 IP 地址。

我正在寻找有关免费 linux 工具的建议，该工具可以收集 netflow v9 流量并将解析的数据存储在时间序列数据库中以供进一步分析。我不需要分析能力，只需要良好的可靠收集和存储性能

我正在尝试使用 logstash 上的弹性搜索过滤器来丰富数据。

我有两个索引，我的目标是从其中一个获取一些数据并将其添加到另一个。

我配置了一个 logstash 过滤器，它在我的 elasticsearch 中搜索，如果匹配，则输出到索引。

但是我的过滤器无法正常工作，因为当我测试过滤器时出现此错误

我认为模板中的变量和弹性搜索之间存在一些问题

我的logstash是7.3.2，我的ES是7.4.2

这是我的设置

我有一个巨大的 NetFlow 数据库，（它包含时间戳、源 IP、目标 IP、协议、源和目标端口号、交换的数据包、字节等）。我想根据当前行和以前的行创建自定义属性。

我想根据当前行的源 ip 和时间戳计算新列。这是我想要在逻辑上做的事情：

• 获取当前行的源 ip。
• 获取当前行的时间戳。
• 根据源 IP 和时间戳，我想获取与源 IP 匹配的整个数据帧的所有先前行，并且通信发生在最后半小时内。这个非常重要。
• 对于符合条件的行（在我的示例中为流）（源 ip 并且发生在过去半小时内），我想计算所有数据包和所有字节的总和和平均值。

数据集中的一行

相关代码片段：

我知道可能有更好、更有效的方法来做到这一点，但遗憾的是我不是最好的程序员。

谢谢。

我正在尝试对提供给我的数据集的 netflow 数据进行一些统计分析，但是我得到了一些不代表正常 UAPRSF 格式的 TCP 标志。

还包括以下十六进制值：

• 0x52
• 0x5a
• 0xc2
• 0xd3
• 0xd6
• 0xd7
• 0xda
• 0xdb
• 0xdf

我知道 TCP 标志最初存储为 HEX，然后翻译成适当的标志，但我不明白附加值来自哪里

我需要在 Postgresql 中存储 netflow 数据。这是有关网络流量的数据。每条记录包含以下内容：

• 连接开始时间
• 连接结束时间
• 传输的数据总量
• 源/目标 IP/ASN
• （还有很多，但这对于这个问题的目的来说已经足够了）。

我的问题是：如何存储这些数据，以便有效地计算过去 X 天/小时的数据传输率？例如，我可能想绘制过去 7 天到 Netflix 的 ASN 的所有流量的图表，每小时分辨率。

连接开始和结束时间之间的差异可能是毫秒，也可能超过一个小时。

我的第一步是将连接存储在带有 GiST 索引的 TSTZRANGE 字段中。然后，查询过去 7 天的每小时流量数据：

1. 使用 CTE 生成一系列每小时时间段
2. 寻找与每个桶重叠的任何 TSTZRANGE
3. 计算重叠的持续时间
4. 以每秒字节数计算记录的数据速率
5. 每秒执行持续时间 * 字节以获取总数据
6. 将其全部分组到存储桶上，对总数据值求和

然而，这听起来像是很多繁重的工作。谁能想到更好的选择？

我在 ClickHouse 表中有一些原始 JSON 数据（实际上是来自 netflow 收集器的 netflow V9），它看起来像这样：

我的任务是通过以下方式将 DataSets 数组转换为另一个 ClickHouse 表：

我需要为我的 logstash 收集器生成 netflow 测试数据。

我尝试了很多软件，但每个人都有很多问题。最后我尝试的是 Debian 下的 flow-tools，但我有以下问题：

• 发送的时间戳设置在 1970 年；
• 我不知道如何设置 sourceip:port/destip:port 来测试地理点等...

我使用这个命令：

这是我在 tcpdump 中收到的：

这是logstash生成的elasticsearch条目：

更新：

这是logstash得到的：

我正在使用带有模块的ELK堆栈。Netflow首先，当我检查CPU使用情况Logstash时使用了大量资源，我决定停止它。这一刻Elasticsearch/Kibana/Logstash停止了。我的意思是，我跑了 command sudo service elasticsearch/kibana/logstash stop。基本上，我认为logstash有问题。当我看到登录时，htop我得到了这样的东西，我不明白为什么。

检查logstash服务状态时，得到类似的东西。

Logstash仍在运行，我正在尝试弄清楚如何阻止它。我想，我一开始就以错误的方式运行它，但为什么不能永远停止它呢？

当我尝试设置 netflow logstash 模块但通过发出此命令 bin/logstash --modules netflow --setup --path.settings /etc/logstash/ 出现以下错误

:::

如何解决这个问题