155 lines
5.1 KiB
Python
155 lines
5.1 KiB
Python
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 <username> <token>'."""
|
|
auth = request.headers.get('Authorization', '')
|
|
if not auth.lower().startswith('bearer '):
|
|
return None, None
|
|
parts = auth.split(' ', 2) # ['Bearer', '<username>', '<token>']
|
|
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 <username> <token>'. 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) |