62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
import hashlib
|
|
import hmac
|
|
import ipaddress
|
|
import socket
|
|
import time
|
|
from urllib.parse import urlsplit
|
|
|
|
from .config import settings
|
|
|
|
|
|
def validate_endpoint_url(value: str, *, websocket: bool) -> str:
|
|
url = value.strip()
|
|
parsed = urlsplit(url)
|
|
allowed_schemes = {"wss"} if websocket else {"https"}
|
|
if settings.environment == "development":
|
|
allowed_schemes.add("ws" if websocket else "http")
|
|
if parsed.scheme.lower() not in allowed_schemes:
|
|
expected = "wss://" if websocket else "https://"
|
|
raise ValueError(f"URL must start with {expected}")
|
|
if not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
|
|
raise ValueError("URL must contain a valid host and no credentials or fragment")
|
|
try:
|
|
address = ipaddress.ip_address(parsed.hostname)
|
|
except ValueError:
|
|
address = None
|
|
if settings.environment != "development" and address and not address.is_global:
|
|
raise ValueError("Private, loopback, and reserved network addresses are not allowed")
|
|
if settings.environment != "development" and not address:
|
|
try:
|
|
resolved = {
|
|
ipaddress.ip_address(item[4][0])
|
|
for item in socket.getaddrinfo(parsed.hostname, parsed.port, type=socket.SOCK_STREAM)
|
|
}
|
|
except socket.gaierror as exc:
|
|
raise ValueError("Endpoint host could not be resolved") from exc
|
|
if not resolved or any(not item.is_global for item in resolved):
|
|
raise ValueError("Endpoint host resolves to a private or reserved network address")
|
|
return url
|
|
|
|
|
|
def endpoint_signing_secret(kind: str, endpoint_id: object) -> str:
|
|
message = f"{kind}:{endpoint_id}".encode()
|
|
digest = hmac.new(settings.auth_secret.encode(), message, hashlib.sha256).hexdigest()
|
|
return f"urmi_{kind}_{digest}"
|
|
|
|
|
|
def signed_headers(kind: str, endpoint_id: object) -> dict[str, str]:
|
|
timestamp = str(int(time.time()))
|
|
secret = endpoint_signing_secret(kind, endpoint_id)
|
|
message = f"{endpoint_id}.{timestamp}".encode()
|
|
signature = hmac.new(secret.encode(), message, hashlib.sha256).hexdigest()
|
|
return {
|
|
"X-Urmi-Endpoint-Id": str(endpoint_id),
|
|
"X-Urmi-Timestamp": timestamp,
|
|
"X-Urmi-Signature": f"sha256={signature}",
|
|
"User-Agent": "Urmi-Relay/1.0",
|
|
}
|
|
|
|
|
|
def payload_signature(kind: str, endpoint_id: object, body: bytes) -> str:
|
|
secret = endpoint_signing_secret(kind, endpoint_id)
|
|
return f"sha256={hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()}"
|