63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
## 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)
|