From 35ebcf9a39c9011e4717b7b05b2b2a5b6fa781f3 Mon Sep 17 00:00:00 2001 From: YuruC3 Date: Sun, 11 Jan 2026 19:24:33 +0100 Subject: [PATCH] Added new Docker implementation --- Docker/Code/INFLUXDB.py | 159 +++++++++++++++++++++++++++++++ Docker/Code/IP2Loc.py | 28 ++++++ Docker/Code/proto.py | 178 +++++++++++++++++++++++++++++++++++ Docker/Code/requirements.txt | 6 ++ Docker/Code/whatDomain.py | 130 +++++++++++++++++++++++++ Docker/Dockerfile/Dockerfile | 28 ++++++ 6 files changed, 529 insertions(+) create mode 100644 Docker/Code/INFLUXDB.py create mode 100644 Docker/Code/IP2Loc.py create mode 100644 Docker/Code/proto.py create mode 100644 Docker/Code/requirements.txt create mode 100644 Docker/Code/whatDomain.py create mode 100644 Docker/Dockerfile/Dockerfile diff --git a/Docker/Code/INFLUXDB.py b/Docker/Code/INFLUXDB.py new file mode 100644 index 0000000..ed9e86a --- /dev/null +++ b/Docker/Code/INFLUXDB.py @@ -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: + #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) diff --git a/Docker/Code/IP2Loc.py b/Docker/Code/IP2Loc.py new file mode 100644 index 0000000..8f82ccf --- /dev/null +++ b/Docker/Code/IP2Loc.py @@ -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")) + diff --git a/Docker/Code/proto.py b/Docker/Code/proto.py new file mode 100644 index 0000000..05e20a6 --- /dev/null +++ b/Docker/Code/proto.py @@ -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)) diff --git a/Docker/Code/requirements.txt b/Docker/Code/requirements.txt new file mode 100644 index 0000000..3194774 --- /dev/null +++ b/Docker/Code/requirements.txt @@ -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 diff --git a/Docker/Code/whatDomain.py b/Docker/Code/whatDomain.py new file mode 100644 index 0000000..68efcfe --- /dev/null +++ b/Docker/Code/whatDomain.py @@ -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")) diff --git a/Docker/Dockerfile/Dockerfile b/Docker/Dockerfile/Dockerfile new file mode 100644 index 0000000..2e65bbf --- /dev/null +++ b/Docker/Dockerfile/Dockerfile @@ -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"]