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_credentials(request): """Extract (username, token) from 'Authorization: Bearer '.""" auth = request.headers.get('Authorization', '') if not auth.lower().startswith('bearer '): return None, None parts = auth.split(' ', 2) # ['Bearer', '', ''] if len(parts) < 3: return None, None return parts[1].strip(), parts[2].strip() def _authenticate_token(request, username, token): import hmac from django.contrib.auth import get_user_model from django.utils import timezone from profiles.models import ApiToken if not username or not token: return False User = get_user_model() # Resolve the user case-insensitively; prefer an exact-case match if it # exists to avoid ambiguity. user = ( User.objects.filter(username=username).first() or User.objects.filter(username__iexact=username).first() ) if user is None: return False prefix = token[:8] # Scope the lookup to the claimed user so a token never authenticates as # anyone but its owner (prevents cross-authentication). for candidate in ApiToken.objects.filter(user=user, 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 credential to bypass the # gate (non-browser downloaders such as the Fabric mod). Credentials # are 'Authorization: Bearer '. Token auth is # CSRF-safe (header-based, not cookie-based), so mark the request as # CSRF-processed to let it reach the view. if _is_token_request(request): username, token = _get_credentials(request) if _authenticate_token(request, username, token): request.csrf_processing_done = True 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)