YuruC3/OPNsense-NetFlow-export
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: YuruC3:main
Branches Tags
YuruC3:main YuruC3:sFlow++
..
compare: YuruC3:sFlow++
Branches Tags
YuruC3:main YuruC3:sFlow++

1 Commits
main ... sFlow++

Author SHA1 Message Date
YuruC3
15428949fe Update README.md 2025-12-20 22:10:15 +00:00
13 changed files with 66 additions and 607 deletions
Expand all files Collapse all files

159
Docker/Code/INFLUXDB.py
View File

@@ -1,159 +0,0 @@
import netflow, socket, json, time, os, influxdb_client, ipaddress
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS, ASYNCHRONOUS, WriteOptions
from datetime import timedelta
from proto import manWhatTheProto
from IP2Loc import ermWhatTheCountry
from whatDomain import ermWhatATheIpFromDomainYaCrazy, ermWhatAAAATheIpFromDomainYaCrazy
from typing import Annotated
# Netentry preconf
# WHAT_THE_NETFLOW_PORT: Final[int] = os.getenv("WHAT_THE_NETFLOW_PORT", "2055")
WHAT_THE_NETFLOW_PORT = 2055
WHAT_THE_NETFLOW_IP = "0.0.0.0"
# INFLUXDB config
token: Final[str] = os.getenv("token", "NotPresent")
# token = "apg1gysUeCcxdcRTMmosJTenbEppmUNi9rXlANDB2oNadBdWAu2GVTDc_q_dyo0iyYsckKaOvPRm6ba2NK0y_A=="
bucket: Final[str] = os.getenv("bucket", "NotPresent")
# bucket = "NETFLOW-7"
org: Final[str] = os.getenv("org", "NotPresent")
# org = "staging"
url: Final[str] = os.getenv("url", "NotPresent")
# url = "http://localhost:8086"
measurement: Final[str] = os.getenv("measurement", "OPNsense-NetFlow-Parser")
# measurement = "testNetFlowPython"
MACHINE_TAG: Final[str] = os.getenv("MACHINE_TAG", socket.gethostname())
# MACHINE_TAG = "YUKIKAZE"
ROUTER_TAG: Final[str] = os.getenv("ROUTER_TAG", socket.gethostname())
# ROUTER_TAG = "HQ"
INFLX_SEPARATE_POINTS: Final[float] = os.getenv("INFLX_SEPARATE_POINTS", 0.1)
# INFLX_SEPARATE_POINTS = 0.1
# Emulate sFlow behaviour
SAMPLING_SIZE: Final(int) = os.getenv("SAMPLING_SIZE", 100)
# Initialize InfluxDB client and influxdb API
inflxdb_client = influxdb_client.InfluxDBClient(url=url, token=token, org=org)
write_api = inflxdb_client.write_api(write_options=SYNCHRONOUS)
# Other preconf
bigDict = {}
inflxdb_Datazz_To_Send = []
# Bind
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.bind((WHAT_THE_NETFLOW_IP, WHAT_THE_NETFLOW_PORT))
print("Ready")
# Flightchecks
flightlist = [token, bucket, org, url]
if "NotPresent" in flightlist:
return 1
while True:
# Get netentry data ig?
payload, client = sock.recvfrom(4096) # experimental, tested with 1464 bytes
p = netflow.parse_packet(payload) # Test result: <ExportPacket v5 with 30 records>
#print(p.entrys) # Test result: 5
#yesyes = p.flows
#print(yesyes.data)
#exit()
for i, entry in enumerate(p.flows, 1):
# prep dict
#tmpEntry = str(entry)
#tmpEntry = tmpEntry[22:-1]
#tmpEntry2 = tmpEntry.replace("'", '"')
#print(tmpEntry2)
#print(entry
#exit()
#dictEntry = json.loads(tmpEntry2)
#bigDict[i] = (dictEntry)
# take data out from netentry
inEntry = entry.data
print(inEntry)
exit()
# Convert IPs and time duration
# IPs
inEntry["IPV4_SRC_ADDR"] = str(ipaddress.IPv4Address(inEntry["IPV4_SRC_ADDR"]))
inEntry["IPV4_DST_ADDR"] = str(ipaddress.IPv4Address(inEntry["IPV4_DST_ADDR"]))
inEntry["NEXT_HOP"] = str(ipaddress.IPv4Address(inEntry["NEXT_HOP"]))
# Convert time from ms to HH:MM:SS
first = int(inEntry["FIRST_SWITCHED"])
last = int(inEntry["LAST_SWITCHED"])
inEntry["FIRST_SWITCHED_HR"] = str(timedelta(milliseconds=first))
inEntry["LAST_SWITCHED_HR"] = str(timedelta(milliseconds=last))
# Prep InfluxDB data
inflxdb_Data_To_Send = (
influxdb_client.Point(f"{measurement}-script")
.tag("MACHINE", MACHINE_TAG)
.tag("ROUTER", ROUTER_TAG)
.field("dstAddr", inEntry["IPV4_DST_ADDR"])
.field("srcAddr", inEntry["IPV4_SRC_ADDR"])
.field("nextHop", inEntry["NEXT_HOP"])
.field("inptInt", inEntry["INPUT"])
.field("outptInt", inEntry["OUTPUT"])
.field("inPackt", inEntry["IN_PACKETS"])
.field("outPakt", inEntry["IN_OCTETS"])
.field("frstSwtchd", inEntry["FIRST_SWITCHED"])
.field("lstSwtchd", inEntry["LAST_SWITCHED"])
.field("srcPort", inEntry["SRC_PORT"])
.field("dstPort", inEntry["DST_PORT"])
.field("tcpFlags", inEntry["TCP_FLAGS"])
.tag("proto", manWhatTheProto(int(inEntry["PROTO"])))
.field("tos", inEntry["TOS"])
.field("srcAS", inEntry["SRC_AS"])
.field("dstAS", inEntry["DST_AS"])
.field("srcMask", inEntry["SRC_MASK"])
.field("dstMask", inEntry["DST_MASK"])
.field("dstCntr", ermWhatTheCountry(str(inEntry["IPV4_DST_ADDR"])))
.field("srcCntr", ermWhatTheCountry(str(inEntry["IPV4_SRC_ADDR"])))
)
inflxdb_Datazz_To_Send.append(inflxdb_Data_To_Send)
#i+=1
#type(tmpEntry)
#print(dictEntry)
#print(tmpEntry.lstrip(20))
print("----------------")
bigDict[i] = (inEntry)
# end while True
print()
print(bigDict)
exit()
# Send data to InfluxDB
write_api.write(bucket=bucket, org="staging", record=inflxdb_Data_To_Send)
time.sleep(INFLX_SEPARATE_POINTS) # separate points
print(f"{len(bigDict)} <--- This many entrys")
# Clean up before another loop
bigDict.clear()
inflxdb_Datazz_To_Send.clear()
#print(bigDict)

