from typing import Optional
import regex as re
from ..base_op import OPERATORS, Mapper
[docs]
@OPERATORS.register_module("clean_ip_mapper")
class CleanIpMapper(Mapper):
"""Cleans IPv4 and IPv6 addresses from text samples.
This operator removes or replaces IPv4 and IPv6 addresses in the text. It uses a regular
expression to identify and clean the IP addresses. By default, it replaces the IP
addresses with an empty string, effectively removing them. The operator can be
configured with a custom pattern and replacement string. If no pattern is provided, a
default pattern for both IPv4 and IPv6 addresses is used. The operator processes samples
in batches.
- Uses a regular expression to find and clean IP addresses.
- Replaces found IP addresses with a specified replacement string.
- Default replacement string is an empty string, which removes the IP addresses.
- Can use a custom regular expression pattern if provided.
- Processes samples in batches for efficiency."""
_batched_op = True
[docs]
def __init__(self, pattern: Optional[str] = None, repl: str = "", *args, **kwargs):
"""
Initialization method.
:param pattern: regular expression pattern to search for within text.
:param repl: replacement string, default is empty string.
:param args: extra args
:param kwargs: extra args
"""
super().__init__(*args, **kwargs)
if pattern is None:
ipv4 = r"(?:(?:1[0-9][0-9]\.)|(?:2[0-4][0-9]\.)|"
ipv4 += r"(?:25[0-5]\.)|(?:[1-9][0-9]\.)|(?:[0-9]\.))"
ipv4 += r"{3}(?:(?:1[0-9][0-9])|(?:2[0-4][0-9])|"
ipv4 += r"(?:25[0-5])|(?:[1-9][0-9])|(?:[0-9]))"
self.pattern = ipv4 + "|"
self.pattern += r"([\da-fA-F]{1,4}:){7}[\da-fA-F]{1,4}" # ipv6
# compressed ipv6 ("::"), not touching other ASCII word characters
# or colons, so "std::cout" and partial matches are left alone
hextet = r"[\da-fA-F]{1,4}"
compressed = "|".join(
[
rf"(?:{hextet}:){{1,7}}:",
rf"(?:{hextet}:){{1,6}}:{hextet}",
rf"(?:{hextet}:){{1,5}}(?::{hextet}){{1,2}}",
rf"(?:{hextet}:){{1,4}}(?::{hextet}){{1,3}}",
rf"(?:{hextet}:){{1,3}}(?::{hextet}){{1,4}}",
rf"(?:{hextet}:){{1,2}}(?::{hextet}){{1,5}}",
rf"{hextet}:(?::{hextet}){{1,6}}",
rf":(?::{hextet}){{1,7}}",
]
)
# ipv6 ending in an ipv4 address (::ffff:192.0.2.128, 64:ff9b::198.51.100.7),
# matched whole; no match may end right before a dotted ipv4 part
embedded = "|".join(
[rf"(?:{hextet}:){{6}}", rf"::(?:{hextet}:){{0,5}}"]
+ [rf"(?:{hextet}:){{{n}}}:(?:{hextet}:){{0,{5 - n}}}" for n in range(1, 6)]
)
# the first lookahead only skips quickly past text where no address can start
self.pattern += rf"|(?<![0-9A-Za-z_:])(?=[\da-fA-F]{{0,4}}:)"
self.pattern += rf"(?:(?:{embedded}){ipv4}|{compressed})(?![0-9A-Za-z_:]|\.\d)"
else:
self.pattern = pattern
if (len(pattern) > 2) and (
pattern.startswith("r'") and pattern.endswith("'") or pattern.startswith('r"') and pattern.endswith('"')
):
self.pattern = pattern[2:-1]
self.repl = repl
[docs]
def process_batched(self, samples):
for idx, text in enumerate(samples[self.text_key]):
if not re.search(self.pattern, text, flags=re.DOTALL):
continue
samples[self.text_key][idx] = re.sub(pattern=self.pattern, repl=self.repl, string=text, flags=re.DOTALL)
return samples
