first commit

backend/app/endpoint_security.py

@@ -0,0 +1,62 @@
+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()}"