import time from django.conf import settings from django.http import HttpResponseRedirect, JsonResponse from django.urls import Resolver404, resolve try: from django.utils.timezone import now except Exception: # pragma: no cover from datetime import datetime def now(): return datetime.now() class TimingMiddleware: def __init__(self, get_response): self.get_response = get_response def __call__(self, request): request._start_time = time.time() response = self.get_response(request) return response def _is_whitelisted(request): """Check whether the resolved view is marked with @gate_whitelist.""" try: resolver_match = resolve(request.path_info) except Resolver404: return False func = resolver_match.func if getattr(func, '__gate_whitelist__', False): return True view_class = getattr(func, 'view_class', None) if view_class and getattr(view_class, '__gate_whitelist__', False): return True return False def _is_token_request(request): """Token is honoured only for non-browser download/API requests.""" path = request.path_info if path.startswith('/api/') or path.startswith('/media/'): return True accept = request.META.get('HTTP_ACCEPT', '') if 'application/json' in accept: return True if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return True return False def _is_api_request(request): """Requests that must 401 (never redirect to an HTML page).""" path = request.path_info if path.startswith('/api/') or path.startswith('/media/'): return True accept = request.META.get('HTTP_ACCEPT', '') if 'application/json' in accept: return True if request.headers.get('X-Requested-With') == 'XMLHttpRequest': return True return False def _get_bearer_token(request): auth = request.headers.get('Authorization', '') if auth.lower().startswith('bearer '): return auth.split(' ', 1)[1].strip() explicit = request.headers.get('X-Auth-Token') return explicit or None def _authenticate_token(request, token): import hmac from django.utils import timezone from profiles.models import ApiToken if not token: return False prefix = token[:8] # Matches on the prefix keyed index, then constant-time compares. for candidate in ApiToken.objects.filter(key_prefix=prefix): if hmac.compare_digest(candidate.token or '', token): candidate.last_used = timezone.now() candidate.save(update_fields=['last_used']) setattr(request, 'user', candidate.user) setattr(request, '_cached_user', candidate.user) return True return False def gate_whitelist(view_func): """Mark a view as public (bypasses the gate).""" view_func.__gate_whitelist__ = True return view_func class GateMiddleware: """Gate every request behind the master-password session or an API token. Public views must be decorated with @gate_whitelist. """ def __init__(self, get_response): self.get_response = get_response def __call__(self, request): # Public static assets must load on the gate page itself. if request.path_info.startswith(settings.STATIC_URL): return self.get_response(request) # @gate_whitelist views (gate page + its POST) pass through. if _is_whitelisted(request): return self.get_response(request) # API/media requests are allowed a bearer token to bypass the gate # (non-browser downloaders such as the Fabric mod). if _is_token_request(request): token = _get_bearer_token(request) if _authenticate_token(request, token): return self.get_response(request) # fall through: reject with 401 for API/media. # Browser session authorized by the master password. if request.session.get('authorized'): return self.get_response(request) # Unauthorized: reject. if _is_api_request(request): return JsonResponse({'detail': 'Unauthorized'}, status=401) return HttpResponseRedirect(settings.GATE_URL)