"""Security helpers shared by middleware and trusted local services.""" import hashlib import hmac import ipaddress import time from urllib.parse import quote from django.conf import settings INTERNAL_SIGNATURE_TTL_SECONDS = 60 def _canonical_internal_request(timestamp, method, path, body): body_hash = hashlib.sha256(body or b"").hexdigest() return "%s\n%s\n%s\n%s" % (timestamp, method.upper(), path, body_hash) def build_internal_auth_headers(method, path, body=b"", timestamp=None): timestamp = str(int(timestamp or time.time())) message = _canonical_internal_request(timestamp, method, path, body).encode("utf-8") signature = hmac.new( settings.MONITOR_INTERNAL_API_SECRET.encode("utf-8"), message, hashlib.sha256 ).hexdigest() return { "X-Monitor-Timestamp": timestamp, "X-Monitor-Signature": signature, } def internal_hook_token_query(): return quote(settings.MONITOR_INTERNAL_API_SECRET, safe="") def _is_loopback(remote_addr): try: return ipaddress.ip_address((remote_addr or "").split("%", 1)[0]).is_loopback except ValueError: return False def verify_internal_request(request, now=None): """Require a direct loopback peer and either HMAC auth or the ZLM hook token.""" if not _is_loopback(request.META.get("REMOTE_ADDR")): return False expected_secret = settings.MONITOR_INTERNAL_API_SECRET token = request.GET.get("token", "") if token and hmac.compare_digest(token, expected_secret): return True timestamp = request.headers.get("X-Monitor-Timestamp", "") supplied = request.headers.get("X-Monitor-Signature", "") try: timestamp_int = int(timestamp) except (TypeError, ValueError): return False current = int(now or time.time()) if abs(current - timestamp_int) > INTERNAL_SIGNATURE_TTL_SECONDS: return False expected = build_internal_auth_headers( request.method, request.path_info, request.body, timestamp=timestamp_int )["X-Monitor-Signature"] return bool(supplied and hmac.compare_digest(supplied, expected))