mirror/component
1
0
Fork 0
mirror of https://github.com/fuckpiracyshield/component.git synced 2024-12-22 18:40:51 +01:00
Code Issues Projects Releases Packages Wiki Activity

Added CIDR verifier. Various fixes.

Browse source
This commit is contained in:
Daniele Maglie 2024-02-07 14:55:23 +01:00
parent 3d3a9cd527
commit 61e2ea5e1f
9 changed files with 432 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
pyproject.toml
View file

@ -1,5 +1,12 @@
[build-system] [build-system]
requires = [ requires = [
"setuptools>=54", "setuptools>=54",
"setuptools-rust"
] ]
build-backend = "setuptools.build_meta" build-backend = "setuptools.build_meta"
[[tool.setuptools-rust.ext-modules]]
target = "piracyshield_component_cidr_verifier"
path = "rs/cidr/Cargo.toml"
binding = "PyO3"
py-limited-api = "auto"

310
rs/cidr/Cargo.lock generated Normal file
View file

@ -0,0 +1,310 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 3
[[package]]
name = "autocfg"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"
[[package]]
name = "bitflags"
version = "1.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
[[package]]
name = "cfg-if"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"
[[package]]
name = "heck"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
[[package]]
name = "indoc"
version = "2.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e186cfbae8084e513daff4240b4797e342f988cecda4fb6c939150f96315fd8"
[[package]]
name = "ipnetwork"
version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bf466541e9d546596ee94f9f69590f89473455f88372423e0008fc1a7daf100e"
dependencies = [
"serde",
]
[[package]]
name = "libc"
version = "0.2.152"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13e3bf6590cbc649f4d1a3eefc9d5d6eb746f5200ffb04e5e142700b8faa56e7"
[[package]]
name = "lock_api"
version = "0.4.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c168f8615b12bc01f9c17e2eb0cc07dcae1940121185446edc3744920e8ef45"
dependencies = [
"autocfg",
"scopeguard",
]
[[package]]
name = "memoffset"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a634b1c61a95585bd15607c6ab0c4e5b226e695ff2800ba0cdccddf208c406c"
dependencies = [
"autocfg",
]
[[package]]
name = "once_cell"
version = "1.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
[[package]]
name = "parking_lot"
version = "0.12.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
dependencies = [
"lock_api",
"parking_lot_core",
]
[[package]]
name = "parking_lot_core"
version = "0.9.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4c42a9226546d68acdd9c0a280d17ce19bfe27a46bf68784e4066115788d008e"
dependencies = [
"cfg-if",
"libc",
"redox_syscall",
"smallvec",
"windows-targets",
]
[[package]]
name = "proc-macro2"
version = "1.0.78"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e2422ad645d89c99f8f3e6b88a9fdeca7fabeac836b1002371c4367c8f984aae"
dependencies = [
"unicode-ident",
]
[[package]]
name = "pyo3"
version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a89dc7a5850d0e983be1ec2a463a171d20990487c3cfcd68b5363f1ee3d6fe0"
dependencies = [
"cfg-if",
"indoc",
"libc",
"memoffset",
"parking_lot",
"pyo3-build-config",
"pyo3-ffi",
"pyo3-macros",
"unindent",
]
[[package]]
name = "pyo3-build-config"
version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "07426f0d8fe5a601f26293f300afd1a7b1ed5e78b2a705870c5f30893c5163be"
dependencies = [
"once_cell",
"target-lexicon",
]
[[package]]
name = "pyo3-ffi"
version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dbb7dec17e17766b46bca4f1a4215a85006b4c2ecde122076c562dd058da6cf1"
dependencies = [
"libc",
"pyo3-build-config",
]
[[package]]
name = "pyo3-macros"
version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "05f738b4e40d50b5711957f142878cfa0f28e054aa0ebdfc3fd137a843f74ed3"
dependencies = [
"proc-macro2",
"pyo3-macros-backend",
"quote",
"syn",
]
[[package]]
name = "pyo3-macros-backend"
version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0fc910d4851847827daf9d6cdd4a823fbdaab5b8818325c5e97a86da79e8881f"
dependencies = [
"heck",
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "quote"
version = "1.0.35"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "291ec9ab5efd934aaf503a6466c5d5251535d108ee747472c3977cc5acc868ef"
dependencies = [
"proc-macro2",
]
[[package]]
name = "redox_syscall"
version = "0.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4722d768eff46b75989dd134e5c353f0d6296e5aaa3132e776cbdb56be7731aa"
dependencies = [
"bitflags",
]
[[package]]
name = "rs_cidr_verifier"
version = "0.1.0"
dependencies = [
"ipnetwork",
"pyo3",
]
[[package]]
name = "scopeguard"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
[[package]]
name = "serde"
version = "1.0.196"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "870026e60fa08c69f064aa766c10f10b1d62db9ccd4d0abb206472bee0ce3b32"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.196"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "33c85360c95e7d137454dc81d9a4ed2b8efd8fbe19cee57357b32b9771fccb67"
dependencies = [
"proc-macro2",
"quote",
"syn",
]
[[package]]
name = "smallvec"
version = "1.13.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e6ecd384b10a64542d77071bd64bd7b231f4ed5940fba55e98c3de13824cf3d7"
[[package]]
name = "syn"
version = "2.0.48"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f3531638e407dfc0814761abb7c00a5b54992b849452a0646b7f65c9f770f3f"
dependencies = [
"proc-macro2",
"quote",
"unicode-ident",
]
[[package]]
name = "target-lexicon"
version = "0.12.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69758bda2e78f098e4ccb393021a0963bb3442eac05f135c30f61b7370bbafae"
[[package]]
name = "unicode-ident"
version = "1.0.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3354b9ac3fae1ff6755cb6db53683adb661634f67557942dea4facebec0fee4b"
[[package]]
name = "unindent"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7de7d73e1754487cb58364ee906a499937a0dfabd86bcb980fa99ec8c8fa2ce"
[[package]]
name = "windows-targets"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
dependencies = [
"windows_aarch64_gnullvm",
"windows_aarch64_msvc",
"windows_i686_gnu",
"windows_i686_msvc",
"windows_x86_64_gnu",
"windows_x86_64_gnullvm",
"windows_x86_64_msvc",
]
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
[[package]]
name = "windows_aarch64_msvc"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
[[package]]
name = "windows_i686_gnu"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
[[package]]
name = "windows_i686_msvc"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
[[package]]
name = "windows_x86_64_gnu"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
[[package]]
name = "windows_x86_64_msvc"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"

17
rs/cidr/Cargo.toml Normal file
View file

@ -0,0 +1,17 @@
[package]
name = "rs_cidr_verifier"
version = "0.1.0"
edition = "2021"
[profile.release]
opt-level = 3
lto = true
[lib]
name = "rs_cidr_verifier"
path = "src/lib.rs"
crate-type = ["cdylib"]
[dependencies]
pyo3 = { version = "0.20.2", features = ["extension-module"] }
ipnetwork = "0.20.0"

2
rs/cidr/MANIFEST.in Normal file
View file

@ -0,0 +1,2 @@
include Cargo.toml
recursive-include src *

34
rs/cidr/src/lib.rs Normal file
View file

@ -0,0 +1,34 @@
use pyo3::prelude::*;
use ipnetwork::{Ipv4Network, Ipv6Network};
use std::net::{Ipv4Addr, Ipv6Addr};
use std::str::FromStr;
#[pyfunction]
fn is_ipv4_in_cidr(ip: &str, cidr: &str) -> PyResult<bool> {
let ip_addr = ip.parse::<Ipv4Addr>()
.map_err(|_| PyErr::new::<pyo3::exceptions::PyValueError, _>("Invalid IPv4 address"))?;
let cidr_net = Ipv4Network::from_str(cidr)
.map_err(|_| PyErr::new::<pyo3::exceptions::PyValueError, _>("Invalid IPv4 CIDR notation"))?;
Ok(cidr_net.contains(ip_addr))
}
#[pyfunction]
fn is_ipv6_in_cidr(ip: &str, cidr: &str) -> PyResult<bool> {
let ip_addr = ip.parse::<Ipv6Addr>()
.map_err(|_| PyErr::new::<pyo3::exceptions::PyValueError, _>("Invalid IPv6 address"))?;
let cidr_net = Ipv6Network::from_str(cidr)
.map_err(|_| PyErr::new::<pyo3::exceptions::PyValueError, _>("Invalid IPv6 CIDR notation"))?;
Ok(cidr_net.contains(ip_addr))
}
#[pymodule]
fn piracyshield_component_cidr_verifier(_py: Python, m: &PyModule) -> PyResult<()> {
m.add_function(wrap_pyfunction!(is_ipv4_in_cidr, m)?)?;
m.add_function(wrap_pyfunction!(is_ipv6_in_cidr, m)?)?;
Ok(())
}

20
src/piracyshield_component/validation/rules/cidr_syntax_ipv6.py
View file

@ -1,6 +1,6 @@
from piracyshield_component.validation.rule import Rule from piracyshield_component.validation.rule import Rule
import ipaddress from ipaddress import ip_network, AddressValueError, NetmaskValueError
class CIDRSyntaxIPv6(Rule): class CIDRSyntaxIPv6(Rule):
@ -14,33 +14,33 @@ class CIDRSyntaxIPv6(Rule):
""" """
Initialize parent __init__. Initialize parent __init__.
""" """
super().__init__() super().__init__()
def __call__(self, value: str) -> None: def __call__(self, value: str) -> bool:
""" """
Checks the validity of the passed string. Checks the validity of the passed string.
:param value: a valid CIDR syntax string. :param value: a valid CIDR syntax string.
""" """
try: try:
# doesn't seem solid enough, but we're not too paranoid
if '/' not in value: if '/' not in value:
self.register_error(self.message) self.register_error(self.message)
return False return False
network = ipaddress.ip_network(value, strict = True) network = ip_network(value, strict=False)
# check for a single IPv6 address # Check for valid prefix length
if network.prefixlen > 128: if not (1 <= network.prefixlen <= 128):
self.register_error(self.message) self.register_error(self.message)
return False return False
# non valid at all # If all checks pass
except ValueError: return True
except (ValueError, AddressValueError, NetmaskValueError):
self.register_error(self.message) self.register_error(self.message)
return False return False

2
src/piracyshield_component/validation/rules/dda.py
View file

@ -10,7 +10,7 @@ class DDA(Rule):
message = 'DDA identifier not valid' message = 'DDA identifier not valid'
expression = r'^[0-9]{3}\/[0-9]{2}\/DDA$' expression = r'^[0-9]{1,4}\/[0-9]{2}\/DDA$'
def __init__(self): def __init__(self):
""" """

12
src/piracyshield_component/validation/rules/ipv4.py
View file

@ -38,6 +38,8 @@ class IPv4(Rule):
if octets_size != 4: if octets_size != 4:
self.register_error(self.octets_message.format(octets_size)) self.register_error(self.octets_message.format(octets_size))
return False
for octet in octets: for octet in octets:
single_octet_size = len(octet) single_octet_size = len(octet)
@ -45,10 +47,18 @@ class IPv4(Rule):
if not octet.isdigit(): if not octet.isdigit():
self.register_error(self.octets_digits_message) self.register_error(self.octets_digits_message)
return False
# with a maximum length of 3 # with a maximum length of 3
if single_octet_size > 3: if single_octet_size > 3:
self.register_error(self.octets_length_message) self.register_error(self.octets_length_message)
return False
int_octet = int(octet)
# between 0 and 255 # between 0 and 255
if single_octet_size < 0 or single_octet_size > 255: if int_octet < 0 or int_octet > 255:
self.register_error(self.octets_digits_size_message) self.register_error(self.octets_digits_size_message)
return False

52
src/piracyshield_component/validation/rules/ipv6.py
View file

@ -6,10 +6,14 @@ class IPv6(Rule):
Rule that checks for a valid IPv6. Rule that checks for a valid IPv6.
""" """
hextets_syntax_message = 'IPv6 not valid, more than one `::` found'
hextets_message = 'IPv6 not valid, expecting eight hextets, got {}' hextets_message = 'IPv6 not valid, expecting eight hextets, got {}'
hextets_digits_message = 'IPv6 not valid, expecting eight hextets of hexadecimal digits' hextets_digits_message = 'IPv6 not valid, expecting eight hextets of hexadecimal digits'
hextets_maximum_length = 'IPv6 not valid, too many hextets'
hextets_length_message = 'IPv6 not valid, one or more hextet(s) too long' hextets_length_message = 'IPv6 not valid, one or more hextet(s) too long'
hextets_digits_size_message = 'IPv6 not valid, expecting hexadecimal digits from 0 to FFFF' hextets_digits_size_message = 'IPv6 not valid, expecting hexadecimal digits from 0 to FFFF'
@ -29,32 +33,56 @@ class IPv6(Rule):
:param value: a valid string. :param value: a valid string.
""" """
hextets = value.split(':') # short syntax
if '::' in value:
parts = value.split('::')
hextets_size = len(hextets) if len(parts) > 2:
self.register_error(self.hextets_syntax_message)
# we're expecting 8 hextets return False
if hextets_size != 8:
self.register_error(self.hextets_message.format(hextets_size)) left_side = parts[0].split(':') if parts[0] else []
right_side = parts[1].split(':') if parts[1] else []
zeros_needed = 8 - len(left_side) - len(right_side)
if zeros_needed < 0:
self.register_error(self.hextets_maximum_length)
return False
hextets = left_side + ['0'] * zeros_needed + right_side
# common syntax
else:
hextets = value.split(':')
if len(hextets) != 8:
self.register_error(self.hextets_message.format(self.hextets_digits_message))
return False
for hextet in hextets: for hextet in hextets:
single_hextet_size = len(hextet)
# each hextet must be composed of hexadecimal digits
if not all(c in '0123456789ABCDEFabcdef' for c in hextet): if not all(c in '0123456789ABCDEFabcdef' for c in hextet):
self.register_error(self.hextets_digits_message) self.register_error(self.hextets_digits_message)
# with a maximum length of 4 return False
if single_hextet_size > 4:
if len(hextet) > 4:
self.register_error(self.hextets_length_message) self.register_error(self.hextets_length_message)
return False
try: try:
# convert the hextet to an integer in base 16
int_value = int(hextet, 16) int_value = int(hextet, 16)
# check if the integer value is within the valid range (0~0xFFFF)
if not (0 <= int_value <= 0xFFFF): if not (0 <= int_value <= 0xFFFF):
self.register_error(self.hextets_digits_size_message) self.register_error(self.hextets_digits_size_message)
return False
except ValueError: except ValueError:
self.register_error(self.hextets_digits_size_message) self.register_error(self.hextets_digits_size_message)
return False