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

Added new Docker implementation

Browse Source
This commit is contained in:
YuruC3
2026-01-11 19:24:33 +01:00
parent 532b078411
commit 35ebcf9a39
6 changed files with 529 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

159
Docker/Code/INFLUXDB.py Normal file
View File

@@ -0,0 +1,159 @@
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 Normal file
View File

@@ -0,0 +1,28 @@
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 Normal file
View File

@@ -0,0 +1,178 @@
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 Normal file
View File

@@ -0,0 +1,6 @@
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 Normal file
View File

@@ -0,0 +1,130 @@
#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"))

28
Docker/Dockerfile/Dockerfile Normal file
View File

@@ -0,0 +1,28 @@
FROM alpine:latest
RUN apk update && \
apk add python3 py3-pip su-exec curl wget && \
mkdir -p /netflowParser
WORKDIR /netflowParser
# Copy over IP2Loc, INFLUXDB, proto and wahtDomain
COPY ../Code/* /netflowParser/
COPY ./entrypoint.sh /entrypoint.sh
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"]