phishing_detection: web_traffic feature from the NetAPI Top 1M instead of the dead Alexa endpoint

Cybersecurity_Tools/phishing_detection/features_extraction.py

@@ -1,5 +1,6 @@
 import re
 import socket
+import time
 import ssl
 import whois
 import requests
@@ -9,6 +10,38 @@
 import dns.resolver
 import tldextract
 
+# NetAPI domain-rank answers shared by all extractor instances: one request per registered domain.
+# The Top 1M is rebuilt daily, so an entry expires after six hours and a long-running process picks up new ranks.
+_RANK_CACHE = {}
+_RANK_CACHE_TTL = 6 * 3600
+
+
+def netapi_domain_rank(registered_domain):
+    """Rank of a registered domain in the NetAPI Top 1M, None when unranked or when the lookup failed.
+    Free calls without a key are limited to 30 a minute and 1,000 a day per IP: a per-minute 429 is waited
+    out once (Retry-After), the daily limit is not, and failed lookups are not cached."""
+    cached = _RANK_CACHE.get(registered_domain)
+    if cached is not None and time.monotonic() - cached[1] < _RANK_CACHE_TTL:
+        return cached[0]
+    for attempt in range(2):
+        response = requests.get(
+            "https://netapi.com/api-json/",
+            params={"method": "domain-rank", "domain": registered_domain},
+            timeout=10,
+        )
+        if response.status_code == 429 and attempt == 0:
+            retry_after = int(response.headers.get("Retry-After", "60"))
+            if retry_after <= 60:
+                time.sleep(retry_after)
+                continue
+        break
+    if response.status_code != 200:
+        return None
+    rank = response.json().get("rank")
+    _RANK_CACHE[registered_domain] = (rank, time.monotonic())
+    return rank
+
+
 class URLFeatureExtractor:
     def __init__(self, url):
         self.url = url
@@ -257,10 +290,20 @@ def check_dns_record(self):
             self.features['DNSRecord'] = -1
 
     def check_web_traffic(self):
-        """Check web traffic (simplified)"""
+        """Check web traffic: position of the domain in the NetAPI Top 1M (the Alexa ranking was retired in 2022).
+        Same scale as the training data: 1 = rank up to 100,000, 0 = ranked lower in the Top 1M, -1 = not ranked."""
         try:
-            response = requests.get(f"https://data.alexa.com/data?cli=10&url={self.domain}")
-            self.features['web_traffic'] = 1 if response.status_code == 200 else -1
+            # private suffixes (github.io, blogspot.com ...) keep the tenant name, so phish.github.io
+            # is looked up as itself and does not inherit the rank of the platform
+            ext = tldextract.extract(self.url, include_psl_private_domains=True)
+            registered_domain = f"{ext.domain}.{ext.suffix}" if ext.suffix else self.domain
+            rank = netapi_domain_rank(registered_domain)
+            if rank is None:
+                self.features['web_traffic'] = -1
+            elif rank <= 100000:
+                self.features['web_traffic'] = 1
+            else:
+                self.features['web_traffic'] = 0
         except:
             self.features['web_traffic'] = -1