4 files changed, 82 insertions, 1 deletions
diff --git a/webmentions_ssg/forms/__init__.py b/webmentions_ssg/forms/__init__.py
index 98962ad..0c54999 100644
--- a/webmentions_ssg/forms/__init__.py
+++ b/webmentions_ssg/forms/__init__.py
@@ -4,7 +4,7 @@ from flask_wtf import FlaskForm
 from wtforms import PasswordField, StringField, SubmitField
 from wtforms.validators import URL, InputRequired, Regexp
 
-from .validators import AllowedHostname, NotEqualTo
+from .validators import AllowedHostname, NotEqualTo, PublicURL
 
 
 class LoginForm(FlaskForm):
@@ -25,6 +25,7 @@ class EndpointForm(FlaskForm):
                 message="source must begin with http or https",
             ),
             NotEqualTo("target"),
+            PublicURL(),
         ],
     )
     target = StringField(
diff --git a/webmentions_ssg/forms/validators.py b/webmentions_ssg/forms/validators.py
index 20d9ebf..e979644 100644
--- a/webmentions_ssg/forms/validators.py
+++ b/webmentions_ssg/forms/validators.py
@@ -3,6 +3,12 @@ from urllib.parse import urlsplit
 from flask import current_app
 from wtforms import ValidationError
 
+from ..url_security import (
+    AddressResolutionError,
+    NonPublicAddressError,
+    ensure_public_url,
+)
+
 
 class NotEqualTo:
     """
@@ -60,3 +66,22 @@ class AllowedHostname:
             message = field.gettext("Invalid input.")
 
         raise ValidationError(message)
+
+
+class PublicURL:
+    def __init__(self, message=None):
+        self.message = message
+
+    def __call__(self, form, field):
+        try:
+            ensure_public_url(field.data)
+        except (
+            AddressResolutionError,
+            NonPublicAddressError,
+        ) as exc:
+            message = self.message
+
+            if message is None:
+                message = field.gettext("URL must resolve to a public address.")
+
+            raise ValidationError(message) from exc
diff --git a/webmentions_ssg/tasks/receiver.py b/webmentions_ssg/tasks/receiver.py
index ea48299..9475866 100644
--- a/webmentions_ssg/tasks/receiver.py
+++ b/webmentions_ssg/tasks/receiver.py
@@ -10,6 +10,11 @@ from .. import APP_NAME, VERSION
 from .. import DATABASE as db
 from .. import HUEY as huey
 from ..models import ReceivedWebmention
+from ..url_security import (
+    AddressResolutionError,
+    NonPublicAddressError,
+    ensure_public_url,
+)
 
 
 class VerificationError(Exception):
@@ -103,6 +108,16 @@ def text_mentions_target(body: str, target_url: str) -> bool:
     return any(match.group() == target_url for match in IRI_PATTERN.finditer(body))
 
 
+def ensure_public_request(request: httpx.Request) -> None:
+    """Prevent requests to non-public network addresses."""
+    try:
+        ensure_public_url(str(request.url))
+    except NonPublicAddressError as exc:
+        raise VerificationError("Source resolves to a non-public address") from exc
+    except AddressResolutionError as exc:
+        raise TemporaryFetchError("Source hostname could not be resolved") from exc
+
+
 def fetch_source(source_url: str) -> tuple[httpx.Response, bytes]:
     """Fetch a source with limits on redirects, time, and response size."""
 
@@ -117,6 +132,9 @@ def fetch_source(source_url: str) -> tuple[httpx.Response, bytes]:
         follow_redirects=True,
         max_redirects=current_app.config.get("WEBMENTIONS_SSG_MAX_REDIRECTS", 20),
         trust_env=False,
+        event_hooks={
+            "request": [ensure_public_request],
+        },
     ) as client:
         with client.stream("GET", source_url) as response:
             match response.status_code:
diff --git a/webmentions_ssg/url_security.py b/webmentions_ssg/url_security.py
new file mode 100644
index 0000000..ce5d700
--- /dev/null
+++ b/webmentions_ssg/url_security.py
@@ -0,0 +1,37 @@
+import ipaddress
+from urllib.parse import urlsplit
+
+import dns.exception
+import dns.resolver
+
+
+class AddressResolutionError(Exception):
+    pass
+
+
+class NonPublicAddressError(Exception):
+    pass
+
+
+def ensure_public_url(url: str) -> None:
+    hostname = urlsplit(url).hostname
+
+    if hostname is None:
+        raise NonPublicAddressError("URL has no hostname")
+
+    try:
+        answers = dns.resolver.resolve_name(hostname)
+    except dns.exception.DNSException as exc:
+        raise AddressResolutionError(
+            f"Could not resolve hostname {hostname!r}"
+        ) from exc
+
+    addresses = {ipaddress.ip_address(address) for address in answers.addresses()}
+
+    if not addresses:
+        raise AddressResolutionError(f"Hostname {hostname!r} did not resolve")
+
+    if any(not address.is_global for address in addresses):
+        raise NonPublicAddressError(
+            f"Hostname {hostname!r} resolves to a non-public address"
+        )