import ipaddress import re from urllib.parse import urlsplit, urlunsplit import validators from shared.enums.target import TargetType HEX_COLOR_LENGTH = 7 # #RRGGBB MAX_NAME_LEN = 120 def clean_name(v: str, *, max_len: int = MAX_NAME_LEN) -> str: v = (v or "").strip() if not v: msg = "Name must not be empty." raise ValueError(msg) if len(v) > max_len: msg = f"Name must be {max_len} characters or fewer." raise ValueError(msg) return v def clean_optional_name(v: str | None, *, max_len: int = MAX_NAME_LEN) -> str | None: """An update that omits the name leaves it alone.""" return None if v is None else clean_name(v, max_len=max_len) def validate_domain(value: str) -> bool: return validators.domain(value) is True def validate_ip(value: str) -> bool: try: ipaddress.ip_address(value) return True except ValueError: return False def validate_ip_range(value: str) -> bool: try: ipaddress.ip_network(value, strict=False) return "/" in value except ValueError: return False MAX_ASN = 4294967295 _ASN_SHAPE = re.compile(r"^AS(\d+)$") def validate_asn(value: str) -> bool: match = _ASN_SHAPE.match(value.upper()) return bool(match) and 0 < int(match.group(1)) <= MAX_ASN WEB_SCHEMES = ("http://", "https://") _SCHEME = re.compile(r"^[A-Za-z][A-Za-z0-9+.\-]*://") def validate_url(value: str) -> bool: return value.lower().startswith(WEB_SCHEMES) and validators.url(value) is True def _lower_host(netloc: str) -> str: """Case-fold the host, leaving any userinfo as written.""" userinfo, sep, host = netloc.rpartition("@") return f"{userinfo}{sep}{host.lower()}" def _normalize_url(value: str) -> str: try: parts = urlsplit(value) except ValueError: return value return urlunsplit( parts._replace(scheme=parts.scheme.lower(), netloc=_lower_host(parts.netloc)) ) def normalize_target_value(value: str) -> str: """Normalized target value. Host names are case-folded so one host is one target.""" v = (value or "").strip() if not v: return v if _SCHEME.match(v): return _normalize_url(v) v = v.removeprefix("*.") return v.rstrip("/").rstrip(".").lower() TARGET_FORMAT_HINT = "Enter a domain, IP address, CIDR range, URL or ASN." _MAX_ECHO = 120 def unrecognised_target(value: str) -> str: """Wording for a target value the scanner will not take.""" shown = (value or "").strip()[:_MAX_ECHO] return _refusal(normalize_target_value(shown), shown) or ( f"Unrecognised target: {shown}. {TARGET_FORMAT_HINT}" ) def _refusal(value: str, shown: str) -> str | None: """The reason, when the value has the right shape but is out of bounds.""" if _ASN_SHAPE.match(value.upper()): return f"{shown} is out of range. An AS number runs from 1 to {MAX_ASN}." try: block = ipaddress.ip_network(value, strict=False) except ValueError: return None if block.prefixlen == 0: return f"{shown} covers every address. Enter a narrower range." return ( f"{shown} is out of scope. Loopback, link-local, multicast and reserved " "address space are not scanned." ) def _routable(address) -> bool: return not ( address.is_unspecified or address.is_loopback or address.is_link_local or address.is_multicast or address.is_reserved ) def scannable_address(value: str) -> bool: """Address space a scan may be aimed at. Private ranges stay in for corporate estates.""" try: return _routable(ipaddress.ip_address(value)) except ValueError: return False def _scannable(value: str, target_type: TargetType) -> bool: """Address space a scan may be aimed at. Private ranges stay in for corporate estates.""" if target_type is TargetType.IP_RANGE: block = ipaddress.ip_network(value, strict=False) return block.prefixlen > 0 and _routable(block.network_address) if target_type is TargetType.IP: return scannable_address(value) return True def validate_target(target_value: str) -> TargetType | None: value = normalize_target_value(target_value) if not value: return None _validators = [ (TargetType.IP_RANGE, validate_ip_range), (TargetType.IP, validate_ip), (TargetType.URL, validate_url), (TargetType.ASN, validate_asn), (TargetType.DOMAIN, validate_domain), ] for target_type, validator_func in _validators: if validator_func(value): return target_type if _scannable(value, target_type) else None return None def normalize_query(query: str, target_type: TargetType) -> str: match target_type: case TargetType.DOMAIN: return normalize_domain(query) case TargetType.URL: return normalize_domain(extract_domain_from_url(query)) case TargetType.ASN: return str(extract_asn_number(query)) case _: return query.strip() def normalize_domain(domain: str) -> str: domain = domain.strip().lower() domain = re.sub(r"^https?://", "", domain) domain = domain.split("/")[0].split("?")[0].split("#")[0].split(":")[0] return domain.rstrip(".") def extract_asn_number(query: str) -> int: cleaned = re.sub(r"^[Aa][Ss]", "", query.strip()) return int(cleaned) def extract_domain_from_url(url: str) -> str: return normalize_domain(url) def validate_hex_color(color: str) -> str: color = color.strip() if not color.startswith("#"): msg = "Color must start with #" raise ValueError(msg) if len(color) != HEX_COLOR_LENGTH: msg = f"Color must be {HEX_COLOR_LENGTH} characters in the form #RRGGBB" raise ValueError(msg) hex_pattern = r"^#[0-9A-Fa-f]{6}$" if not re.match(hex_pattern, color): msg = "Color must use hexadecimal digits 0-9 and A-F" raise ValueError(msg) return color.upper() _HOST_RE = re.compile( r"^(?=.{1,253}$)([a-z0-9_](?:[a-z0-9_-]{0,62}[a-z0-9_])?\.)+[a-z0-9][a-z0-9-]{0,62}$" ) def normalize_host(raw: str) -> str | None: """A bare host name, wildcard label and trailing dot stripped. None when it is not one.""" if not raw: return None name = raw.strip().lower().rstrip(".") if name.startswith("*."): name = name[2:] if not name or "." not in name or " " in name or "/" in name or "@" in name: return None if not _HOST_RE.match(name): return None return name