fix: improve IP classification and source fallback

vpn_utils.py

@@ -15,11 +15,18 @@ from typing import Any
 ROOT_DIR = Path(__file__).resolve().parent
 DATA_DIR = Path(os.environ["VPNGATE_DATA_DIR"]).resolve() if os.environ.get("VPNGATE_DATA_DIR") else ROOT_DIR / "vpngate_data"
 IP_CACHE_FILE = DATA_DIR / "ip_cache.json"
+IP_CLASSIFICATION_VERSION = 2
+IP_CACHE_TTL_SECONDS = 7 * 24 * 3600
 
 ip_cache_lock = threading.RLock()
 physical_interface_lock = threading.Lock()
 physical_interface_cache: tuple[str | None, float] = (None, 0.0)
 
+DATACENTER_PROVIDER_PATTERN = re.compile(
+    r"(?:\b(?:cloud|colo|colocation|data[ -]?center|hosting|servers?|vps)\b|softether)",
+    re.IGNORECASE,
+)
+
 COUNTRY_TRANSLATIONS = {
     "Japan": "日本",
     "Korea Republic of": "韩国",
@@ -389,6 +396,39 @@ def save_ip_cache(cache: dict[str, dict[str, Any]]) -> None:
         except Exception:
             pass
 
+def classify_ip_type(item: dict[str, Any]) -> tuple[str, str]:
+    """Classify network ownership without confusing VPN use with hosting."""
+    if item.get("mobile"):
+        return "mobile", "mobile_flag"
+    if item.get("hosting"):
+        return "hosting", "hosting_flag"
+
+    provider_text = " ".join(
+        str(item.get(key) or "")
+        for key in ("isp", "org", "as", "asname")
+    )
+    if item.get("proxy") and DATACENTER_PROVIDER_PATTERN.search(provider_text):
+        return "hosting", "proxy_provider_datacenter"
+
+    # A residential volunteer running VPNGate is commonly marked as a proxy.
+    # Proxy use is retained in quality/is_proxy and must not change ownership.
+    return "residential", "consumer_or_unclassified_network"
+
+def apply_ip_cache_entry(node: dict[str, Any], entry: dict[str, Any]) -> None:
+    for key in (
+        "owner",
+        "asn",
+        "as_name",
+        "location",
+        "ip_type",
+        "quality",
+        "is_proxy",
+        "is_hosting",
+        "is_mobile",
+        "ip_type_reason",
+    ):
+        node[key] = entry.get(key, "")
+
 def enrich_ip_info(nodes: list[dict[str, Any]]) -> None:
     # 1. Read cache thread-safely
     with ip_cache_lock:
@@ -401,14 +441,14 @@ def enrich_ip_info(nodes: list[dict[str, Any]]) -> None:
         ip = node.get("ip") or node.get("remote_host")
         if not ip:
             continue
-        if ip in cache and now - cache[ip].get("cached_at", 0) < 7 * 24 * 3600:
+        cache_entry = cache.get(ip) if isinstance(cache.get(ip), dict) else None
+        if (
+            cache_entry
+            and cache_entry.get("classification_version") == IP_CLASSIFICATION_VERSION
+            and now - cache_entry.get("cached_at", 0) < IP_CACHE_TTL_SECONDS
+        ):
             cached = cache[ip]
-            node["owner"] = cached.get("owner", "")
-            node["asn"] = cached.get("asn", "")
-            node["as_name"] = cached.get("as_name", "")
-            node["location"] = cached.get("location", "")
-            node["ip_type"] = cached.get("ip_type", "")
-            node["quality"] = cached.get("quality", "")
+            apply_ip_cache_entry(node, cached)
         else:
             if ip not in ips_to_query:
                 ips_to_query.append(ip)
@@ -425,7 +465,10 @@ def enrich_ip_info(nodes: list[dict[str, Any]]) -> None:
         request = urllib.request.Request(
             "http://ip-api.com/batch?lang=zh-CN&fields=status,message,query,country,regionName,city,isp,org,as,asname,proxy,hosting,mobile",
             data=payload,
-            headers={"Content-Type": "application/json", "User-Agent": "vpngate-manager/2.2"},
+            headers={
+                "Content-Type": "application/json",
+                "User-Agent": f"AimiliVPN-IP-Classifier/{IP_CLASSIFICATION_VERSION}",
+            },
             method="POST",
         )
         try:
@@ -442,19 +485,15 @@ def enrich_ip_info(nodes: list[dict[str, Any]]) -> None:
                     if not query_ip:
                         continue
 
-                    ip_type = "residential"
-                    if item.get("mobile"):
-                        ip_type = "mobile"
-                    elif item.get("hosting") or item.get("proxy"):
-                        ip_type = "hosting"
+                    ip_type, ip_type_reason = classify_ip_type(item)
 
                     quality = "normal"
-                    if item.get("proxy"):
-                        quality = "proxy"
+                    if item.get("mobile"):
+                        quality = "mobile"
                     elif item.get("hosting"):
                         quality = "datacenter"
-                    elif item.get("mobile"):
-                        quality = "mobile"
+                    elif item.get("proxy"):
+                        quality = "proxy"
 
                     loc = " ".join(part for part in [item.get("country"), item.get("regionName"), item.get("city")] if part)
 
@@ -465,6 +504,11 @@ def enrich_ip_info(nodes: list[dict[str, Any]]) -> None:
                         "location": loc,
                         "ip_type": ip_type,
                         "quality": quality,
+                        "is_proxy": bool(item.get("proxy")),
+                        "is_hosting": bool(item.get("hosting")),
+                        "is_mobile": bool(item.get("mobile")),
+                        "ip_type_reason": ip_type_reason,
+                        "classification_version": IP_CLASSIFICATION_VERSION,
                         "cached_at": now,
                     }
         except Exception as e:
@@ -484,12 +528,7 @@ def enrich_ip_info(nodes: list[dict[str, Any]]) -> None:
         ip = node.get("ip") or node.get("remote_host")
         if ip in new_entries:
             cached = new_entries[ip]
-            node["owner"] = cached.get("owner", "")
-            node["asn"] = cached.get("asn", "")
-            node["as_name"] = cached.get("as_name", "")
-            node["location"] = cached.get("location", "")
-            node["ip_type"] = cached.get("ip_type", "")
-            node["quality"] = cached.get("quality", "")
+            apply_ip_cache_entry(node, cached)
 
 
 def diagnose_api_failure(api_url: str = "https://www.vpngate.net/api/iphone/") -> tuple[int, str]: