""" URL features for the CyberForge detection models. One function, `extract_url_features`, is used both when the models are trained (trainer.py) and when they answer (app.py), so the two cannot drift apart. It is plain string processing: no ML library, no network, and it never raises. Two groups of features come back in one dictionary: * the twelve original names (url_length, is_https, ...), kept because API responses and the backend's risk factors use them; * scheme-free features, which the models are trained on. Public URL collections mostly list benign URLs without "http://" and malicious ones with it, so anything computed from the scheme would teach a model the way the list was written instead of anything about the URL. The models therefore see the URL with its scheme removed. Whether a page uses HTTPS is still reported, as a risk factor, not as model input. """ import math import re from collections import Counter from typing import Dict, List, Optional from urllib.parse import unquote, urlparse SUSPICIOUS_TLDS = (".xyz", ".tk", ".ml", ".ga", ".cf", ".top", ".buzz") # The twelve features the first models used. Still returned for compatibility. LEGACY_FEATURES = [ "url_length", "hostname_length", "path_length", "is_https", "has_ip_address", "has_suspicious_tld", "subdomain_count", "has_port", "query_params_count", "has_at_symbol", "has_double_slash", "special_char_count", ] # What a URL's host and path look like. Used by the phishing, malware and # anomaly models. LEXICAL_FEATURES = [ "bare_length", "host_length", "bare_path_length", "query_length", "host_dots", "host_subdomains", "host_hyphens", "host_digits", "host_digit_ratio", "host_entropy", "host_longest_label", "host_vowel_ratio", "host_is_ip", "host_has_port", "host_suspicious_tld", "tld_length", "has_punycode", "path_depth", "path_digits", "path_entropy", "path_longest_token", "query_params", "query_longest_value", "bare_at_symbol", "path_double_slash", "count_percent", "count_equals", "count_underscore", "count_hyphen_path", "count_tilde", "bare_special_chars", "has_executable_extension", "has_script_extension", "has_archive_extension", "credential_words", "is_shortener", "has_embedded_url", "domain_popularity", "is_user_content_host", ] # What an injected payload looks like, read from the decoded path and query. # Used by the web-attack model, together with a few lexical features. PAYLOAD_FEATURES = [ "payload_length", "payload_quotes", "payload_angle_brackets", "payload_parentheses", "payload_semicolons", "payload_spaces", "payload_sql_keywords", "payload_script_tokens", "payload_comment_markers", "payload_traversal", "payload_pipes_backticks", "payload_encoded_specials", "payload_special_ratio", "payload_longest_run_specials", "payload_equals_in_value", "payload_null_bytes", ] ATTACK_FEATURES = PAYLOAD_FEATURES + [ "bare_path_length", "query_length", "query_params", "count_percent", "path_depth", "query_longest_value", ] ALL_FEATURES = list(dict.fromkeys(LEGACY_FEATURES + LEXICAL_FEATURES + PAYLOAD_FEATURES)) EXECUTABLE_EXTENSIONS = (".exe", ".dll", ".scr", ".msi", ".bat", ".cmd", ".com", ".pif", ".apk", ".jar", ".bin", ".elf", ".dmg", ".iso") SCRIPT_EXTENSIONS = (".sh", ".ps1", ".vbs", ".js", ".hta", ".py", ".pl", ".wsf", ".lnk") ARCHIVE_EXTENSIONS = (".zip", ".rar", ".7z", ".gz", ".tar", ".tgz", ".cab", ".img") CREDENTIAL_WORDS = ("login", "signin", "sign-in", "verify", "account", "update", "secure", "banking", "confirm", "password", "webscr", "wallet", "recover", "unlock", "suspend", "billing") SHORTENERS = {"bit.ly", "tinyurl.com", "goo.gl", "t.co", "ow.ly", "is.gd", "buff.ly", "cutt.ly", "rebrand.ly", "tiny.cc", "shorturl.at", "rb.gy", "t.ly", "s.id", "lnkd.in"} SQL_KEYWORDS = re.compile( r"\b(select|union|insert|update|delete|drop|from|where|having|sleep|benchmark|waitfor|" r"information_schema|load_file|outfile|concat|char|cast|exec|xp_cmdshell|order\s+by|group\s+by|or\s+1\s*=\s*1|and\s+1\s*=\s*1)\b", re.IGNORECASE) SCRIPT_TOKENS = re.compile( r"(<\s*script|<\s*img|<\s*svg|<\s*iframe|<\s*body|<\s*object|<\s*embed|javascript:|vbscript:|data:text/html|" r"on(?:error|load|click|mouseover|focus|toggle|start)\s*=|alert\s*\(|prompt\s*\(|confirm\s*\(|eval\s*\(|" r"document\.(?:cookie|domain|write)|fromcharcode|expression\s*\()", re.IGNORECASE) COMMENT_MARKERS = re.compile(r"(--|/\*|\*/|#\s*$|;\s*--)") ENCODED_SPECIALS = re.compile(r"%(?:27|22|3c|3e|28|29|3b|2d%2d|00|0a|0d|7c|60|5c)", re.IGNORECASE) SPECIAL_RUN = re.compile(r"[^A-Za-z0-9\s/._~-]{2,}") LEGACY_SPECIALS = "!@#$%^&*()+={}[]|\\:;<>?," SCHEME = re.compile(r"^[a-zA-Z][a-zA-Z0-9+.-]*://") EMBEDDED_URL = re.compile(r"(https?:|https?%3a|www\.)", re.IGNORECASE) # --------------------------------------------------------------------------- # Domain reputation # --------------------------------------------------------------------------- # The text of a URL cannot tell paypal.com/signin from a copy of it: a real # sign-in page and a fake one look alike. What separates them is who the site # is. `domain_popularity` says how high the URL's registered domain ranks in a # public list of the most visited sites (Tranco), from 1.0 for the top of the # list to 0.0 for a domain that is not on it. Free hosting and user-content # platforms are on that list too, but a page on one of them is its author's, # not the platform's, so those hosts count as unknown and are marked. USER_CONTENT_HOSTS = ( "github.io", "gitlab.io", "blogspot.com", "wordpress.com", "weebly.com", "wixsite.com", "wix.com", "000webhostapp.com", "firebaseapp.com", "web.app", "pages.dev", "vercel.app", "netlify.app", "herokuapp.com", "glitch.me", "repl.co", "replit.app", "godaddysites.com", "square.site", "webflow.io", "myshopify.com", "tumblr.com", "livejournal.com", "medium.com", "notion.site", "sites.google.com", "storage.googleapis.com", "firebasestorage.googleapis.com", "appspot.com", "azurewebsites.net", "blob.core.windows.net", "amazonaws.com", "cloudfront.net", "r2.dev", "workers.dev", "duckdns.org", "ngrok.io", "ngrok-free.app", "trycloudflare.com", "surge.sh", "hf.space", "onrender.com", "fly.dev", "start.page", "carrd.co", "linktr.ee", "yolasite.com", ) SECOND_LEVEL_SUFFIXES = {"co", "com", "org", "net", "gov", "edu", "ac"} _popular: Dict[str, int] = {} def build_popularity(domains: List[str]) -> Dict[str, int]: """{domain: rank} from a list in rank order (most visited first).""" table: Dict[str, int] = {} for rank, domain in enumerate(domains, start=1): domain = str(domain).strip().lower() if domain and domain not in table: table[domain] = rank return table def set_popular_domains(domains: List[str]) -> None: """Install the list the service scores URLs with.""" global _popular _popular = build_popularity(domains) def popular_domains_loaded() -> int: return len(_popular) def is_user_content_host(host: str) -> int: host = host.lower() return 1 if any(host == platform or host.endswith("." + platform) for platform in USER_CONTENT_HOSTS) else 0 def domain_popularity(host: str, popular: Optional[Dict[str, int]] = None) -> float: """0.0 for a domain that is not on the list (or a user-content host); up to 1.0 for the top of the list. The most specific listed name wins, and a bare second-level suffix such as "co.uk" never counts.""" table = _popular if popular is None else popular host = host.lower().strip(".") if not table or not host or is_user_content_host(host): return 0.0 labels = host.split(".") for start in range(0, len(labels) - 1): candidate = ".".join(labels[start:]) parts = labels[start:] if len(parts) == 2 and parts[0] in SECOND_LEVEL_SUFFIXES and len(parts[1]) == 2: continue rank = table.get(candidate) if rank: return round(max(0.05, 1.0 - math.log10(rank) / 6.0), 4) return 0.0 def rescale_score(probability: float, threshold: float) -> float: """Map a model's raw probability so that its decision threshold lands on 0.5, keeping the order of scores. Every model in the service is read the same way (0.5 or more is a threat) whatever threshold it was given when it was trained.""" if not 0.0 < threshold < 1.0: return float(probability) if probability < threshold: return 0.5 * probability / threshold return 0.5 + 0.5 * (probability - threshold) / (1.0 - threshold) def split_url(url: str) -> Dict: """The parts of a URL that the features are built from. Never raises: text that is not a well-formed URL (an unclosed "[" in the host, a port that is not a number) still yields whatever could be read, because callers also score URLs picked out of free text.""" try: parsed = urlparse(url) parts = { "scheme": parsed.scheme, "hostname": parsed.hostname or "", "path": parsed.path or "", "query": parsed.query or "", } except ValueError: scheme = url.split("://", 1)[0].lower() if "://" in url else "" return {"scheme": scheme, "hostname": "", "path": "", "query": "", "has_port": 0} try: parts["has_port"] = 1 if parsed.port else 0 except ValueError: parts["has_port"] = 1 # a port is given, but it is not a valid number return parts def bare_url(url: str) -> str: """The URL as the models see it: no scheme, no leading "www.", no surrounding space. Like the scheme, "www." says how a list was written, not what the site is: one collection writes every benign URL without it and another with it.""" bare = SCHEME.sub("", str(url).strip(), count=1) return bare[4:] if bare[:4].lower() == "www." and len(bare) > 4 else bare def _entropy(text: str) -> float: if not text: return 0.0 total = len(text) return round(-sum((n / total) * math.log2(n / total) for n in Counter(text).values()), 4) def _is_ip(host: str) -> int: labels = host.split(".") if len(labels) == 4 and all(label.isdigit() for label in labels): return 1 return 1 if ":" in host and all(c in "0123456789abcdefABCDEF:." for c in host) else 0 def registered_domain(url: str) -> str: """A rough registered domain, used only to keep one site's URLs together when data is split for testing. No public-suffix list: the last two labels, or three when the second-to-last is a common second-level suffix.""" host = split_url("//" + bare_url(url))["hostname"].lower() if not host or _is_ip(host): return host or "unknown" labels = host.split(".") if len(labels) >= 3 and labels[-2] in {"co", "com", "org", "net", "gov", "edu", "ac"} and len(labels[-1]) == 2: return ".".join(labels[-3:]) return ".".join(labels[-2:]) def extract_url_features(url: str, popular: Optional[Dict[str, int]] = None) -> Dict: """Every feature for one URL: the twelve original names and the scheme-free ones. All values are numbers. `popular` is a {domain: rank} table to use instead of the installed one (training passes its own).""" url = "" if url is None else str(url) parts = split_url(url) hostname = parts["hostname"] # --- the twelve original features, computed exactly as before --- features = { "url_length": len(url), "hostname_length": len(hostname), "path_length": len(parts["path"]), "is_https": 1 if parts["scheme"] == "https" else 0, "has_ip_address": 1 if all(p.isdigit() for p in hostname.split(".")) and len(hostname.split(".")) == 4 else 0, "has_suspicious_tld": 1 if any(hostname.endswith(t) for t in SUSPICIOUS_TLDS) else 0, "subdomain_count": max(0, len(hostname.split(".")) - 2), "has_port": parts["has_port"], "query_params_count": len(parts["query"].split("&")) if parts["query"] else 0, "has_at_symbol": 1 if "@" in url else 0, "has_double_slash": 1 if "//" in parts["path"] else 0, "special_char_count": sum(1 for c in url if c in LEGACY_SPECIALS), } # --- scheme-free features: the URL is read again without its scheme, so a # URL listed as "example.com/x" and one listed as "http://example.com/x" # give the same values. They have their own names; the twelve above are # never overwritten --- bare = bare_url(url) seen = split_url("//" + bare) host = seen["hostname"].lower() path = seen["path"] query = seen["query"] labels = [label for label in host.split(".") if label] is_ip = _is_ip(host) tld = "" if is_ip or len(labels) < 2 else labels[-1] path_tokens = [token for token in re.split(r"[/._\-]", path) if token] values = [pair.split("=", 1)[1] if "=" in pair else pair for pair in query.split("&")] if query else [] lower_path = path.lower() lower_rest = (path + "?" + query).lower() if query else lower_path host_letters = [c for c in host if c.isalpha()] features.update({ "bare_length": len(bare), "host_length": len(host), "bare_path_length": len(path), "query_length": len(query), "host_dots": host.count("."), "host_subdomains": 0 if is_ip else max(0, len(labels) - 2), "host_hyphens": host.count("-"), "host_digits": sum(c.isdigit() for c in host), "host_digit_ratio": round(sum(c.isdigit() for c in host) / len(host), 4) if host else 0.0, "host_entropy": _entropy(host), "host_longest_label": max((len(label) for label in labels), default=0), "host_vowel_ratio": round(sum(c in "aeiou" for c in host_letters) / len(host_letters), 4) if host_letters else 0.0, "host_is_ip": is_ip, "host_has_port": seen["has_port"], "host_suspicious_tld": 1 if any(host.endswith(t) for t in SUSPICIOUS_TLDS) else 0, "tld_length": len(tld), "has_punycode": 1 if "xn--" in host else 0, "path_depth": len([segment for segment in path.split("/") if segment]), "path_digits": sum(c.isdigit() for c in path), "path_entropy": _entropy(path), "path_longest_token": max((len(token) for token in path_tokens), default=0), "query_params": len(query.split("&")) if query else 0, "query_longest_value": max((len(value) for value in values), default=0), "bare_at_symbol": 1 if "@" in bare else 0, "path_double_slash": 1 if "//" in path else 0, "count_percent": bare.count("%"), "count_equals": bare.count("="), "count_underscore": bare.count("_"), "count_hyphen_path": path.count("-"), "count_tilde": bare.count("~"), "bare_special_chars": sum(1 for c in bare if c in LEGACY_SPECIALS), "has_executable_extension": 1 if lower_path.endswith(EXECUTABLE_EXTENSIONS) else 0, "has_script_extension": 1 if lower_path.endswith(SCRIPT_EXTENSIONS) else 0, "has_archive_extension": 1 if lower_path.endswith(ARCHIVE_EXTENSIONS) else 0, "credential_words": sum(1 for word in CREDENTIAL_WORDS if word in bare.lower()), "is_shortener": 1 if host in SHORTENERS else 0, "has_embedded_url": 1 if EMBEDDED_URL.search(lower_rest) else 0, "domain_popularity": 0.0 if is_ip else domain_popularity(host, popular), "is_user_content_host": is_user_content_host(host), }) # --- payload features: what is carried in the path and query, decoded once --- carried = path + ("?" + query if query else "") try: decoded = unquote(carried.replace("+", " ")) except Exception: decoded = carried specials = sum(1 for c in decoded if not (c.isalnum() or c.isspace() or c in "/._~-")) features.update({ "payload_length": len(decoded), "payload_quotes": decoded.count("'") + decoded.count('"') + decoded.count("`"), "payload_angle_brackets": decoded.count("<") + decoded.count(">"), "payload_parentheses": decoded.count("(") + decoded.count(")"), "payload_semicolons": decoded.count(";"), "payload_spaces": sum(c.isspace() for c in decoded), "payload_sql_keywords": len(SQL_KEYWORDS.findall(decoded)), "payload_script_tokens": len(SCRIPT_TOKENS.findall(decoded)), "payload_comment_markers": len(COMMENT_MARKERS.findall(decoded)), "payload_traversal": decoded.count("../") + decoded.count("..\\"), "payload_pipes_backticks": decoded.count("|") + decoded.count("`") + decoded.count("$("), "payload_encoded_specials": len(ENCODED_SPECIALS.findall(carried)), "payload_special_ratio": round(specials / len(decoded), 4) if decoded else 0.0, "payload_longest_run_specials": max((len(run) for run in SPECIAL_RUN.findall(decoded)), default=0), "payload_equals_in_value": sum(value.count("=") for value in values), "payload_null_bytes": carried.lower().count("%00") + decoded.count("\x00"), }) return features def feature_row(features: Dict, names: List[str]) -> List[float]: """The named features as a list of floats, in the order given.""" return [float(features.get(name, 0) or 0) for name in names]