Fix IP geolocation service: change to HTTPS and add IP validation

parent feafe0db
import asyncio import asyncio
import httpx import httpx
import ipaddress
from typing import Dict, Optional from typing import Dict, Optional
# Global cache for IP -> country mappings # Global cache for IP -> country mappings
...@@ -7,15 +8,22 @@ _ip_country_cache: Dict[str, Optional[str]] = {} ...@@ -7,15 +8,22 @@ _ip_country_cache: Dict[str, Optional[str]] = {}
async def get_ip_country(ip: str) -> Optional[str]: async def get_ip_country(ip: str) -> Optional[str]:
"""Get country code for IP address using ipapi.co, with caching. """Get country code for IP address using ipapi.co, with caching.
Returns country code (e.g., 'IL') or None if failed. Returns country code (e.g., 'IL') or None if failed.
""" """
# Validate IP address format
try:
ipaddress.ip_address(ip)
except ValueError:
_ip_country_cache[ip] = None
return None
if ip in _ip_country_cache: if ip in _ip_country_cache:
return _ip_country_cache[ip] return _ip_country_cache[ip]
try: try:
async with httpx.AsyncClient(timeout=5.0) as client: async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(f"http://ipapi.co/{ip}/country/") response = await client.get(f"https://ipapi.co/{ip}/country/")
if response.status_code == 200: if response.status_code == 200:
country = response.text.strip().upper() country = response.text.strip().upper()
_ip_country_cache[ip] = country _ip_country_cache[ip] = country
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment