65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
"""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))
|