28
Docker/Code/IP2Loc.py
View File

@@ -1,28 +0,0 @@
import IP2Location
from typing import Optional, Annotated
# Load database once
ip2loc_db: IP2Location = IP2Location.IP2Location("IP2LOCATION-LITE-DB9.BIN", "SHARED_MEMORY")
def ermWhatTheCountry(inpIpAddress: Annotated[str, "Some IP address that ya want to get country for"]) -> str:
try:
skibidi = ip2loc_db.get_all(inpIpAddress)
#return rec.country_long # Full country name, e.g. "Sweden"
return skibidi.country_short
except Exception as errrrrr:
return f"Error: {errrrrr}"
def ermWhatTheISP(inpIpAddress: Annotated[str, "Some IP address that ya want to get ISP for"]) -> str:
try:
skibidi = ip2loc_db.get_all(inpIpAddress)
#return rec.country_long # Full country name, e.g. "Sweden"
return skibidi.isp
except Exception as errrrrr:
return f"Error: {errrrrr}"
#print(ermWhatTheCountry("65.109.142.32"))

178
Docker/Code/proto.py
View File

@@ -1,178 +0,0 @@
from typing import Optional, Annotated
# Source
# https://en.wikipedia.org/wiki/List_of_IP_protocol_numbers
PROTO_MAP = {
0: "HOPOPT",
1: "ICMP",
2: "IGMP",
3: "GGP",
4: "IPv4",
5: "ST",
6: "TCP",
7: "CBT",
8: "EGP",
9: "IGP",
10: "BBN-RCC-MON",
11: "NVP-II",
12: "PUP",
13: "ARGUS",
14: "EMCON",
15: "XNET",
16: "CHAOS",
17: "UDP",
18: "MUX",
19: "DCN-MEAS",
20: "HMP",
21: "PRM",
22: "XNS-IDP",
23: "TRUNK-1",
24: "TRUNK-2",
25: "LEAF-1",
26: "LEAF-2",
27: "RDP",
28: "IRTP",
29: "ISO-TP4",
30: "NETBLT",
31: "MFE-NSP",
32: "MERIT-INP",
33: "DCCP",
34: "3PC",
35: "IDPR",
36: "XTP",
37: "DDP",
38: "IDPR-CMTP",
39: "TP++",
40: "IL",
41: "IPv6",
42: "SDRP",
43: "IPv6-Route",
44: "IPv6-Frag",
45: "IDRP",
46: "RSVP",
47: "GRE",
48: "DSR",
49: "BNA",
50: "ESP",
51: "AH",
52: "I-NLSP",
53: "SWIPE",
54: "NARP",
55: "MOBILE",
56: "TLSP",
57: "SKIP",
58: "IPv6-ICMP",
59: "IPv6-NoNxt",
60: "IPv6-Opts",
61: "ANY_HOST_INTERNAL",
62: "CFTP",
63: "ANY_LOCAL_NETWORK",
64: "SAT-EXPAK",
65: "KRYPTOLAN",
66: "RVD",
67: "IPPC",
68: "ANY_DISTRIBUTED_FS",
69: "SAT-MON",
70: "VISA",
71: "IPCV",
72: "CPNX",
73: "CPHB",
74: "WSN",
75: "PVP",
76: "BR-SAT-MON",
77: "SUN-ND",
78: "WB-MON",
79: "WB-EXPAK",
80: "ISO-IP",
81: "VMTP",
82: "SECURE-VMTP",
83: "VINES",
84: "TTP",
85: "NSFNET-IGP",
86: "DGP",
87: "TCF",
88: "EIGRP",
89: "OSPF",
90: "Sprite-RPC",
91: "LARP",
92: "MTP",
93: "AX.25",
94: "IPIP",
95: "MICP",
96: "SCC-SP",
97: "ETHERIP",
98: "ENCAP",
99: "ANY_PRIVATE_ENCRYPTION",
100: "GMTP",
101: "IFMP",
102: "PNNI",
103: "PIM",
104: "ARIS",
105: "SCPS",
106: "QNX",
107: "A/N",
108: "IPComp",
109: "SNP",
110: "Compaq-Peer",
111: "IPX-in-IP",
112: "VRRP",
113: "PGM",
114: "ANY_0_HOP",
115: "L2TP",
116: "DDX",
117: "IATP",
118: "STP",
119: "SRP",
120: "UTI",
121: "SMP",
122: "SM",
123: "PTP",
124: "ISIS over IPv4",
125: "FIRE",
126: "CRTP",
127: "CRUDP",
128: "SSCOPMCE",
129: "IPLT",
130: "SPS",
131: "PIPE",
132: "SCTP",
133: "FC",
134: "RSVP-E2E-IGNORE",
135: "Mobility Header",
136: "UDPLite",
137: "MPLS-in-IP",
138: "manet",
139: "HIP",
140: "Shim6",
141: "WESP",
142: "ROHC",
143: "Ethernet",
144: "AGGFRAG",
145: "NSH"
}
def manWhatTheProto(inpProtoNumbrMaybe: Annotated[int, "Protocol number goes here"]) -> int:
if inpProtoNumbrMaybe <= 145:
return PROTO_MAP.get(inpProtoNumbrMaybe)
elif inpProtoNumbrMaybe >= 146 and inpProtoNumbrMaybe <= 252:
return "Unassigned"
elif inpProtoNumbrMaybe >= 253 and inpProtoNumbrMaybe <= 254:
# Use for experimentation and testing
return "RFC3692"
elif inpProtoNumbrMaybe == 255:
return "Reserved"
elif inpProtoNumbrMaybe not in PROTO_MAP:
return inpProtoNumbrMaybe
else:
return -1
#outPotentialProtoNameIfItExistsInInternalList = PROTO_MAP.get(inpProtoNumbrMaybe)
#print(manWhatTheProto(253))
#print( PROTO_MAP.get(2))

