SA-4 SSRF protections implemented.

backend/app/services/mastodon_security.py

@@ -0,0 +1,62 @@
+## Copyright © 2026 Olaf Kolkman
+## SPDX-License-Identifier: GPL-3.0-or-later
+
+import ipaddress
+import socket
+from urllib.parse import urlsplit
+from urllib.error import HTTPError
+from urllib.request import HTTPRedirectHandler, Request, build_opener
+
+
+class RejectRedirectHandler(HTTPRedirectHandler):
+    def redirect_request(self, request, file, code, msg, headers, newurl):
+        raise HTTPError(request.full_url, code, 'Redirects are not allowed', headers, None)
+
+
+NO_REDIRECT_OPENER = build_opener(RejectRedirectHandler)
+
+
+def _is_blocked_address(address: str) -> bool:
+    parsed = ipaddress.ip_address(address)
+    mapped = parsed.ipv4_mapped if isinstance(parsed, ipaddress.IPv6Address) else None
+    candidates = (parsed, mapped) if mapped else (parsed,)
+    return any(
+        candidate.is_loopback
+        or candidate.is_link_local
+        or candidate.is_private
+        or candidate.is_multicast
+        or candidate.is_unspecified
+        or candidate.is_reserved
+        for candidate in candidates
+    )
+
+
+def validate_public_instance(instance: str) -> str:
+    value = instance.strip().rstrip('/')
+    if not value:
+        raise ValueError('Mastodon instance is required')
+    if '://' not in value:
+        value = f'https://{value}'
+    parsed = urlsplit(value)
+    if parsed.scheme.lower() != 'https' or not parsed.hostname:
+        raise ValueError('Mastodon instance must be an HTTPS public hostname')
+    if parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/'):
+        raise ValueError('Mastodon instance must be a hostname-only HTTPS URL')
+    try:
+        port = parsed.port
+    except ValueError as error:
+        raise ValueError('Mastodon instance has an invalid port') from error
+    if port not in (None, 443):
+        raise ValueError('Mastodon instance must use HTTPS port 443')
+    hostname = parsed.hostname.rstrip('.').lower()
+    try:
+        addresses = {result[4][0] for result in socket.getaddrinfo(hostname, port or 443, type=socket.SOCK_STREAM)}
+    except socket.gaierror as error:
+        raise ValueError('Mastodon instance hostname could not be resolved') from error
+    if not addresses or any(_is_blocked_address(address) for address in addresses):
+        raise ValueError('Mastodon instance must resolve only to public IP addresses')
+    return f'https://{hostname}'
+
+
+def open_no_redirect(request: Request, timeout: int = 10):
+    return NO_REDIRECT_OPENER.open(request, timeout=timeout)