"""Visitor tracking, IP blocking and live analytics services.""" from __future__ import annotations import logging from dataclasses import dataclass from datetime import timedelta from typing import Optional from django.conf import settings from django.core.cache import cache from django.db import DatabaseError from django.db.models import Count, F, Q from django.utils import timezone from cms.models import BlockedIP, VisitorPageView, VisitorSession from cms.visitor_intel import ( IPGeolocationService, get_client_ip, is_bot, is_private_ip, is_valid_ip, parse_user_agent, visitor_fingerprint, ) logger = logging.getLogger(__name__) _BLOCK_CACHE_KEY = 'ipblock:list:v1' # --------------------------------------------------------------------------- # Block list # --------------------------------------------------------------------------- @dataclass class BlockMatch: id: int ip_address: str reason: str source: str class BlocklistService: """Fast, cache-backed access to the :class:`BlockedIP` table.""" def __init__(self): self.cache_seconds = int(getattr(settings, 'IP_BLOCKLIST_CACHE_SECONDS', 30)) # -- reading ------------------------------------------------------------ def _load_blocks(self) -> list: now = timezone.now() try: rows = ( BlockedIP.objects.filter(is_active=True) .filter(Q(expires_at__isnull=True) | Q(expires_at__gt=now)) .values_list('id', 'ip_address', 'block_type', 'reason', 'source') ) blocks = list(rows) except DatabaseError: blocks = [] try: cache.set(_BLOCK_CACHE_KEY, blocks, self.cache_seconds) except Exception: # pragma: no cover pass return blocks def _get_blocks(self) -> list: blocks = cache.get(_BLOCK_CACHE_KEY) if blocks is None: blocks = self._load_blocks() return blocks def invalidate(self): try: cache.delete(_BLOCK_CACHE_KEY) except Exception: # pragma: no cover pass def is_blocked(self, ip_address: str) -> Optional[BlockMatch]: """Return a :class:`BlockMatch` if the IP is blocked, else ``None``.""" if not is_valid_ip(ip_address) or is_private_ip(ip_address): return None if self._is_whitelisted(ip_address): return None import ipaddress for block_id, addr, block_type, reason, source in self._get_blocks(): try: if block_type == 'cidr' or '/' in addr: network = ipaddress.ip_network(addr.strip(), strict=False) if ipaddress.ip_address(ip_address) in network: return BlockMatch(block_id, addr, reason, source) elif block_type == 'prefix': if ip_address.startswith(addr): return BlockMatch(block_id, addr, reason, source) elif ip_address == addr.strip(): return BlockMatch(block_id, addr, reason, source) except (ValueError, TypeError): continue return None def _is_whitelisted(self, ip_address: str) -> bool: whitelist = getattr(settings, 'IP_BLOCK_WHITELIST', []) or [] return ip_address in whitelist # -- writing ------------------------------------------------------------ def add_block( self, ip_address: str, *, reason: str = '', source: str = 'manual', notes: str = '', block_type: str = 'single', created_by=None, expires_at=None, ) -> Optional[BlockedIP]: ip_address = (ip_address or '').strip() if not ip_address: return None if block_type == 'single' and is_private_ip(ip_address): logger.info("Refusing to block private/reserved IP %s", ip_address) return None if block_type == 'single' and self._is_whitelisted(ip_address): logger.info("Refusing to block whitelisted IP %s", ip_address) return None block, created = BlockedIP.objects.update_or_create( ip_address=ip_address, defaults={ 'block_type': block_type, 'reason': reason[:255], 'source': source, 'notes': notes, 'is_active': True, 'expires_at': expires_at, 'created_by': created_by, }, ) self.invalidate() return block def remove_block(self, ip_address: str) -> int: deleted, _ = BlockedIP.objects.filter(ip_address=(ip_address or '').strip()).delete() self.invalidate() return deleted def record_hit(self, block_id: int, ip_address: str): """Increment the block's hit counter, throttled to at most once/minute.""" throttle_key = f'ipblock:hit:{block_id}' if cache.get(throttle_key): return cache.set(throttle_key, True, 60) try: BlockedIP.objects.filter(id=block_id).update( hit_count=F('hit_count') + 1, last_hit=timezone.now(), ) except Exception: # pragma: no cover pass # --------------------------------------------------------------------------- # Automatic blocking of abusive sources # --------------------------------------------------------------------------- class AutoBlockService: """Track repeat offenders and block them automatically.""" VIOLATION_KEY = 'ipviolation:{ip}' def register_violation(self, request, reason: str, source: str = 'auto') -> bool: """Record a rule violation for the request IP. Returns ``True`` when the IP was (or already is) blocked. """ if not getattr(settings, 'IP_AUTO_BLOCK_ENABLED', True): return False ip_address = get_client_ip(request) if not is_valid_ip(ip_address) or is_private_ip(ip_address): return False if BLOCKLIST.is_blocked(ip_address): return True window = int(getattr(settings, 'IP_AUTO_BLOCK_WINDOW_SECONDS', 3600)) threshold = int(getattr(settings, 'IP_AUTO_BLOCK_THRESHOLD', 3)) key = self.VIOLATION_KEY.format(ip=ip_address) violations = (cache.get(key) or 0) + 1 cache.set(key, violations, window) if violations >= threshold: duration_hours = int(getattr(settings, 'IP_AUTO_BLOCK_DURATION_HOURS', 0)) expires_at = ( timezone.now() + timedelta(hours=duration_hours) if duration_hours else None ) BLOCKLIST.add_block( ip_address, reason=f'Auto-blocked after {violations} violations: {reason}'[:255], source=source, block_type='single', expires_at=expires_at, ) cache.delete(key) logger.warning("Auto-blocked IP %s after %s violations (%s)", ip_address, violations, reason) return True return False # --------------------------------------------------------------------------- # Page view / session tracking # --------------------------------------------------------------------------- class VisitorTrackingService: """Persist visitor sessions and page views for the analytics dashboard.""" def __init__(self): self.enabled = bool(getattr(settings, 'VISITOR_TRACKING_ENABLED', True)) self.online_window = int(getattr(settings, 'VISITOR_ONLINE_WINDOW_SECONDS', 300)) self.touch_interval = int(getattr(settings, 'VISITOR_TOUCH_INTERVAL_SECONDS', 60)) self.pageview_interval = int(getattr(settings, 'VISITOR_PAGEVIEW_INTERVAL_SECONDS', 3)) self.track_pageviews = bool(getattr(settings, 'VISITOR_TRACK_PAGEVIEWS', True)) self.geo = IPGeolocationService() # -- helpers ------------------------------------------------------------ @staticmethod def should_track(request) -> bool: path = request.path or '' if request.method not in ('GET', 'HEAD'): return False if path.startswith(('/static/', '/media/', '/admin/', '/__')): return False if path in ('/robots.txt', '/sitemap.xml', '/llms.txt', '/favicon.ico'): return False if any(path.endswith(ext) for ext in ('.md', '.xml', '.txt', '.json', '.js', '.css', '.map')): return False return True @staticmethod def _region_code(path: str) -> str: for code in ('ca', 'uae', 'ireland'): if path.startswith(f'/{code}/') or path == f'/{code}': return code return 'uk' def _visitor_key(self, request, ip_address: str, user_agent: str, bot: bool) -> str: session_key = '' if not bot: session_key = getattr(request.session, 'session_key', None) or '' if not session_key: try: request.session.save() session_key = request.session.session_key except Exception: # pragma: no cover session_key = '' return visitor_fingerprint(ip_address, user_agent, session_key) # -- main entry point --------------------------------------------------- def track(self, request, response=None): if not self.enabled or not self.should_track(request): return None try: ip_address = get_client_ip(request) user_agent = request.META.get('HTTP_USER_AGENT', '') or '' bot = is_bot(user_agent) key = self._visitor_key(request, ip_address, user_agent, bot) session = self._get_or_create_session(key, request, ip_address, user_agent, bot) if session is None: return None now = timezone.now() update_fields = [] # Path / navigation updates (cheap fields, always refreshed on touch) path = request.path[:500] touch_key = f'visitor:touch:{key}' if not cache.get(touch_key): cache.set(touch_key, True, self.touch_interval) session.last_seen = now session.last_path = path if not session.landing_path: session.landing_path = path session.region_code = session.region_code or self._region_code(path) session.page_views = (session.page_views or 0) + 1 update_fields += ['last_seen', 'last_path', 'landing_path', 'region_code', 'page_views'] if not session.country_code and not is_private_ip(ip_address) and not bot: self._apply_geo(session, ip_address) update_fields += [ 'country_code', 'country_name', 'region', 'city', 'postal_code', 'latitude', 'longitude', 'timezone_name', 'isp', 'org', 'asn', 'is_proxy', 'is_hosting', ] try: session.save(update_fields=list(dict.fromkeys(update_fields))) except Exception: logger.debug("Failed to update visitor session %s", key, exc_info=True) if self.track_pageviews and not bot: pv_key = f'visitor:pv:{key}' if not cache.get(pv_key): cache.set(pv_key, True, self.pageview_interval) try: VisitorPageView.objects.create( session=session, ip_address=ip_address if is_valid_ip(ip_address) else None, path=path, method=request.method, status_code=getattr(response, 'status_code', None), referer=(request.META.get('HTTP_REFERER', '') or '')[:500], user_agent=user_agent, ) except Exception: logger.debug("Failed to record page view for %s", key, exc_info=True) request.visitor_session = session return session except Exception: logger.debug("Visitor tracking failed", exc_info=True) return None def _get_or_create_session(self, key, request, ip_address, user_agent, bot): try: session = VisitorSession.objects.filter(session_key=key).first() except DatabaseError: return None if session is not None: return session parsed = parse_user_agent(user_agent) session = VisitorSession( session_key=key, ip_address=ip_address if is_valid_ip(ip_address) else None, ip_version=(6 if ':' in ip_address else 4) if is_valid_ip(ip_address) else 0, is_bot=bot, user_agent=user_agent[:2000], device_type=parsed['device_type'], browser=parsed['browser'][:50], os=parsed['os'][:50], referer=(request.META.get('HTTP_REFERER', '') or '')[:500], landing_path=request.path[:500], last_path=request.path[:500], region_code=self._region_code(request.path), page_views=0, first_seen=timezone.now(), last_seen=timezone.now(), ) if not bot and not is_private_ip(ip_address) and ip_address: self._apply_geo(session, ip_address) try: session.save() except Exception: # Race: another worker created it first. session = VisitorSession.objects.filter(session_key=key).first() return session def _apply_geo(self, session, ip_address): profile = self.geo.lookup(ip_address) session.country_code = (profile.country_code or '')[:2] session.country_name = profile.country_name[:100] session.region = profile.region[:100] session.city = profile.city[:100] session.postal_code = profile.postal_code[:20] session.latitude = profile.latitude session.longitude = profile.longitude session.timezone_name = (profile.timezone or '')[:64] session.isp = profile.isp[:200] session.org = profile.org[:200] session.asn = profile.asn[:50] session.is_proxy = profile.is_proxy session.is_hosting = profile.is_hosting # --------------------------------------------------------------------------- # Live analytics # --------------------------------------------------------------------------- class VisitorAnalytics: """Aggregate live visitor statistics for the admin dashboard.""" def __init__(self): self.online_window = int(getattr(settings, 'VISITOR_ONLINE_WINDOW_SECONDS', 300)) def online_now(self, minutes: int = 5) -> int: cache_key = f'visitor:online:{minutes}' value = cache.get(cache_key) if value is not None: return value since = timezone.now() - timedelta(minutes=minutes) try: value = VisitorSession.objects.filter(last_seen__gte=since, is_bot=False).count() except DatabaseError: value = 0 cache.set(cache_key, value, 15) return value def online_humans(self) -> int: return self.online_now(5) def pageviews_last(self, minutes: int = 60) -> int: since = timezone.now() - timedelta(minutes=minutes) try: return VisitorPageView.objects.filter(viewed_at__gte=since).count() except DatabaseError: return 0 def sessions_today(self) -> int: since = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) try: return VisitorSession.objects.filter(first_seen__gte=since, is_bot=False).count() except DatabaseError: return 0 def pageviews_today(self) -> int: since = timezone.now().replace(hour=0, minute=0, second=0, microsecond=0) try: return VisitorPageView.objects.filter(viewed_at__gte=since).count() except DatabaseError: return 0 def bot_sessions(self, hours: int = 24) -> int: since = timezone.now() - timedelta(hours=hours) try: return VisitorSession.objects.filter(last_seen__gte=since, is_bot=True).count() except DatabaseError: return 0 def top_countries(self, hours: int = 24, limit: int = 8): since = timezone.now() - timedelta(hours=hours) try: return list( VisitorSession.objects.filter(last_seen__gte=since, is_bot=False) .exclude(country_code='') .values('country_code', 'country_name') .annotate(total=Count('id')) .order_by('-total')[:limit] ) except DatabaseError: return [] def top_pages(self, hours: int = 24, limit: int = 8): since = timezone.now() - timedelta(hours=hours) try: return list( VisitorPageView.objects.filter(viewed_at__gte=since) .values('path') .annotate(total=Count('id')) .order_by('-total')[:limit] ) except DatabaseError: return [] def top_referers(self, hours: int = 24, limit: int = 6): since = timezone.now() - timedelta(hours=hours) try: return list( VisitorSession.objects.filter(last_seen__gte=since) .exclude(referer='') .values('referer') .annotate(total=Count('id')) .order_by('-total')[:limit] ) except DatabaseError: return [] def blocked_count(self) -> int: try: return BlockedIP.objects.filter(is_active=True).count() except DatabaseError: return 0 def summary(self) -> dict: return { 'online_now': self.online_now(5), 'online_30m': self.online_now(30), 'sessions_today': self.sessions_today(), 'pageviews_today': self.pageviews_today(), 'pageviews_1h': self.pageviews_last(60), 'bot_sessions_24h': self.bot_sessions(24), 'blocked_ips': self.blocked_count(), 'top_countries': self.top_countries(), 'top_pages': self.top_pages(), 'top_referers': self.top_referers(), } # --------------------------------------------------------------------------- # Module-level singletons # --------------------------------------------------------------------------- BLOCKLIST = BlocklistService() AUTO_BLOCK = AutoBlockService() ANALYTICS = VisitorAnalytics()