6
Docker/Code/requirements.txt
View File

@@ -1,6 +0,0 @@
netflow==0.12.2
influxdb-client==1.48.0
ipaddress==1.0.23
typing==3.7.4.3
IP2Location==8.10.5
nslookup==1.8.1

130
Docker/Code/whatDomain.py
View File

@@ -1,130 +0,0 @@
#from nslookup import Nslookup
from typing import Optional, Annotated
import dns, dns.resolver
from typing import Final
# https://www.codeunderscored.com/nslookup-python/
def ermWhatATheIpFromDomainYaCrazy(inpDomainNameOrSomething: Annotated[str, "Domain name to lookup IP for"]) -> dict:
#dns_query = Nslookup()
"""
Tells you what IPv4 address/es a domain point to.
Returns:
dict: A dictionary with IP addresses associated with that domain.
"""
# i = 0
outDict: dict = {}
#result = dns_query.dns_lookup("example.com")
#result = Nslookup.dns_lookup(inpDomainNameOrSomething)
try:
result = dns.resolver.resolve(inpDomainNameOrSomething, 'A')
except dns.resolver.NoAnswer:
print("\nDNS ERROR")
print("No answer from dns server.\n")
return 1
except dns.resolver.NoNameservers:
print("\nDNS ERROR")
print("All nameservers failed to answer the query.\n Fix your DNS servers.\n")
return 1
except dns.resolver.NXDOMAIN:
print("\nDNS ERROR")
print("The DNS query name does not exist.\n")
return 1
except dns.resolver.LifetimeTimeout:
print("\nDNS ERROR")
print("The DNS querry got timed out.\nVerify that your FW or PiHole isn't blocking requests for that domain.\n")
return 1
for i, something in enumerate(result):
outDict[i] = something.to_text()
# i += 1
return outDict
def ermWhatAAAATheIpFromDomainYaCrazy(inpDomainNameOrSomething: Annotated[str, "Domain name to lookup IP for"]) -> dict:
#dns_query = Nslookup()
"""
Tells you what IPv6 address/es a domain point to.
Returns:
dict: A dictionary with IP addresses associated with that domain.
"""
# i = 0
outDict: dict = {}
#result = dns_query.dns_lookup("example.com")
#result = Nslookup.dns_lookup(inpDomainNameOrSomething)
try:
result = dns.resolver.resolve(inpDomainNameOrSomething, 'AAAA')
except dns.resolver.NoAnswer:
print("\nDNS ERROR")
print("No answer from dns server.\n")
return 1
except dns.resolver.NoNameservers:
print("\nDNS ERROR")
print("All nameservers failed to answer the query.\n Fix your DNS servers.\n")
return 1
except dns.resolver.NXDOMAIN:
print("\nDNS ERROR")
print("The DNS query name does not exist.\n")
return 1
except dns.resolver.LifetimeTimeout:
print("\nDNS ERROR")
print("The DNS querry got timed out.\nVerify that your FW or PiHole isn't blocking requests for that domain.\n")
return 1
for i, something in enumerate(result):
outDict[i] = something.to_text()
# i += 1
return outDict
def ermWhatPTRTheIpFromDomainYaCrazy(inpIpAddressOrSomething: Annotated[str, "IP address to lookup domain for"]) -> dict:
#dns_query = Nslookup()
"""
Tells you what IPv6 address/es a domain point to.
Returns:
dict: A dictionary with IP addresses associated with that domain.
"""
WHATTOCHECK: Final[str] = inpIpAddressOrSomething + ".in-addr.arpa"
# i = 0
outDict: dict = {}
#result = dns_query.dns_lookup("example.com")
#result = Nslookup.dns_lookup(inpDomainNameOrSomething)
try:
result = dns.resolver.resolve(WHATTOCHECK, 'PTR')
except dns.resolver.NoAnswer:
print("\nDNS ERROR")
print("No answer from dns server.\n")
return 1
except dns.resolver.NoNameservers:
print("\nDNS ERROR")
print("All nameservers failed to answer the query.\n Fix your DNS servers.\n")
return 1
except dns.resolver.NXDOMAIN:
print("\nDNS ERROR")
print("The DNS query name does not exist.\n")
return 1
except dns.resolver.LifetimeTimeout:
print("\nDNS ERROR")
print("The DNS querry got timed out.\nVerify that your FW or PiHole isn't blocking requests for that domain.\n")
return 1
for i, something in enumerate(result):
outDict[i] = something.to_text()
# i += 1
return outDict
#print(ermWhatATheIpFromDomainYaCrazy("fubukus.net"))
#print(ermWhatAAAATheIpFromDomainYaCrazy("fubukus.net"))
#print(ermWhatPTRTheIpFromDomainYaCrazy("192.168.1.226"))

