diff options
| -rw-r--r-- | pyproject.toml | 1 | ||||
| -rw-r--r-- | tests/conftest.py | 24 | ||||
| -rw-r--r-- | tests/test_forms.py | 2 | ||||
| -rw-r--r-- | tests/test_url_security.py | 42 | ||||
| -rw-r--r-- | tests/test_views.py | 3 | ||||
| -rw-r--r-- | uv.lock | 11 | ||||
| -rw-r--r-- | webmentions_ssg/forms/__init__.py | 3 | ||||
| -rw-r--r-- | webmentions_ssg/forms/validators.py | 25 | ||||
| -rw-r--r-- | webmentions_ssg/tasks/receiver.py | 18 | ||||
| -rw-r--r-- | webmentions_ssg/url_security.py | 37 |
10 files changed, 165 insertions, 1 deletions
diff --git a/pyproject.toml b/pyproject.toml index 9da3352..9af850a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,6 +17,7 @@ dependencies = [ "beautifulsoup4>=4.15.0", "bootstrap-flask>=2.5.0", "coloredlogs>=15.0.1", + "dnspython>=2.8.0", "flask>=3.1.3", "flask-login>=0.6.3", "flask-migrate>=4.1.0", diff --git a/tests/conftest.py b/tests/conftest.py index 94ba650..f1ef4b5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -18,6 +18,14 @@ HTTPHandler = Callable[ ] +class FakeHostAnswers: + def __init__(self, *addresses: str): + self._addresses = addresses + + def addresses(self): + return iter(self._addresses) + + @pytest.fixture def app() -> Iterator[Flask]: """ @@ -101,6 +109,12 @@ def install_httpx_mock( Replace the HTTPX transport without replacing HTTPX itself. """ + monkeypatch.setattr( + receiver_module, + "ensure_public_url", + lambda url: None, + ) + real_client = httpx.Client def install(handler: HTTPHandler) -> None: @@ -123,3 +137,13 @@ def install_httpx_mock( ) return install + + +@pytest.fixture +def public_dns( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + "webmentions_ssg.url_security.dns.resolver.resolve_name", + lambda hostname: FakeHostAnswers("93.184.216.34"), + ) diff --git a/tests/test_forms.py b/tests/test_forms.py index 9be6ca8..2d5fc01 100644 --- a/tests/test_forms.py +++ b/tests/test_forms.py @@ -89,6 +89,7 @@ VALID_TARGET = "https://dennisfink.me/blog/example/" ) def test_endpoint_form_rejects_invalid_data( app: Flask, + public_dns: None, form_data: dict[str, str], invalid_field: str, expected_error: str | None, @@ -111,6 +112,7 @@ def test_endpoint_form_rejects_invalid_data( def test_endpoint_form_accepts_valid_data( app: Flask, + public_dns: None, ) -> None: with app.test_request_context( "/endpoint", diff --git a/tests/test_url_security.py b/tests/test_url_security.py new file mode 100644 index 0000000..fba6d7d --- /dev/null +++ b/tests/test_url_security.py @@ -0,0 +1,42 @@ +import pytest + +from webmentions_ssg.url_security import ( + NonPublicAddressError, + ensure_public_url, +) + + +class FakeAnswers: + def __init__(self, *addresses: str): + self._addresses = addresses + + def addresses(self): + return iter(self._addresses) + + +@pytest.mark.parametrize( + ("url", "resolved_addresses"), + [ + ("http://127.0.0.1/", ["127.0.0.1"]), + ("http://127.0.0.1:8080/test", ["127.0.0.1"]), + ("http://[::1]/", ["::1"]), + ("http://10.0.0.1/", ["10.0.0.1"]), + ("http://172.16.0.1/", ["172.16.0.1"]), + ("http://192.168.1.1/", ["192.168.1.1"]), + ("http://169.254.169.254/", ["169.254.169.254"]), + ("http://localhost/", ["127.0.0.1", "::1"]), + ("http://internal.example/", ["192.168.1.10"]), + ], +) +def test_ensure_public_url_rejects_non_public_addresses( + monkeypatch: pytest.MonkeyPatch, + url: str, + resolved_addresses: list[str], +) -> None: + monkeypatch.setattr( + "webmentions_ssg.url_security.dns.resolver.resolve_name", + lambda hostname: FakeAnswers(*resolved_addresses), + ) + + with pytest.raises(NonPublicAddressError): + ensure_public_url(url) diff --git a/tests/test_views.py b/tests/test_views.py index d7fd798..70ed732 100644 --- a/tests/test_views.py +++ b/tests/test_views.py @@ -42,6 +42,7 @@ def test_endpoint_creates_webmention( client: FlaskClient, views_module: ModuleType, monkeypatch: pytest.MonkeyPatch, + public_dns: None, ) -> None: queued: list[uuid.UUID] = [] @@ -85,6 +86,7 @@ def test_endpoint_is_idempotent( client: FlaskClient, views_module: ModuleType, monkeypatch: pytest.MonkeyPatch, + public_dns: None, ) -> None: queued: list[uuid.UUID] = [] @@ -138,6 +140,7 @@ def test_resending_resets_failure_state( client: FlaskClient, views_module: ModuleType, monkeypatch: pytest.MonkeyPatch, + public_dns: None, ) -> None: monkeypatch.setattr( views_module, @@ -145,6 +145,15 @@ wheels = [ ] [[package]] +name = "dnspython" +version = "2.8.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/8c/8b/57666417c0f90f08bcafa776861060426765fdb422eb10212086fb811d26/dnspython-2.8.0.tar.gz", hash = "sha256:181d3c6996452cb1189c4046c61599b84a5a86e099562ffde77d26984ff26d0f", size = 368251, upload-time = "2025-09-07T18:58:00.022Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ba/5a/18ad964b0086c6e62e2e7500f7edc89e3faa45033c71c1893d34eed2b2de/dnspython-2.8.0-py3-none-any.whl", hash = "sha256:01d9bbc4a2d76bf0db7c1f729812ded6d912bd318d3b1cf81d30c0f845dbf3af", size = 331094, upload-time = "2025-09-07T18:57:58.071Z" }, +] + +[[package]] name = "flask" version = "3.1.3" source = { registry = "https://pypi.org/simple" } @@ -541,6 +550,7 @@ dependencies = [ { name = "beautifulsoup4" }, { name = "bootstrap-flask" }, { name = "coloredlogs" }, + { name = "dnspython" }, { name = "flask" }, { name = "flask-login" }, { name = "flask-migrate" }, @@ -566,6 +576,7 @@ requires-dist = [ { name = "beautifulsoup4", specifier = ">=4.15.0" }, { name = "bootstrap-flask", specifier = ">=2.5.0" }, { name = "coloredlogs", specifier = ">=15.0.1" }, + { name = "dnspython", specifier = ">=2.8.0" }, { name = "flask", specifier = ">=3.1.3" }, { name = "flask-login", specifier = ">=0.6.3" }, { name = "flask-migrate", specifier = ">=4.1.0" }, 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" + ) |