feat: Add machine learning components (#11)

* feat: cat cat

* fix: resolve frame pool exhaustion and packet forwarding issues

* refactor: implement non-blocking ML detection with dedicated threads

* add: Add GeoIP

* fix: Fix IP address sequence reversal issue

* fix: Fix access control IP address sequence reversal issue

* add: Add onnx models

* wip

* wip: refactor

* wip: Re-form ml

---------

Co-authored-by: DaLaw2 <t20040421@gmail.com>
This commit is contained in:
ParrotXray 2026-01-19 18:07:49 +08:00 committed by GitHub
parent 1c778cc5bb
commit ef020a1c9c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
39 changed files with 21962 additions and 220 deletions

BIN
.github/images/map.png vendored Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 187 KiB

Binary file not shown.

Before

Width:  |  Height:  |  Size: 84 KiB

After

Width:  |  Height:  |  Size: 122 KiB

358
Cargo.lock generated
View File

@ -11,7 +11,7 @@ dependencies = [
"actix-macros",
"actix-rt",
"actix_derive",
"bitflags 2.9.4",
"bitflags",
"bytes",
"crossbeam-channel",
"futures-core",
@ -33,7 +33,7 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f7b0a21988c1bf877cf4759ef5ddaac04c1c9fe808c9142ecb78ba97d97a28a"
dependencies = [
"bitflags 2.9.4",
"bitflags",
"bytes",
"futures-core",
"futures-sink",
@ -70,17 +70,17 @@ dependencies = [
"actix-service",
"actix-utils",
"base64",
"bitflags 2.9.4",
"bitflags",
"brotli",
"bytes",
"bytestring",
"derive_more",
"encoding_rs",
"flate2",
"foldhash",
"foldhash 0.1.4",
"futures-core",
"h2",
"http",
"http 0.2.12",
"httparse",
"httpdate",
"itoa",
@ -116,7 +116,7 @@ checksum = "13d324164c51f63867b57e73ba5936ea151b8a41a1d23d1031eeb9f70d0236f8"
dependencies = [
"bytestring",
"cfg-if",
"http",
"http 0.2.12",
"regex",
"regex-lite",
"serde",
@ -192,7 +192,7 @@ dependencies = [
"cookie",
"derive_more",
"encoding_rs",
"foldhash",
"foldhash 0.1.4",
"futures-core",
"futures-util",
"impl-more",
@ -316,7 +316,7 @@ checksum = "d18bc4e506fbb85ab7392ed993a7db4d1a452c71b75a246af4a80ab8c9d2dd50"
dependencies = [
"assert_matches",
"aya-obj",
"bitflags 2.9.4",
"bitflags",
"bytes",
"libc",
"log",
@ -429,7 +429,7 @@ checksum = "c51b96c5a8ed8705b40d655273bc4212cbbf38d4e3be2788f36306f154523ec7"
dependencies = [
"bytes",
"core-error",
"hashbrown",
"hashbrown 0.15.2",
"log",
"object",
"thiserror 1.0.69",
@ -457,10 +457,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
[[package]]
name = "bitflags"
version = "1.3.2"
name = "bindgen"
version = "0.72.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
checksum = "993776b509cfb49c750f11b8f07a46fa23e0a1386ffc01fb1e7d343efc387895"
dependencies = [
"bitflags",
"cexpr",
"clang-sys",
"itertools",
"log",
"prettyplease",
"proc-macro2",
"quote",
"regex",
"rustc-hash",
"shlex",
"syn",
]
[[package]]
name = "bitflags"
@ -570,21 +584,48 @@ dependencies = [
[[package]]
name = "cc"
version = "1.2.14"
version = "1.2.47"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c3d1b2e905a3a7b00a6141adb0e4c0bb941d11caf55349d863942a1cc44e3c9"
checksum = "cd405d82c84ff7f35739f175f67d8b9fb7687a0e84ccdc78bd3568839827cf07"
dependencies = [
"find-msvc-tools",
"jobserver",
"libc",
"shlex",
]
[[package]]
name = "cexpr"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766"
dependencies = [
"nom",
]
[[package]]
name = "cfg-if"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "cfg_aliases"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
[[package]]
name = "clang-sys"
version = "1.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4"
dependencies = [
"glob",
"libc",
"libloading",
]
[[package]]
name = "common"
version = "0.1.0"
@ -699,6 +740,12 @@ dependencies = [
"typenum",
]
[[package]]
name = "data-encoding"
version = "2.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476"
[[package]]
name = "deranged"
version = "0.3.11"
@ -763,10 +810,15 @@ dependencies = [
"aya-ebpf",
"aya-log-ebpf",
"common",
"network-types",
"which",
]
[[package]]
name = "either"
version = "1.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
[[package]]
name = "encoding_rs"
version = "0.8.35"
@ -808,6 +860,12 @@ dependencies = [
"windows-sys 0.59.0",
]
[[package]]
name = "find-msvc-tools"
version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a3076410a55c90011c298b04d0cfa770b00fa04e1e3c97d3f6c9de105a03844"
[[package]]
name = "flate2"
version = "1.0.35"
@ -831,20 +889,68 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a0d2fde1f7b3d48b8395d5f2de76c18a528bd6a9cdde438df747bfcba3e05d6f"
[[package]]
name = "form_urlencoded"
version = "1.2.1"
name = "foldhash"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e13624c2627564efccf4934284bdd98cbaa14e79b0b5a141218e507b3a823456"
checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb"
[[package]]
name = "form_urlencoded"
version = "1.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf"
dependencies = [
"percent-encoding",
]
[[package]]
name = "futures"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876"
dependencies = [
"futures-channel",
"futures-core",
"futures-executor",
"futures-io",
"futures-sink",
"futures-task",
"futures-util",
]
[[package]]
name = "futures-channel"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10"
dependencies = [
"futures-core",
"futures-sink",
]
[[package]]
name = "futures-core"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e"
[[package]]
name = "futures-executor"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f"
dependencies = [
"futures-core",
"futures-task",
"futures-util",
]
[[package]]
name = "futures-io"
version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6"
[[package]]
name = "futures-macro"
version = "0.3.31"
@ -874,9 +980,13 @@ version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81"
dependencies = [
"futures-channel",
"futures-core",
"futures-io",
"futures-macro",
"futures-sink",
"futures-task",
"memchr",
"pin-project-lite",
"pin-utils",
"slab",
@ -910,6 +1020,12 @@ version = "0.31.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f"
[[package]]
name = "glob"
version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280"
[[package]]
name = "h2"
version = "0.3.26"
@ -921,7 +1037,7 @@ dependencies = [
"futures-core",
"futures-sink",
"futures-util",
"http",
"http 0.2.12",
"indexmap",
"slab",
"tokio",
@ -937,7 +1053,18 @@ checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
dependencies = [
"allocator-api2",
"equivalent",
"foldhash",
"foldhash 0.1.4",
]
[[package]]
name = "hashbrown"
version = "0.16.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
dependencies = [
"allocator-api2",
"equivalent",
"foldhash 0.2.0",
]
[[package]]
@ -951,6 +1078,16 @@ dependencies = [
"itoa",
]
[[package]]
name = "http"
version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a"
dependencies = [
"bytes",
"itoa",
]
[[package]]
name = "httparse"
version = "1.10.0"
@ -1083,9 +1220,9 @@ dependencies = [
[[package]]
name = "idna"
version = "1.0.3"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "686f825264d630750a544639377bae737628043f20d38bbc029e8f29ea968a7e"
checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de"
dependencies = [
"idna_adapter",
"smallvec",
@ -1115,7 +1252,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8c9c992b02b5b4c94ea26e32fe5bccb7aa7d9f390ab5c1221ff895bc7ea8b652"
dependencies = [
"equivalent",
"hashbrown",
"hashbrown 0.15.2",
]
[[package]]
@ -1135,11 +1272,26 @@ version = "0.7.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "046fa2d4d00aea763528b4950358d0ead425372445dc8ff86312b3c69ff7727b"
dependencies = [
"bitflags 2.9.4",
"bitflags",
"cfg-if",
"libc",
]
[[package]]
name = "ipnetwork"
version = "0.21.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf370abdafd54d13e54a620e8c3e1145f28e46cc9d704bc6d94414559df41763"
[[package]]
name = "itertools"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186"
dependencies = [
"either",
]
[[package]]
name = "itoa"
version = "1.0.14"
@ -1169,11 +1321,12 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
[[package]]
name = "libbpf-sys"
version = "0.6.2+v0.6.1"
version = "1.5.1+v1.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1c2c15bbeae2b87e3a63feea85a579272ed082f0f4d8f0e7968cc9a17e1b4d69"
checksum = "912fae30b08bcbdb861d4b85bd09c05352c0ac9d7b93765ced5ca23709e7e590"
dependencies = [
"cc",
"nix",
"pkg-config",
]
@ -1183,6 +1336,27 @@ version = "0.2.174"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776"
[[package]]
name = "libloading"
version = "0.8.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55"
dependencies = [
"cfg-if",
"windows-link 0.2.0",
]
[[package]]
name = "libxdp-sys"
version = "0.2.3+1.5.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "43e5550dcb7e33b49e821ef0a1a274b4094010dae2f1fe7b48fae4c16ef15580"
dependencies = [
"bindgen",
"cc",
"libbpf-sys",
]
[[package]]
name = "linux-raw-sys"
version = "0.9.4"
@ -1227,6 +1401,15 @@ version = "0.4.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34080505efa8e45a4b816c349525ebe327ceaa8559756f0356cba97ef3bf7432"
[[package]]
name = "lru"
version = "0.16.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96051b46fc183dc9cd4a223960ef37b9af631b55191852a8274bfef064cda20f"
dependencies = [
"hashbrown 0.16.1",
]
[[package]]
name = "macros"
version = "0.1.0"
@ -1245,6 +1428,19 @@ dependencies = [
"regex-automata",
]
[[package]]
name = "maxminddb"
version = "0.26.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a197e44322788858682406c74b0b59bf8d9b4954fe1f224d9a25147f1880bba"
dependencies = [
"ipnetwork",
"log",
"memchr",
"serde",
"thiserror 2.0.16",
]
[[package]]
name = "memchr"
version = "2.7.4"
@ -1276,6 +1472,12 @@ dependencies = [
"unicase",
]
[[package]]
name = "minimal-lexical"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
[[package]]
name = "miniz_oxide"
version = "0.8.4"
@ -1311,10 +1513,14 @@ dependencies = [
"common",
"crossbeam",
"dotenvy",
"futures",
"futures-util",
"libc",
"lru",
"macros",
"maxminddb",
"mime_guess",
"network-types",
"parking_lot",
"rust-embed",
"serde",
@ -1322,10 +1528,12 @@ dependencies = [
"sysinfo",
"thiserror 2.0.16",
"tokio",
"tokio-tungstenite",
"toml 0.9.5",
"tracing",
"tracing-appender",
"tracing-subscriber",
"url",
"xsk-rs",
]
@ -1338,6 +1546,28 @@ dependencies = [
"memoffset",
]
[[package]]
name = "nix"
version = "0.30.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "74523f3a35e05aba87a1d978330aef40f67b0304ac79c1c00b294c9830543db6"
dependencies = [
"bitflags",
"cfg-if",
"cfg_aliases",
"libc",
]
[[package]]
name = "nom"
version = "7.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
dependencies = [
"memchr",
"minimal-lexical",
]
[[package]]
name = "ntapi"
version = "0.4.1"
@ -1397,7 +1627,7 @@ version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1c10c2894a6fed806ade6027bcd50662746363a9589d3ec9d9bef30a4e4bc166"
dependencies = [
"bitflags 2.9.4",
"bitflags",
]
[[package]]
@ -1417,7 +1647,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
dependencies = [
"crc32fast",
"hashbrown",
"hashbrown 0.15.2",
"indexmap",
"memchr",
]
@ -1468,9 +1698,9 @@ checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
[[package]]
name = "percent-encoding"
version = "2.3.1"
version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "pin-project-lite"
@ -1505,6 +1735,16 @@ dependencies = [
"zerocopy",
]
[[package]]
name = "prettyplease"
version = "0.2.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6837b9e10d61f45f987d50808f83d1ee3d206c66acf650c3e4ae2e1f6ddedf55"
dependencies = [
"proc-macro2",
"syn",
]
[[package]]
name = "proc-macro-error"
version = "1.0.4"
@ -1587,7 +1827,7 @@ version = "0.5.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "03a862b389f93e68874fbf580b9de08dd02facb9a788ebadaf4a3fd33cf58834"
dependencies = [
"bitflags 2.9.4",
"bitflags",
]
[[package]]
@ -1665,13 +1905,19 @@ version = "0.1.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "719b953e2095829ee67db738b3bfa9fa368c94900df327b3f07fe6e794d2fe1f"
[[package]]
name = "rustc-hash"
version = "2.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "357703d41365b4b27c590e3ed91eabb1b663f07c4c084095e60cbed4362dff0d"
[[package]]
name = "rustix"
version = "1.0.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "11181fbabf243db407ef8df94a6ce0b2f9a733bd8be4ad02b4eda9602296cac8"
dependencies = [
"bitflags 2.9.4",
"bitflags",
"errno",
"libc",
"linux-raw-sys",
@ -2042,6 +2288,18 @@ dependencies = [
"syn",
]
[[package]]
name = "tokio-tungstenite"
version = "0.28.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857"
dependencies = [
"futures-util",
"log",
"tokio",
"tungstenite",
]
[[package]]
name = "tokio-util"
version = "0.7.13"
@ -2202,6 +2460,23 @@ dependencies = [
"tracing-log",
]
[[package]]
name = "tungstenite"
version = "0.28.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442"
dependencies = [
"bytes",
"data-encoding",
"http 1.4.0",
"httparse",
"log",
"rand",
"sha1",
"thiserror 2.0.16",
"utf-8",
]
[[package]]
name = "typeid"
version = "1.0.3"
@ -2234,15 +2509,22 @@ checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
[[package]]
name = "url"
version = "2.5.4"
version = "2.5.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32f8b686cadd1473f4bd0117a5d28d36b1ade384ea9b5069a1c40aefed7fda60"
checksum = "08bc136a29a3d1758e07a9cca267be308aeebf5cfd5a10f3f67ab2097683ef5b"
dependencies = [
"form_urlencoded",
"idna",
"percent-encoding",
"serde",
]
[[package]]
name = "utf-8"
version = "0.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
[[package]]
name = "utf16_iter"
version = "1.0.5"
@ -2568,14 +2850,14 @@ checksum = "1e9df38ee2d2c3c5948ea468a8406ff0db0b29ae1ffde1bcf20ef305bcc95c51"
[[package]]
name = "xsk-rs"
version = "0.4.1"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bb9ed3e0b3a5cdd470cf7bc74000ec4888881cb275a558aaead48c32fd801dee"
checksum = "d1fef46e3505c5055082f52ada0a7f8e5dcaebdbb9eccf8e978c32382c159270"
dependencies = [
"bitflags 1.3.2",
"bitflags",
"cfg-if",
"libbpf-sys",
"libc",
"libxdp-sys",
"log",
]

View File

@ -12,7 +12,7 @@ cargo_metadata = { version = "0.22.0", default-features = false }
libc = { version = "0.2.159", default-features = false }
network-types = "0.1.0"
serde = { version = "1.0.215", features = ["derive"] }
xsk-rs = { version = "0.4.1", default-features = false }
xsk-rs = { version = "0.8.0", default-features = false }
[profile.dev]
panic = "abort"

View File

@ -1,12 +1,14 @@
[Config]
ingress_ifname = "enp4s0f0" # Ingress NIC Name
egress_ifname = "enp4s0f1" # Egress NIC Name
ingress_ifname = "enp4s0f1" # Ingress NIC Name
egress_ifname = "enp4s0f0" # Egress NIC Name
geoip_db_path = "net-guardia/static/geo/GeoLite2-City.mmdb"
combined_queue_count = 8 # NIC Combined Queue Count (ethtool -l <NIC>)
fill_queue_size = 2048 # Umem Used (Should not modify)
comp_queue_size = 2048 # Umem Used (Should not modify)
tx_queue_size = 2048 # Umem Used (Should not modify)
rx_queue_size = 2048 # Umem Used (Should not modify)
frame_size = 2048 # Umem Used (Should not modify)
frame_count = 2048 # Umem Used (Should not modify)
xsk_channel_size = 4096
fill_queue_size = 4096 # Umem Used (Should not modify)
comp_queue_size = 4096 # Umem Used (Should not modify)
tx_queue_size = 4096 # Umem Used (Should not modify)
rx_queue_size = 4096 # Umem Used (Should not modify)
frame_size = 4096 # Umem Used (Should not modify)
frame_count = 4096 # Umem Used (Should not modify)
http_server_bind_port = 8080 # Http Server Listen Port
refresh_interval = 5 # Statistics Refresh Time
refresh_interval = 5 # Statistics Refresh Time

View File

@ -8,7 +8,6 @@ common = { path = "../common", features = ["kernel"] }
aya-ebpf = { workspace = true }
aya-log-ebpf = { workspace = true }
network-types = { workspace = true }
[build-dependencies]
which = "8.0.0"

20317
models/deep_autoencoder.onnx Normal file

File diff suppressed because it is too large Load Diff

BIN
models/mlp.onnx Normal file

Binary file not shown.

View File

@ -0,0 +1,587 @@
{
"threshold": 0.15133114984430773,
"strategy_name": "Max",
"clip_params": {
"Destination Port": {
"lower": 22.0,
"upper": 63734.0
},
"Flow Duration": {
"lower": 1.0,
"upper": 118756600.72000001
},
"Total Fwd Packets": {
"lower": 1.0,
"upper": 98.0
},
"Total Backward Packets": {
"lower": 0.0,
"upper": 126.0
},
"Total Length of Fwd Packets": {
"lower": 0.0,
"upper": 13119.520000000019
},
"Total Length of Bwd Packets": {
"lower": 0.0,
"upper": 186394.64000000013
},
"Fwd Packet Length Max": {
"lower": 0.0,
"upper": 5840.0
},
"Fwd Packet Length Min": {
"lower": 0.0,
"upper": 98.0
},
"Fwd Packet Length Mean": {
"lower": 0.0,
"upper": 1932.5
},
"Fwd Packet Length Std": {
"lower": 0.0,
"upper": 2376.49858
},
"Bwd Packet Length Max": {
"lower": 0.0,
"upper": 4380.0
},
"Bwd Packet Length Min": {
"lower": 0.0,
"upper": 308.0
},
"Bwd Packet Length Mean": {
"lower": 0.0,
"upper": 1715.628244226656
},
"Bwd Packet Length Std": {
"lower": 0.0,
"upper": 1167.9936145600002
},
"Flow Bytes/s": {
"lower": 0.0,
"upper": 37000000.0
},
"Flow Packets/s": {
"lower": 0.041637660508,
"upper": 2000000.0
},
"Flow IAT Mean": {
"lower": 1.0,
"upper": 30800000.0
},
"Flow IAT Std": {
"lower": 0.0,
"upper": 44253994.319940574
},
"Flow IAT Max": {
"lower": 1.0,
"upper": 96199268.00000003
},
"Flow IAT Min": {
"lower": 0.0,
"upper": 1999969.04
},
"Fwd IAT Total": {
"lower": 0.0,
"upper": 119000000.0
},
"Fwd IAT Mean": {
"lower": 0.0,
"upper": 74300000.0
},
"Fwd IAT Std": {
"lower": 0.0,
"upper": 35894248.669783354
},
"Fwd IAT Max": {
"lower": 0.0,
"upper": 97800000.0
},
"Fwd IAT Min": {
"lower": 0.0,
"upper": 74300000.0
},
"Bwd IAT Total": {
"lower": 0.0,
"upper": 118000000.0
},
"Bwd IAT Mean": {
"lower": 0.0,
"upper": 73734410.76000024
},
"Bwd IAT Std": {
"lower": 0.0,
"upper": 30100000.0
},
"Bwd IAT Max": {
"lower": 0.0,
"upper": 93800000.0
},
"Bwd IAT Min": {
"lower": 0.0,
"upper": 73734410.76000024
},
"Fwd PSH Flags": {
"lower": 0.0,
"upper": 1.0
},
"Bwd PSH Flags": {
"lower": 0.0,
"upper": 0.0
},
"Fwd URG Flags": {
"lower": 0.0,
"upper": 0.0
},
"Bwd URG Flags": {
"lower": 0.0,
"upper": 0.0
},
"Fwd Header Length": {
"lower": 20.0,
"upper": 2376.0
},
"Bwd Header Length": {
"lower": 0.0,
"upper": 3312.0
},
"Fwd Packets/s": {
"lower": 0.02417299188,
"upper": 2000000.0
},
"Bwd Packets/s": {
"lower": 0.0,
"upper": 142857.1429
},
"Min Packet Length": {
"lower": 0.0,
"upper": 89.0
},
"Max Packet Length": {
"lower": 0.0,
"upper": 5840.0
},
"Packet Length Mean": {
"lower": 0.0,
"upper": 1153.4362532
},
"Packet Length Std": {
"lower": 0.0,
"upper": 1845.154148000012
},
"Packet Length Variance": {
"lower": 0.0,
"upper": 3404593.931120044
},
"FIN Flag Count": {
"lower": 0.0,
"upper": 1.0
},
"SYN Flag Count": {
"lower": 0.0,
"upper": 1.0
},
"RST Flag Count": {
"lower": 0.0,
"upper": 0.0
},
"PSH Flag Count": {
"lower": 0.0,
"upper": 1.0
},
"ACK Flag Count": {
"lower": 0.0,
"upper": 1.0
},
"URG Flag Count": {
"lower": 0.0,
"upper": 1.0
},
"CWE Flag Count": {
"lower": 0.0,
"upper": 0.0
},
"ECE Flag Count": {
"lower": 0.0,
"upper": 0.0
},
"Down/Up Ratio": {
"lower": 0.0,
"upper": 5.0
},
"Average Packet Size": {
"lower": 0.0,
"upper": 1182.362785582229
},
"Avg Fwd Segment Size": {
"lower": 0.0,
"upper": 1932.5
},
"Avg Bwd Segment Size": {
"lower": 0.0,
"upper": 1715.628244226656
},
"Fwd Header Length.1": {
"lower": 20.0,
"upper": 2376.0
},
"Fwd Avg Bytes/Bulk": {
"lower": 0.0,
"upper": 0.0
},
"Fwd Avg Packets/Bulk": {
"lower": 0.0,
"upper": 0.0
},
"Fwd Avg Bulk Rate": {
"lower": 0.0,
"upper": 0.0
},
"Bwd Avg Bytes/Bulk": {
"lower": 0.0,
"upper": 0.0
},
"Bwd Avg Packets/Bulk": {
"lower": 0.0,
"upper": 0.0
},
"Bwd Avg Bulk Rate": {
"lower": 0.0,
"upper": 0.0
},
"Subflow Fwd Packets": {
"lower": 1.0,
"upper": 98.0
},
"Subflow Fwd Bytes": {
"lower": 0.0,
"upper": 13119.520000000019
},
"Subflow Bwd Packets": {
"lower": 0.0,
"upper": 126.0
},
"Subflow Bwd Bytes": {
"lower": 0.0,
"upper": 186394.64000000013
},
"Init_Win_bytes_forward": {
"lower": -1.0,
"upper": 65535.0
},
"Init_Win_bytes_backward": {
"lower": -1.0,
"upper": 65535.0
},
"act_data_pkt_fwd": {
"lower": 0.0,
"upper": 49.0
},
"min_seg_size_forward": {
"lower": 20.0,
"upper": 40.0
},
"Active Mean": {
"lower": 0.0,
"upper": 3410152.040000004
},
"Active Std": {
"lower": 0.0,
"upper": 1944723.8431600018
},
"Active Max": {
"lower": 0.0,
"upper": 5717825.840000004
},
"Active Min": {
"lower": 0.0,
"upper": 2918887.7600000002
},
"Idle Mean": {
"lower": 0.0,
"upper": 94100000.0
},
"Idle Std": {
"lower": 0.0,
"upper": 20712312.015200634
},
"Idle Max": {
"lower": 0.0,
"upper": 94800000.0
},
"Idle Min": {
"lower": 0.0,
"upper": 94100000.0
}
},
"scaler_mean": [
9415.502043247605,
11217541.457604988,
5.41509711200182,
5.010050165039152,
508.74335117243135,
3302.166099044608,
217.24886707430434,
20.105687086824716,
65.40745127720037,
68.76070937185386,
394.97690859650953,
49.88046220640826,
159.85952962743096,
121.94653847258593,
872411.6556228747,
64701.11201063558,
840255.3525863544,
1516373.8724808302,
4422301.697900706,
24666.656195771673,
10897580.686685171,
1785372.7881935516,
1206278.3042178946,
4276119.216091527,
1025069.5899462275,
10158901.07420493,
1666490.9735317046,
939816.6614413982,
3689910.1546691586,
1003727.4123986625,
0.054886351088404936,
0.0,
0.0,
0.0,
140.54050707030981,
133.63606216540694,
58358.168874140014,
4983.791329731815,
19.812641079549177,
480.4855221752525,
108.29833052637488,
144.4437888623996,
99406.03379366906,
0.01811977227544623,
0.054886351088404936,
0.0,
0.25800834720207716,
0.28694244020382764,
0.11596161536441252,
0.0,
0.0,
0.6969992921551522,
122.46836156918589,
65.40745127720037,
159.85952962743843,
140.54050707030981,
0.0,
0.0,
0.0,
0.0,
0.0,
0.0,
5.41509711200182,
508.74335117243135,
5.010050165039152,
3302.166099044608,
6891.551141020379,
2404.6474695096604,
2.6520584031389776,
25.79473555242033,
61328.50855670946,
36311.697575423306,
137445.81057140988,
37549.65545208145,
3689065.3106244486,
167385.3556173429,
3883594.4613142335,
3458258.8992590285
],
"scaler_std": [
19743.81786383566,
30111390.85802201,
10.612016675700064,
12.872110268702288,
1653.7982707262067,
17387.56157755931,
645.437387791972,
22.356844266679282,
193.61621129020307,
244.137250250046,
801.628533600607,
65.65797274491328,
277.7559653593753,
269.57514761088515,
3622907.235127058,
237431.11489053545,
3293477.440659972,
5309087.889319858,
14143198.815741453,
160458.44923675407,
29985211.254190512,
8130184.699376457,
4247601.6170286415,
14271787.10745024,
7769581.505106356,
29154202.839787327,
8031691.1024753135,
3622760.030182759,
13368925.151392205,
7684129.1959599955,
0.22775829195136954,
1.0,
1.0,
1.0,
271.11142624961946,
342.17520432711024,
231133.09437362824,
14026.707464402314,
22.005832334437958,
970.4322795025132,
179.9690556845557,
280.2535015764328,
343919.1355100174,
0.13338457979891152,
0.22775829195136954,
1.0,
0.4375386154114052,
0.4523344738284939,
0.32017888613474904,
1.0,
1.0,
0.6456077980707177,
185.26118186235436,
193.61621129020307,
277.7559653594075,
271.11142624961946,
1.0,
1.0,
1.0,
1.0,
1.0,
1.0,
10.612016675700064,
1653.7982707262067,
12.872110268702288,
17387.56157755931,
14922.88749872485,
9322.158869764009,
5.784504103070997,
6.252894021045679,
322010.01591284445,
201975.98444339563,
641116.7578408231,
245957.32066446543,
13068297.83388166,
1617503.1921646637,
13724469.425904194,
12765671.772576509
],
"post_clip_min": -5.0,
"post_clip_max": 5.0,
"ae_normalization": {
"min": 4.147972858692375e-06,
"max": 1.292201307125899,
"mean": 0.0026365920400427795,
"std": 0.011092374621751085,
"median": 0.00023979156462091921,
"p90": 0.005233132974117131,
"p95": 0.01092618446409825,
"p99": 0.04010412447293844
},
"attack_labels": {
"0": "BENIGN",
"1": "Bot",
"2": "DDoS",
"3": "DoS GoldenEye",
"4": "DoS Hulk",
"5": "DoS Slowhttptest",
"6": "DoS slowloris",
"7": "FTP-Patator",
"8": "Heartbleed",
"9": "Infiltration",
"10": "PortScan",
"11": "SSH-Patator",
"12": "Web Attack <20> Brute Force",
"13": "Web Attack <20> Sql Injection",
"14": "Web Attack <20> XSS"
},
"feature_names": [
"Destination Port",
"Flow Duration",
"Total Fwd Packets",
"Total Backward Packets",
"Total Length of Fwd Packets",
"Total Length of Bwd Packets",
"Fwd Packet Length Max",
"Fwd Packet Length Min",
"Fwd Packet Length Mean",
"Fwd Packet Length Std",
"Bwd Packet Length Max",
"Bwd Packet Length Min",
"Bwd Packet Length Mean",
"Bwd Packet Length Std",
"Flow Bytes/s",
"Flow Packets/s",
"Flow IAT Mean",
"Flow IAT Std",
"Flow IAT Max",
"Flow IAT Min",
"Fwd IAT Total",
"Fwd IAT Mean",
"Fwd IAT Std",
"Fwd IAT Max",
"Fwd IAT Min",
"Bwd IAT Total",
"Bwd IAT Mean",
"Bwd IAT Std",
"Bwd IAT Max",
"Bwd IAT Min",
"Fwd PSH Flags",
"Bwd PSH Flags",
"Fwd URG Flags",
"Bwd URG Flags",
"Fwd Header Length",
"Bwd Header Length",
"Fwd Packets/s",
"Bwd Packets/s",
"Min Packet Length",
"Max Packet Length",
"Packet Length Mean",
"Packet Length Std",
"Packet Length Variance",
"FIN Flag Count",
"SYN Flag Count",
"RST Flag Count",
"PSH Flag Count",
"ACK Flag Count",
"URG Flag Count",
"CWE Flag Count",
"ECE Flag Count",
"Down/Up Ratio",
"Average Packet Size",
"Avg Fwd Segment Size",
"Avg Bwd Segment Size",
"Fwd Header Length.1",
"Fwd Avg Bytes/Bulk",
"Fwd Avg Packets/Bulk",
"Fwd Avg Bulk Rate",
"Bwd Avg Bytes/Bulk",
"Bwd Avg Packets/Bulk",
"Bwd Avg Bulk Rate",
"Subflow Fwd Packets",
"Subflow Fwd Bytes",
"Subflow Bwd Packets",
"Subflow Bwd Bytes",
"Init_Win_bytes_forward",
"Init_Win_bytes_backward",
"act_data_pkt_fwd",
"min_seg_size_forward",
"Active Mean",
"Active Std",
"Active Max",
"Active Min",
"Idle Mean",
"Idle Std",
"Idle Max",
"Idle Min"
]
}

BIN
models/random_forest.onnx Normal file

Binary file not shown.

View File

@ -13,6 +13,7 @@ actix-web = "4.11.0"
actix-ws = "0.3.0"
aya = { workspace = true }
aya-log = { workspace = true }
network-types = { workspace = true }
crossbeam = "0.8.4"
futures-util = "0.3.30"
libc = { workspace = true }
@ -24,11 +25,16 @@ serde_json = "1.0.143"
sysinfo = "0.37.0"
thiserror = "2.0.3"
tokio = { version = "1.40.0", features = ["full", "macros"] }
tokio-tungstenite = "0.28.0"
toml = "0.9.5"
tracing = "0.1.41"
tracing-appender = "0.2.3"
tracing-subscriber = { version = "0.3.20", features = ["env-filter"] }
url = "2.5.7"
xsk-rs = { workspace = true }
maxminddb = "0.26.0"
lru = "0.16.2"
futures = "0.3.31"
[build-dependencies]
cargo_metadata = { workspace = true }

View File

@ -66,7 +66,7 @@ impl AccessControl {
list_type: ListType,
address: SocketAddrV4,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let ip: u32 = (*address.ip()).to_bits().to_be();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write().await,
@ -83,7 +83,7 @@ impl AccessControl {
list_type: ListType,
address: SocketAddrV6,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let ip: u128 = (*address.ip()).to_bits().to_be();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write().await,
@ -100,7 +100,7 @@ impl AccessControl {
list_type: ListType,
address: SocketAddrV4,
) -> Result<(), Error> {
let ip: u32 = (*address.ip()).into();
let ip: u32 = (*address.ip()).to_bits().to_be();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv4_src_whitelist.write().await,
@ -117,7 +117,7 @@ impl AccessControl {
list_type: ListType,
address: SocketAddrV6,
) -> Result<(), Error> {
let ip: u128 = (*address.ip()).into();
let ip: u128 = (*address.ip()).to_bits().to_be();
let port = address.port();
let mut map_wrapper = match (direction, list_type) {
(FlowDirection::Source, ListType::White) => self.ipv6_src_whitelist.write().await,

View File

@ -2,7 +2,6 @@ pub mod access_control;
pub mod service;
pub mod statistics;
pub mod xsk_manager;
pub mod health;
use std::sync::Arc;
use std::time::Duration;
@ -16,7 +15,7 @@ use crate::core::ebpf::access_control::AccessControl;
use crate::core::ebpf::service::Service;
use crate::core::ebpf::statistics::Statistics;
use crate::core::ebpf::xsk_manager::XskManager;
use crate::core::ebpf::health::SystemHealth;
use crate::core::infrastructure::health::SystemHealth;
use crate::core::infrastructure::app_config::AppConfig;
use crate::model::error::system::SystemError;
use crate::model::error::Error;
@ -31,7 +30,11 @@ pub struct EbpfServices {
}
impl EbpfServices {
pub fn new(app_config: Arc<AppConfig>, ingress_ebpf: &mut Ebpf, egress_ebpf: &mut Ebpf) -> Result<Self, Error> {
pub fn new(
app_config: Arc<AppConfig>,
ingress_ebpf: &mut Ebpf,
egress_ebpf: &mut Ebpf,
) -> Result<Self, Error> {
let xsk_manager = XskManager::new(app_config.clone(), ingress_ebpf)?;
let access_control = AccessControl::new(ingress_ebpf)?;
let health = SystemHealth::new(&app_config)?;
@ -71,4 +74,4 @@ impl EbpfServices {
}
}
}
}
}

View File

@ -1,19 +1,24 @@
use std::collections::HashMap;
use std::net::{SocketAddrV4, SocketAddrV6};
use std::net::{IpAddr, SocketAddrV4, SocketAddrV6};
use std::sync::Arc;
use aya::maps::{HashMap as AyaHashMap, MapData};
use aya::{Ebpf, Pod};
use common::model::flow_stats::FlowStats;
use common::model::ip_address::{AddrPortV4, AddrPortV6};
use futures::future::join_all;
use macros::log;
use tokio::select;
use tokio::sync::{oneshot, RwLock};
use tokio::time::{sleep, Duration};
use tokio::sync::{RwLock, oneshot};
use tokio::time::{Duration, sleep};
use crate::core::infrastructure::app_config::AppConfig;
use crate::core::infrastructure::geoip::GeoIpService;
use crate::model::direction::{Direction, FlowDirection};
use crate::model::error::ebpf::EbpfError;
use crate::model::error::Error;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::misc::MiscError;
use crate::model::geo_stats::FlowStatsWithGeo;
use crate::model::ip_address::NativeConvert;
use crate::model::time_type::TimeType;
use crate::utils::boot_time::boot_time;
@ -21,6 +26,7 @@ use crate::utils::boot_time::boot_time;
pub struct Statistics {
app_config: Arc<AppConfig>,
boot_time: u64,
geo_ip: Option<Arc<GeoIpService>>,
ipv4_maps: HashMap<(Direction, FlowDirection, TimeType), RwLock<FlowMap<AddrPortV4>>>,
ipv6_maps: HashMap<(Direction, FlowDirection, TimeType), RwLock<FlowMap<AddrPortV6>>>,
}
@ -88,6 +94,13 @@ impl Statistics {
let boot_time = boot_time();
let mut ipv4_maps = HashMap::new();
let mut ipv6_maps = HashMap::new();
let geo_ip = match GeoIpService::new(&app_config.geoip_db_path) {
Ok(service) => Some(Arc::new(service)),
Err(err) => {
log!(MiscError::InvalidGeoIPConfiguration(err));
None
}
};
for (key, (ipv4_name, ipv6_name)) in Self::INGRESS_MAPS {
ipv4_maps.insert(key, RwLock::new(FlowMap::new(ingress_ebpf, ipv4_name)?));
ipv6_maps.insert(key, RwLock::new(FlowMap::new(ingress_ebpf, ipv6_name)?));
@ -99,6 +112,7 @@ impl Statistics {
let statistics = Statistics {
app_config,
boot_time,
geo_ip,
ipv4_maps,
ipv6_maps,
};
@ -142,13 +156,34 @@ impl Statistics {
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> HashMap<SocketAddrV4, FlowStats> {
self.ipv4_maps
) -> HashMap<SocketAddrV4, FlowStatsWithGeo> {
let flow_data = self
.ipv4_maps
.get(&(direction, flow_direction, time_type))
.unwrap()
.write()
.await
.get_map()
.get_map();
if let Some(ref geo_ip) = self.geo_ip {
let futures: Vec<_> = flow_data
.into_iter()
.map(|(addr, stats)| {
let geo_ip = geo_ip.clone();
async move {
let ip = IpAddr::V4(*addr.ip());
let geo = geo_ip.lookup(ip).await.ok().flatten();
(addr, FlowStatsWithGeo { stats, geo })
}
})
.collect();
join_all(futures).await.into_iter().collect()
} else {
flow_data
.into_iter()
.map(|(addr, stats)| (addr, FlowStatsWithGeo { stats, geo: None }))
.collect()
}
}
pub async fn get_ipv6_flow_data(
@ -156,13 +191,34 @@ impl Statistics {
direction: Direction,
flow_direction: FlowDirection,
time_type: TimeType,
) -> HashMap<SocketAddrV6, FlowStats> {
self.ipv6_maps
) -> HashMap<SocketAddrV6, FlowStatsWithGeo> {
let flow_data = self
.ipv6_maps
.get(&(direction, flow_direction, time_type))
.unwrap()
.write()
.await
.get_map()
.get_map();
if let Some(ref geo_ip) = self.geo_ip {
let futures: Vec<_> = flow_data
.into_iter()
.map(|(addr, stats)| {
let geo_ip = geo_ip.clone();
async move {
let ip = IpAddr::V6(*addr.ip());
let geo = geo_ip.lookup(ip).await.ok().flatten();
(addr, FlowStatsWithGeo { stats, geo })
}
})
.collect();
join_all(futures).await.into_iter().collect()
} else {
flow_data
.into_iter()
.map(|(addr, stats)| (addr, FlowStatsWithGeo { stats, geo: None }))
.collect()
}
}
}

View File

@ -1,26 +1,27 @@
use std::error::Error as StdError;
use std::ffi::CString;
use std::io::Write;
use std::num::NonZero;
use std::os::fd::AsRawFd;
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use aya::maps::{MapData, XskMap};
use aya::Ebpf;
use aya::maps::{MapData, XskMap};
use crossbeam::channel::{Receiver, Sender, bounded};
use crossbeam::queue::SegQueue;
use macros::log;
use parking_lot::Mutex;
use tokio::select;
use tokio::sync::oneshot;
use tokio::time::sleep;
use xsk_rs::config::{BindFlags, FrameSize, Interface, QueueSize, SocketConfig, UmemConfig, LibbpfFlags};
use xsk_rs::config::{BindFlags, FrameSize, Interface, LibxdpFlags, QueueSize, SocketConfig, UmemConfig};
use xsk_rs::{CompQueue, FillQueue, FrameDesc, RxQueue, Socket, TxQueue, Umem};
use crate::core::infrastructure::app_config::AppConfig;
use crate::model::direction::Direction;
use crate::model::config::Config;
use crate::model::error::Error;
use crate::model::error::ebpf::EbpfError;
use crate::model::error::system::SystemError;
use crate::model::error::Error;
use crate::model::log::ebpf::EbpfLog;
pub struct XskManager {
@ -30,7 +31,10 @@ pub struct XskManager {
}
impl XskManager {
pub fn new(app_config: Arc<AppConfig>, ebpf: &mut Ebpf) -> Result<Self, Error> {
pub fn new(
app_config: Arc<AppConfig>,
ebpf: &mut Ebpf,
) -> Result<Self, Error> {
let map = ebpf.take_map("XSKS_MAP").ok_or(EbpfError::MapNotFound)?;
let xsk_map = XskMap::try_from(map).map_err(EbpfError::MapOperationError)?;
@ -45,11 +49,40 @@ impl XskManager {
let config = self.app_config.config.clone();
let combined_queue_count = config.combined_queue_count;
let mut xsk_map = self.xsk_map.lock();
for queue_id in 0..combined_queue_count {
let xsk = Xsk::new(config.clone(), &mut xsk_map, queue_id)?;
let shutdown = xsk.run();
self.shutdowns.push(shutdown);
let (ingress_to_egress_tx, ingress_to_egress_rx) = bounded(config.xsk_channel_size);
let (egress_to_ingress_tx, egress_to_ingress_rx) = bounded(config.xsk_channel_size);
let ingress_xsk = XskPair::new(
config.clone(),
queue_id,
&config.ingress_ifname,
&config.egress_ifname,
Direction::Ingress
)?;
let egress_xsk = XskPair::new(
config.clone(),
queue_id,
&config.egress_ifname,
&config.ingress_ifname,
Direction::Egress,
)?;
let mut xsk_map = self.xsk_map.lock();
let ingress_fd = ingress_xsk.rx.fd().as_raw_fd();
xsk_map
.set(queue_id, ingress_fd, 0)
.map_err(EbpfError::AfXdpSetFailed)?;
drop(xsk_map);
let ingress_shutdown = ingress_xsk.run(ingress_to_egress_tx, egress_to_ingress_rx)?;
self.shutdowns.push(ingress_shutdown);
let egress_shutdown = egress_xsk.run(egress_to_ingress_tx, ingress_to_egress_rx)?;
self.shutdowns.push(egress_shutdown);
log!(EbpfLog::QueuePairStarted(queue_id));
}
Ok(())
@ -64,17 +97,26 @@ impl XskManager {
}
}
pub struct Xsk {
umem: Umem,
pub struct XskPair {
direction: Direction,
umem: Arc<Umem>,
fill_queue: FillQueue,
comp_queue: CompQueue,
tx: TxQueue,
rx: RxQueue,
frame_pool: Arc<Mutex<Vec<FrameDesc>>>,
}
impl Xsk {
pub fn new(config: Config, xsk_map: &mut XskMap<MapData>, queue_id: u32) -> Result<Self, Error> {
let ifname = CString::new(config.ingress_ifname.as_str()).map_err(|_| SystemError::UnknownError)?;
impl XskPair {
pub fn new(
config: Config,
queue_id: u32,
rx_ifname: &str,
tx_ifname: &str,
direction: Direction,
) -> Result<Self, Error> {
let rx_ifname_c = CString::new(rx_ifname).map_err(|_| SystemError::UnknownError)?;
let fill_queue_size = QueueSize::new(config.fill_queue_size).map_err(|_| SystemError::InvalidConfig)?;
let comp_queue_size = QueueSize::new(config.comp_queue_size).map_err(|_| SystemError::InvalidConfig)?;
let tx_queue_size = QueueSize::new(config.tx_queue_size).map_err(|_| SystemError::InvalidConfig)?;
@ -96,164 +138,218 @@ impl Xsk {
.tx_queue_size(tx_queue_size)
.rx_queue_size(rx_queue_size)
.bind_flags(BindFlags::XDP_ZEROCOPY)
.libbpf_flags(LibbpfFlags::XSK_LIBBPF_FLAGS_INHIBIT_PROG_LOAD)
.libxdp_flags(LibxdpFlags::XSK_LIBXDP_FLAGS_INHIBIT_PROG_LOAD)
.build();
let interface = Interface::new(ifname);
let (tx, rx, queue) =
Socket::new(socket_config, &umem, &interface, queue_id).map_err(EbpfError::SocketSetFailed)?;
let interface = Interface::new(rx_ifname_c);
let (tx, rx, queue) = unsafe {
Socket::new(socket_config, &umem, &interface, queue_id)
.map_err(EbpfError::SocketSetFailed)?
};
let (mut fill_queue, comp_queue) = queue.ok_or(EbpfError::UnknownError)?;
let socket_fd = rx.fd().as_raw_fd();
let total_frames = frame_descs.len();
let fill_frames_count = (total_frames / 2).min(config.fill_queue_size as usize);
xsk_map.set(queue_id, socket_fd, 0).map_err(EbpfError::AfXdpSetFailed)?;
let fill_frames: Vec<FrameDesc> = frame_descs.iter().take(fill_frames_count).copied().collect();
let frames: Vec<FrameDesc> = frame_descs
.iter()
.take(config.fill_queue_size as usize)
.copied()
.collect();
let submitted = unsafe { fill_queue.produce(&frames) };
if submitted != frames.len() {
let submitted = unsafe { fill_queue.produce(&fill_frames) };
if submitted != fill_frames.len() {
log!(EbpfLog::QueueInitIncomplete);
}
Ok(Self {
umem,
let pool_frames: Vec<FrameDesc> = frame_descs.iter().skip(fill_frames_count).copied().collect();
let xsk_pair = Self {
direction,
umem: Arc::new(umem),
fill_queue,
comp_queue,
tx,
rx,
})
frame_pool: Arc::new(Mutex::new(pool_frames)),
};
Ok(xsk_pair)
}
pub fn run(mut self) -> oneshot::Sender<()> {
let (sender, receiver) = oneshot::channel();
tokio::spawn(async move {
let mut receiver = receiver;
loop {
select! {
biased;
_ = &mut receiver => break,
_ = self.process_events() => {},
pub fn run(
mut self,
forward_tx: Sender<Vec<u8>>,
forward_rx: Receiver<Vec<u8>>,
) -> Result<oneshot::Sender<()>, EbpfError> {
let (shutdown_tx, shutdown_rx) = oneshot::channel();
let thread_name = format!("xsk-{:?}", self.direction);
thread::Builder::new()
.name(thread_name.clone())
.spawn(move || {
let mut shutdown_rx = Some(shutdown_rx);
loop {
if let Some(ref mut rx) = shutdown_rx {
match rx.try_recv() {
Ok(_) | Err(oneshot::error::TryRecvError::Closed) => {
break;
}
Err(oneshot::error::TryRecvError::Empty) => {}
}
}
if let Err(e) = self.process_comp_queue() {
log!(EbpfLog::CompQueueError {
error: format!("{:?}", e)
});
}
if let Err(e) = self.process_rx_queue(&forward_tx) {
log!(EbpfLog::RXQueueError {
error: format!("{:?}", e)
});
}
if let Err(e) = self.process_tx_queue(&forward_rx) {
log!(EbpfLog::TXQueueError {
error: format!("{:?}", e)
});
}
thread::sleep(Duration::from_micros(1));
}
}
});
sender
log!(EbpfLog::XSKShutdown);
})
.map(|_| shutdown_tx)
.map_err(|e| {
log!(EbpfLog::ThreadSpawnFailed {
thread_name: thread_name.clone(),
error: e.to_string()
});
EbpfError::ThreadSpawnFailed(e)
})
}
async fn process_events(&mut self) {
let mut comp_descs = vec![FrameDesc::default(); 64];
let comp_count = unsafe { self.comp_queue.consume(&mut comp_descs) };
fn process_comp_queue(&mut self) -> Result<(), EbpfError> {
let mut comp_descs = vec![FrameDesc::default(); 256];
if comp_count > 0 {
let submitted = unsafe { self.fill_queue.produce(&comp_descs[..comp_count]) };
if submitted != comp_count {
log!(EbpfLog::QueueRefillIncomplete);
let nb_completed = unsafe { self.comp_queue.consume(&mut comp_descs) };
if nb_completed > 0 {
let mut pool = self.frame_pool.lock();
for desc in comp_descs.iter().take(nb_completed) {
pool.push(*desc);
}
}
let mut packet_count = 0;
let mut tx_descs = Vec::with_capacity(64);
let mut rx_descs = vec![FrameDesc::default(); 64];
Ok(())
}
fn process_rx_queue(&mut self, forward_tx: &Sender<Vec<u8>>) -> Result<(), EbpfError> {
let mut rx_descs = vec![FrameDesc::default(); 64];
let rx_count = unsafe { self.rx.consume(&mut rx_descs) };
for rx_desc in &rx_descs[..rx_count] {
let data = unsafe { self.umem.data(rx_desc) };
let packet_data = &data.contents()[..rx_desc.lengths().data()];
if rx_count > 0 {
for rx_desc in rx_descs.iter().take(rx_count) {
let lengths = rx_desc.lengths();
let packet_len = lengths.data() as usize;
let packet_copy = packet_data.to_vec();
let data = unsafe { self.umem.data(rx_desc) };
let packet_data = data.contents()[..packet_len].to_vec();
Self::print_packet_info(&packet_copy);
tx_descs.push(*rx_desc);
packet_count += 1;
}
if !tx_descs.is_empty() {
let tx_submitted = unsafe { self.tx.produce(&tx_descs) };
if tx_submitted != tx_descs.len() {
log!(EbpfLog::QueueRefillIncomplete);
for desc in &tx_descs[tx_submitted..] {
unsafe {
let _ = self.fill_queue.produce(&[*desc]);
if let Err(e) = forward_tx.try_send(packet_data) {
match e {
crossbeam::channel::TrySendError::Full(_) => {
log!(EbpfLog::ForwardChannelFull);
}
crossbeam::channel::TrySendError::Disconnected(_) => {
log!(EbpfLog::ForwardChannelDisconnected);
}
}
}
} else {
if let Err(err) = self.tx.wakeup() {
log!(EbpfError::WakeupTXFailed(err))
}
unsafe {
let produced = self.fill_queue.produce(&rx_descs[..rx_count]);
if produced != rx_count {
log!(EbpfLog::FillQueueIncomplete {
produced,
expected: rx_count
});
}
}
}
if packet_count > 0 {
println!("Processed {} packets", packet_count);
}
if packet_count == 0 {
sleep(Duration::from_millis(1)).await;
}
Ok(())
}
fn print_packet_info(packet_data: &[u8]) {
if packet_data.len() < 14 {
println!("Packet too small: {} bytes", packet_data.len());
return;
}
println!("Received packet: {} bytes", packet_data.len());
let print_len = std::cmp::min(64, packet_data.len());
print!("Data: ");
for i in 0..print_len {
print!("{:02x} ", packet_data[i]);
if (i + 1) % 16 == 0 {
print!("\n ");
fn process_tx_queue(&mut self, forward_rx: &Receiver<Vec<u8>>) -> Result<(), EbpfError> {
let mut packets_to_send = Vec::with_capacity(64);
while let Ok(packet) = forward_rx.try_recv() {
packets_to_send.push(packet);
if packets_to_send.len() >= 64 {
break;
}
}
println!();
let dst_mac = &packet_data[0..6];
let src_mac = &packet_data[6..12];
let eth_type = u16::from_be_bytes([packet_data[12], packet_data[13]]);
if packets_to_send.is_empty() {
return Ok(());
}
println!(
"Ethernet: src={:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}, dst={:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}, type=0x{:04x}",
src_mac[0],
src_mac[1],
src_mac[2],
src_mac[3],
src_mac[4],
src_mac[5],
dst_mac[0],
dst_mac[1],
dst_mac[2],
dst_mac[3],
dst_mac[4],
dst_mac[5],
eth_type
);
let _ = self.process_comp_queue();
match eth_type {
0x0800 => {
println!(" -> IPv4 packet");
if packet_data.len() >= 34 {
let src_ip = &packet_data[26..30];
let dst_ip = &packet_data[30..34];
println!(
" IP: {}.{}.{}.{} -> {}.{}.{}.{}",
src_ip[0], src_ip[1], src_ip[2], src_ip[3], dst_ip[0], dst_ip[1], dst_ip[2], dst_ip[3]
);
let pool_size = {
let pool = self.frame_pool.lock();
pool.len()
};
if pool_size == 0 {
log!(EbpfLog::FramePoolExhausted {
send_len: packets_to_send.len()
});
return Ok(());
}
let mut frames = Vec::with_capacity(packets_to_send.len());
{
let mut pool = self.frame_pool.lock();
let available = pool.len().min(packets_to_send.len());
for _ in 0..available {
if let Some(frame) = pool.pop() {
frames.push(frame);
}
}
0x86DD => println!(" -> IPv6 packet"),
0x0806 => println!(" -> ARP packet"),
_ => println!(" -> Unknown protocol"),
}
println!("---");
if frames.is_empty() {
log!(EbpfLog::NoFramesAvailable);
return Ok(());
}
for (frame, packet) in frames.iter_mut().zip(packets_to_send.iter()) {
unsafe {
self.umem
.data_mut(frame)
.cursor()
.write_all(packet)
.map_err(EbpfError::AfXdpSetFailed)?;
}
}
let _nb_submitted = unsafe { self.tx.produce(&frames) };
if let Err(e) = self.tx.wakeup() {
if e.kind() != std::io::ErrorKind::WouldBlock {
log!(EbpfLog::TXWakeupFailed {
error: e.to_string()
});
}
}
Ok(())
}
}
}

View File

@ -0,0 +1,113 @@
use std::net::IpAddr;
use std::path::Path;
use std::sync::Arc;
use maxminddb::{geoip2, MaxMindDbError, Reader};
use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use lru::LruCache;
use std::num::NonZeroUsize;
use tokio::task;
use crate::model::geo_stats::GeoLocation;
pub struct GeoIpService {
reader: Arc<Reader<Vec<u8>>>,
cache: Arc<RwLock<LruCache<IpAddr, Option<GeoLocation>>>>,
}
impl GeoIpService {
pub fn new<P: AsRef<Path>>(db_path: P) -> Result<Self, MaxMindDbError> {
Self::with_cache_size(db_path, 10000)
}
pub fn with_cache_size<P: AsRef<Path>>(
db_path: P,
cache_size: usize,
) -> Result<Self, MaxMindDbError> {
let reader = Reader::open_readfile(db_path)?;
let cache_capacity = NonZeroUsize::new(cache_size)
.unwrap_or_else(|| NonZeroUsize::new(10000).unwrap());
Ok(Self {
reader: Arc::new(reader),
cache: Arc::new(RwLock::new(LruCache::new(cache_capacity))),
})
}
pub async fn lookup(&self, ip: IpAddr) -> Result<Option<GeoLocation>, MaxMindDbError> {
{
let cache = self.cache.read().await;
if let Some(cached) = cache.peek(&ip) {
return Ok(cached.clone());
}
}
let reader = self.reader.clone();
let result = task::spawn_blocking(move || {
Self::lookup_from_db_blocking(&reader, ip)
})
.await
.map_err(|e| MaxMindDbError::InvalidDatabase(format!("Task join error: {}", e)))??;
{
let mut cache = self.cache.write().await;
cache.put(ip, result.clone());
}
Ok(result)
}
fn lookup_from_db_blocking(
reader: &Reader<Vec<u8>>,
ip: IpAddr,
) -> Result<Option<GeoLocation>, MaxMindDbError> {
let city_option: Option<geoip2::City> = reader.lookup(ip)?;
Ok(city_option.map(|city| {
let country_name = city
.country
.as_ref()
.and_then(|c| c.names.as_ref())
.and_then(|n| n.get("en"))
.map(|s| s.to_string());
let country_code = city
.country
.as_ref()
.and_then(|c| c.iso_code)
.map(|s| s.to_string());
let city_name = city
.city
.as_ref()
.and_then(|c| c.names.as_ref())
.and_then(|n| n.get("en"))
.map(|s| s.to_string());
let latitude = city.location.as_ref().and_then(|l| l.latitude);
let longitude = city.location.as_ref().and_then(|l| l.longitude);
let timezone = city
.location
.as_ref()
.and_then(|l| l.time_zone)
.map(|s| s.to_string());
GeoLocation {
country: country_name,
country_code,
city: city_name,
latitude,
longitude,
timezone,
}
}))
}
pub async fn cache_stats(&self) -> (usize, usize) {
let cache = self.cache.read().await;
(cache.len(), cache.cap().get())
}
}

View File

@ -1 +1,3 @@
pub mod app_config;
pub mod health;
pub mod geoip;

View File

@ -15,6 +15,8 @@ use crate::model::error::ebpf::EbpfError;
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::error::Error;
use crate::model::error::ml::MLError;
use crate::model::log::ml::MLLog;
use crate::model::log::system::SystemLog;
use crate::utils::logging::Logging;
use crate::web::api::{control, default, misc};
@ -22,7 +24,6 @@ use crate::web::api::{control, default, misc};
pub struct System {
pub app_config: Arc<AppConfig>,
pub ebpf_services: Arc<EbpfServices>,
pub ingress_ebpf: Ebpf,
pub egress_ebpf: Ebpf,
#[allow(dead_code)]
@ -36,11 +37,13 @@ impl System {
let (mut ingress_ebpf, ingress_program_array) = System::get_ingress_ebpf()?;
let (mut egress_ebpf, egress_program_array) = System::get_egress_ebpf()?;
let app_config = Arc::new(AppConfig::new()?);
let ebpf_services = Arc::new(EbpfServices::new(
app_config.clone(),
&mut ingress_ebpf,
&mut egress_ebpf,
)?);
let system = System {
app_config,
ebpf_services,
@ -59,6 +62,7 @@ impl System {
self.aya_log_init()?;
log!(SystemLog::InitializeComplete);
self.attach_ebpf()?;
ebpf_services.run().await?;
self.run_http_server().await?;
Ok(())
@ -67,6 +71,7 @@ impl System {
pub async fn terminate(&self) -> Result<(), Error> {
let ebpf_services = self.ebpf_services.clone();
log!(SystemLog::Terminating);
ebpf_services.terminate();
log!(SystemLog::TerminateComplete);
Ok(())
@ -143,7 +148,7 @@ impl System {
env!("OUT_DIR"),
"/net-guardia-ingress"
)))
.map_err(EbpfError::EbpfNotFound)?;
.map_err(EbpfError::EbpfNotFound)?;
let program_array = ingress_ebpf.take_map("PROGRAM_ARRAY").ok_or(EbpfError::MapNotFound)?;
let mut program_array = ProgramArray::try_from(program_array).map_err(EbpfError::MapOperationError)?;
Self::load_program(&mut ingress_ebpf, &mut program_array, "access_control", ingress::ACCESS_CONTROL)?;
@ -158,7 +163,7 @@ impl System {
env!("OUT_DIR"),
"/net-guardia-egress"
)))
.map_err(EbpfError::EbpfNotFound)?;
.map_err(EbpfError::EbpfNotFound)?;
let program_array = egress_ebpf.take_map("PROGRAM_ARRAY").ok_or(EbpfError::MapNotFound)?;
let mut program_array = ProgramArray::try_from(program_array).map_err(EbpfError::MapOperationError)?;
Self::load_program(&mut egress_ebpf, &mut program_array, "statistics", egress::STATISTICS)?;

View File

@ -2,6 +2,7 @@ mod core;
mod model;
mod utils;
mod web;
mod ml;
use crate::core::system::System;
use crate::model::error::Error;

View File

View File

@ -1,4 +1,5 @@
use serde::{Deserialize, Serialize};
use std::path::PathBuf;
#[derive(Debug, Deserialize)]
pub struct ConfigTable {
@ -10,7 +11,9 @@ pub struct ConfigTable {
pub struct Config {
pub ingress_ifname: String,
pub egress_ifname: String,
pub geoip_db_path: PathBuf,
pub combined_queue_count: u32,
pub xsk_channel_size: usize,
pub fill_queue_size: u32,
pub comp_queue_size: u32,
pub tx_queue_size: u32,
@ -18,5 +21,5 @@ pub struct Config {
pub frame_size: u32,
pub frame_count: u32,
pub refresh_interval: u64,
pub http_server_bind_port: u16,
pub http_server_bind_port: u16
}

View File

@ -1,6 +1,6 @@
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Copy, Clone, Eq, PartialEq, Hash)]
#[derive(Serialize, Deserialize, Copy, Clone, Eq, PartialEq, Hash, Debug)]
#[serde(rename_all = "lowercase")]
pub enum Direction {
Ingress,

View File

@ -1,4 +1,4 @@
use macros::traceable;
use macros::{loggable, traceable};
use tracing;
traceable! {
@ -52,5 +52,18 @@ traceable! {
#[no_source]
#[error("Unknown error")]
UnknownError => tracing::Level::ERROR,
#[error("Failed to spawn XSK thread")]
ThreadSpawnFailed => tracing::Level::ERROR,
#[error("Completion queue processing failed")]
CompQueueError => tracing::Level::ERROR,
#[error("RX queue processing failed")]
RXQueueError => tracing::Level::ERROR,
#[error("TX queue processing failed")]
TXQueueError => tracing::Level::ERROR,
}
}

View File

@ -18,5 +18,8 @@ traceable! {
#[no_source]
#[error("Network interface '{interface}' not found")]
NetworkInterfaceNotFound { interface: String } => tracing::Level::ERROR,
#[error("Invalid GeoIP configuration")]
InvalidGeoIPConfiguration => tracing::Level::ERROR,
}
}

View File

@ -0,0 +1,10 @@
use std::path::PathBuf;
use macros::traceable;
traceable! {
MLError {
#[error("Initialize Machine Learning detection failed")]
InitializeFailed => tracing::Level::ERROR,
}
}

View File

@ -2,6 +2,7 @@ pub mod ebpf;
pub mod http;
pub mod io;
pub mod misc;
pub mod ml;
pub mod system;
use serde::{Deserialize, Serialize};
@ -10,6 +11,7 @@ use crate::model::error::ebpf::EbpfError;
use crate::model::error::http::HttpError;
use crate::model::error::io::IOError;
use crate::model::error::misc::MiscError;
use crate::model::error::ml::MLError;
use crate::model::error::system::SystemError;
#[derive(Clone, Debug, thiserror::Error, Serialize, Deserialize)]
@ -19,6 +21,8 @@ pub enum Error {
#[error("{0}")]
Http(HttpError),
#[error("{0}")]
ML(MLError),
#[error("{0}")]
IO(IOError),
#[error("{0}")]
Misc(MiscError),

View File

@ -0,0 +1,19 @@
use serde::{Deserialize, Serialize};
use common::model::flow_stats::FlowStats;
#[derive(Debug, Clone, Serialize)]
pub struct FlowStatsWithGeo {
#[serde(flatten)]
pub stats: FlowStats,
pub geo: Option<GeoLocation>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GeoLocation {
pub country: Option<String>,
pub country_code: Option<String>,
pub city: Option<String>,
pub latitude: Option<f64>,
pub longitude: Option<f64>,
pub timezone: Option<String>,
}

View File

@ -13,11 +13,13 @@ impl NativeConvert for IPv4 {
type Native = Ipv4Addr;
fn into_native(self) -> Self::Native {
Ipv4Addr::from(self)
// eBPF 儲存的是 big-endian需要轉換成 host order
Ipv4Addr::from(u32::from_be(self))
}
fn from_native(native: Self::Native) -> Self {
native.to_bits()
// 轉回 big-endian 給 eBPF
native.to_bits().to_be()
}
}
@ -25,11 +27,11 @@ impl NativeConvert for IPv6 {
type Native = Ipv6Addr;
fn into_native(self) -> Self::Native {
Ipv6Addr::from(self)
Ipv6Addr::from(u128::from_be(self))
}
fn from_native(native: Self::Native) -> Self {
native.to_bits()
native.to_bits().to_be()
}
}
@ -37,11 +39,17 @@ impl NativeConvert for AddrPortV4 {
type Native = SocketAddrV4;
fn into_native(self) -> Self::Native {
SocketAddrV4::new(Ipv4Addr::from(self.ip()), self.port())
SocketAddrV4::new(
Ipv4Addr::from(u32::from_be(self.ip())),
self.port()
)
}
fn from_native(native: Self::Native) -> Self {
AddrPortV4::new((*native.ip()).to_bits(), native.port())
AddrPortV4::new(
native.ip().to_bits().to_be(),
native.port()
)
}
}
@ -49,10 +57,18 @@ impl NativeConvert for AddrPortV6 {
type Native = SocketAddrV6;
fn into_native(self) -> Self::Native {
SocketAddrV6::new(Ipv6Addr::from(self.ip()), self.port(), 0, 0)
SocketAddrV6::new(
Ipv6Addr::from(u128::from_be(self.ip())),
self.port(),
0,
0
)
}
fn from_native(native: Self::Native) -> Self {
AddrPortV6::new((*native.ip()).to_bits(), native.port())
AddrPortV6::new(
native.ip().to_bits().to_be(),
native.port()
)
}
}
}

View File

@ -14,5 +14,41 @@ loggable! {
#[error("No frames submit to queue")]
NoFrameSubmit => tracing::Level::WARN,
#[error("Queue pair {queue_id} started successfully")]
QueuePairStarted { queue_id: u32 } => tracing::Level::INFO,
#[error("XSK thread shutting down")]
XSKShutdown => tracing::Level::INFO,
#[error("Frame pool exhausted! Pending TX: {send_len} packets")]
FramePoolExhausted { send_len: usize } => tracing::Level::WARN,
#[error("No frames available for TX")]
NoFramesAvailable => tracing::Level::WARN,
#[error("TX wakeup failed: {error}")]
TXWakeupFailed { error: String } => tracing::Level::WARN,
#[error("Completion queue processing error: {error}")]
CompQueueError { error: String } => tracing::Level::ERROR,
#[error("RX queue processing error: {error}")]
RXQueueError { error: String } => tracing::Level::ERROR,
#[error("TX queue processing error: {error}")]
TXQueueError { error: String } => tracing::Level::ERROR,
#[error("Failed to spawn thread '{thread_name}': {error}")]
ThreadSpawnFailed { thread_name: String, error: String } => tracing::Level::ERROR,
#[error("Forward channel full, dropping packet")]
ForwardChannelFull => tracing::Level::WARN,
#[error("Forward channel disconnected")]
ForwardChannelDisconnected => tracing::Level::ERROR,
#[error("Fill queue incomplete: produced {produced}, expected {expected}")]
FillQueueIncomplete { produced: usize, expected: usize } => tracing::Level::WARN,
}
}
}

View File

@ -0,0 +1,9 @@
use macros::loggable;
use tracing;
loggable! {
MiscLog {
#[error("GeoIP features will be disabled")]
GeoIPDisabled => tracing::Level::WARN,
}
}

View File

@ -0,0 +1,45 @@
use macros::loggable;
use tracing;
loggable! {
MLLog {
#[error("Initializing Machine Learning with inference URL: {url}")]
Initializing { url: String } => tracing::Level::INFO,
#[error("Continuing without Machine Learning detection")]
Skiped => tracing::Level::WARN,
#[error("Machine Learning detection is disabled (no ml_inference_url configured)")]
Disabled => tracing::Level::INFO,
#[error("Machine Learning detection starting")]
Starting => tracing::Level::INFO,
#[error("Machine Learning detection ready")]
Ready => tracing::Level::INFO,
#[error("Machine Learning detection shutdown")]
Shutdown => tracing::Level::INFO,
#[error("Machine Learning channel disconnected")]
ChannelDisconnected => tracing::Level::WARN,
#[error("Failed to forward packet: {error}")]
ForwardPacketFailed { error: String } => tracing::Level::WARN,
#[error("Attach XDP program success")]
AttachProgramSuccess => tracing::Level::INFO,
#[error("Queue initialization incomplete")]
QueueInitIncomplete => tracing::Level::WARN,
#[error("Queue refill incomplete")]
QueueRefillIncomplete => tracing::Level::WARN,
#[error("No frames submit to queue")]
NoFrameSubmit => tracing::Level::WARN,
#[error("Queue pair {queue_id} started successfully")]
QueuePairStarted { queue_id: u32 } => tracing::Level::INFO,
}
}

View File

@ -1,3 +1,5 @@
pub mod ebpf;
pub mod http;
pub mod ml;
pub mod system;
mod misc;

View File

@ -1,6 +1,7 @@
pub mod config;
pub mod direction;
pub mod error;
pub mod geo_stats;
pub mod ip_address;
pub mod list_type;
pub mod log;

View File

@ -1,4 +1,6 @@
pub mod ip_address;
pub mod logging;
pub mod static_files;
pub mod boot_time;
pub mod packet_parser;
pub mod ip_address;

View File

@ -0,0 +1,112 @@
use common::model::event::{Event, IPv4Event, IPv6Event};
use network_types::ip::IpProto;
/// Parse raw packet bytes into an Event
pub fn parse_packet(packet_data: &[u8]) -> Option<Event> {
if packet_data.len() < 14 {
return None;
}
let eth_type = u16::from_be_bytes([packet_data[12], packet_data[13]]);
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()?
.as_secs();
match eth_type {
0x0800 => parse_ipv4(packet_data, timestamp),
0x86DD => parse_ipv6(packet_data, timestamp),
_ => None,
}
}
fn parse_ipv4(packet_data: &[u8], timestamp: u64) -> Option<Event> {
// Ethernet header (14) + minimum IPv4 header (20) = 34 bytes
if packet_data.len() < 34 {
return None;
}
let ip_header = &packet_data[14..];
// Parse IPv4 header
let protocol = ip_header[9];
let source_ip = u32::from_be_bytes([ip_header[12], ip_header[13], ip_header[14], ip_header[15]]);
let destination_ip = u32::from_be_bytes([ip_header[16], ip_header[17], ip_header[18], ip_header[19]]);
// Get IP header length
let ihl = (ip_header[0] & 0x0F) as usize * 4;
// Total length
let total_len = u16::from_be_bytes([ip_header[2], ip_header[3]]) as u32;
// Parse transport layer (TCP/UDP)
let (source_port, destination_port) = if packet_data.len() >= 14 + ihl + 4 {
let transport_header = &ip_header[ihl..];
let src_port = u16::from_be_bytes([transport_header[0], transport_header[1]]);
let dst_port = u16::from_be_bytes([transport_header[2], transport_header[3]]);
(src_port, dst_port)
} else {
(0, 0)
};
let event = IPv4Event {
protocol: unsafe { std::mem::transmute::<u8, IpProto>(protocol) },
source_ip,
destination_ip,
source_port,
destination_port,
len: total_len,
timestamp,
};
Some(Event::IPv4(event))
}
fn parse_ipv6(packet_data: &[u8], timestamp: u64) -> Option<Event> {
// Ethernet header (14) + minimum IPv6 header (40) = 54 bytes
if packet_data.len() < 54 {
return None;
}
let ip_header = &packet_data[14..];
// Parse IPv6 header
let protocol = ip_header[6];
// Source IPv6 address (16 bytes starting at offset 8)
let mut source_ip_bytes = [0u8; 16];
source_ip_bytes.copy_from_slice(&ip_header[8..24]);
let source_ip = u128::from_be_bytes(source_ip_bytes);
// Destination IPv6 address (16 bytes starting at offset 24)
let mut dest_ip_bytes = [0u8; 16];
dest_ip_bytes.copy_from_slice(&ip_header[24..40]);
let destination_ip = u128::from_be_bytes(dest_ip_bytes);
// Payload length
let payload_len = u16::from_be_bytes([ip_header[4], ip_header[5]]) as u32;
let total_len = payload_len + 40; // IPv6 header is always 40 bytes
// Parse transport layer (TCP/UDP)
let (source_port, destination_port) = if packet_data.len() >= 54 + 4 {
let transport_header = &ip_header[40..];
let src_port = u16::from_be_bytes([transport_header[0], transport_header[1]]);
let dst_port = u16::from_be_bytes([transport_header[2], transport_header[3]]);
(src_port, dst_port)
} else {
(0, 0)
};
let event = IPv6Event {
protocol: unsafe { std::mem::transmute::<u8, IpProto>(protocol) },
source_ip,
destination_ip,
source_port,
destination_port,
len: total_len,
timestamp,
};
Some(Event::IPv6(event))
}

View File

@ -1,6 +1,6 @@
use actix_web::{get, web, HttpRequest, HttpResponse, Responder, Scope};
use crate::core::ebpf::health::SystemHealth;
use crate::core::infrastructure::health::SystemHealth;
use crate::web::websocket::health_websocket;
pub fn initialize() -> Scope {

View File

@ -1,12 +1,10 @@
use std::sync::Arc;
use actix_web::{web, HttpRequest, HttpResponse, Result};
use actix_ws::{handle, Message, MessageStream, Session};
use futures_util::StreamExt;
use macros::log;
use tokio::sync::broadcast;
use crate::core::ebpf::health::{SystemHealth, SystemHealthMetrics};
use crate::core::infrastructure::health::{SystemHealth, SystemHealthMetrics};
use crate::model::error::http::HttpError;
use crate::model::error::misc::MiscError;
use crate::model::log::http::HttpLog;
@ -86,7 +84,7 @@ async fn send_metrics(session: &mut Session, metrics: &SystemHealthMetrics) -> b
Ok(json) => session.text(json).await.is_ok(),
Err(err) => {
log!(MiscError::SerializeError(err));
true
false
}
}
}

Binary file not shown.

After

Width:  |  Height:  |  Size: 59 MiB