29
Docker/Dockerfile
View File

@@ -1,29 +0,0 @@
FROM alpine:latest
RUN apk update && \
apk add python3 py3-pip su-exec curl wget unzip && \
mkdir -p /netflowParser
WORKDIR /netflowParser
# Copy over IP2Loc, INFLUXDB, proto and wahtDomain
COPY ./Code/* /netflowParser/
COPY ./entrypoint.sh /entrypoint.sh
COPY ./updateIP2Lbin.sh /netflowParser/
RUN chmod +x /entrypoint.sh
RUN python3 -m venv venv && \
/netflowParser/venv/bin/python3 -m pip install --upgrade pip && \
/netflowParser/venv/bin/pip3 install -r requirements.txt
# venv/bin/pip3 install -r requirements.txt
# Set user. No need for root past this point
# USER "${USER}"
ENTRYPOINT ["/entrypoint.sh"]
EXPOSE 2055
CMD ["/netflowParser/venv/bin/python3", "/netflowParser/INFLUXDB.py"]

38
Docker/entrypoint.sh
View File

@@ -1,38 +0,0 @@
#!/bin/sh
set -e
# Default to 1600 if not provided
PUID="${PUID:-1600}"
PGID="${PGID:-1600}"
USER="${USER:-apiusr}"
GROUP="${USER:-apiusr}"
CHOWNPATH="/netflowParser"
# Create group if missing
if ! getent group "$GROUP" >/dev/null 2>&1; then
addgroup -g "$PGID" "$GROUP"
fi
# Create user if missing
if ! id -u "$USER" >/dev/null 2>&1; then
adduser -D -u "$PUID" -G "$GROUP" "$USER"
fi
# Download IP2Location database
if [[ -n "$IP2LOC_TOKEN" ]]; then
wget -O "$CHOWNPATH"/IP2LOCATION-LITE-DB9.BIN "https://www.ip2location.com/download/?token="$IP2LOC_TOKEN"&file=DB9LITEBIN"
fi
# Fix permissions (only do this on /app/API)
chown -R "$PUID:$PGID" "$CHOWNPATH"
# Create an updater process
(
while true; do
sleep 12h
su-exec "$PUID:$PGID" "/bin/sh" "$CHOWNPATH"/updateIP2Lbin.sh || true
done
) &
# Drop privileges & run command
exec su-exec "$PUID:$PGID" "$@"

29
Docker/updateIP2Lbin.sh
View File

@@ -1,29 +0,0 @@
#!/bin/bash
# -----------------------------
# IP2Location DB1 Updater
# -----------------------------
# Change this URL to your personal download link if needed
DOWNLOAD_URL="https://www.ip2location.com/download/?token="$IP2LOC_TOKEN"&file=DB9LITEBIN"
# Define filenames
ZIP_FILE="IP2LOCATION-LITE-DB9.BIN.ZIP"
BIN_FILE="IP2LOCATION-LITE-DB9.BIN"
echo "Downloading latest IP2Location DB1..."
wget -O "$CHOWNPATH"/"$BIN_FILE"
echo "Unzipping BIN file..."
unzip -o "$ZIP_FILE"
if [ ! -f "$BIN_FILE" ]; then
echo "Unzip failed or $BIN_FILE not found."
exit 1
fi
echo "Cleaning up ZIP..."
rm "$ZIP_FILE"
echo "Update complete. BIN file ready: $BIN_FILE"

13
NetFlowInflux.conf Normal file
View File

@@ -0,0 +1,13 @@
#Put it in /etc/rsyslog.d/
if $programname == 'NetFlowInflux' then /var/log/NetFlowInflux.log
& stop
#then run
#
#touch /var/log/yapyap
#chown syslog /var/log/yapyap
#ls -l /var/log/yapyap.log
# And then
# systemctl restart rsyslog

11
README.md
View File

@@ -1,15 +1,9 @@
# NetFlowV5 collector and InfluxDB and Grafana # NetFlowV5 collector and InfluxDB and Grafana
Here I'll try integrating both NetFlow and sFlow into my python code so that either can be choosen.
It works. It works.
## IP2Location-DB
To get IP2Locatio-Lite-DB go to [lite.ip2location](https://lite.ip2location.com/) and create a account.
Then you need to get your **Token**. You can find it [here](https://lite.ip2location.com/database-download) and under **Download Token**
Then pass that token as ```IP2LOC_TOKEN``` environment variable to the container
## Python script ## Python script
Install required modules with Install required modules with
``` ```
@@ -26,4 +20,3 @@ Second is when there are a ton of flows that need to be collected. More flows ak
## sysctl.d ## sysctl.d
Place it in /etc/sysctl.d/ and apply with ```sysctl -p``` Place it in /etc/sysctl.d/ and apply with ```sysctl -p```

48
docker-compose.yml Normal file
View File

@@ -0,0 +1,48 @@
version: '3.8'
services:
redis:
image: redis:latest
container_name: redis-test
ports:
- "6379:6379"
volumes:
- ./redis_data:/data
restart: unless-stopped
networks:
- db
influxdb:
image: influxdb:2.7
container_name: influxdb-test
ports:
- "8086:8086"
environment:
- DOCKER_INFLUXDB_INIT_MODE=setup
- DOCKER_INFLUXDB_INIT_USERNAME=yuru
- DOCKER_INFLUXDB_INIT_PASSWORD=2214112137
- DOCKER_INFLUXDB_INIT_ORG=staging
- DOCKER_INFLUXDB_INIT_BUCKET=STAGING-NETFLOW
- DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=2214112137
volumes:
- ./influxdb_data:/var/lib/influxdb2
restart: unless-stopped
networks:
- db
grafana:
image: grafana/grafana:latest
container_name: grafana
restart: unless-stopped
ports:
- '3000:3000'
volumes:
- ./grafana-storage:/var/lib/grafana
depends_on:
- influxdb
networks:
- db
networks:
db:
driver: bridge

2
requirements.txt
View File

@@ -1,4 +1,6 @@
netflow==0.12.2 netflow==0.12.2
#socket==
#json==
influxdb-client==1.48.0 influxdb-client==1.48.0
ipaddress==1.0.23 ipaddress==1.0.23
typing==3.7.4.3 typing==3.7.4.3

2
updateIP2Lbin.sh
View File

@@ -5,7 +5,7 @@
# ----------------------------- # -----------------------------
# Change this URL to your personal download link if needed # Change this URL to your personal download link if needed
DOWNLOAD_URL="https://www.ip2location.com/download/?token=yes&file=DB9LITEBIN" DOWNLOAD_URL="https://www.ip2location.com/download/?token=MkyoKFL854ID0FOWoeTWVRsw0SVcbA7ey6tvuzHchsIQ6AMGy7YXIDfwrEEA4Ozn&file=DB9LITEBIN"
# Define filenames # Define filenames
ZIP_FILE="IP2LOCATION-LITE-DB9.BIN.ZIP" ZIP_FILE="IP2LOCATION-LITE-DB9.BIN.ZIP"