379 lines
12 KiB
Python
379 lines
12 KiB
Python
"""Submitting, listing, the endpoint, rechecks and the directory page."""
|
||
|
||
from datetime import timedelta
|
||
import functools
|
||
|
||
import pytest
|
||
|
||
from mews import app, check, db, pages
|
||
from mews.fetch import Fetcher, registered_domain
|
||
|
||
|
||
@pytest.fixture
|
||
def loopback(monkeypatch):
|
||
"""Let the service read the test server, and give it a registrable domain.
|
||
|
||
The test server answers on 127.0.0.1, which has no public suffix, so the
|
||
name it would be listed under is supplied here rather than relaxing the
|
||
rule in the service.
|
||
"""
|
||
monkeypatch.setattr(app, "Fetcher", functools.partial(Fetcher, allow_loopback=True))
|
||
monkeypatch.setattr(
|
||
check,
|
||
"registered_domain",
|
||
lambda host: "example.test" if host == "127.0.0.1" else registered_domain(host),
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def service(tmp_path, monkeypatch, loopback):
|
||
"""The app, wired to an empty database and a temporary directory page."""
|
||
database = tmp_path / "mews.db"
|
||
monkeypatch.setenv("MEWS_DB", str(database))
|
||
monkeypatch.setenv("MEWS_DIRECTORY", str(tmp_path / "directory.html"))
|
||
handle = db.connect(str(database))
|
||
db.init(handle)
|
||
handle.close()
|
||
return tmp_path
|
||
|
||
|
||
def post(url: str = "", *, listing: bool = False, client: str = "203.0.113.5"):
|
||
"""Send one form submission to the WSGI app and return status and body."""
|
||
from io import BytesIO
|
||
from urllib.parse import urlencode
|
||
|
||
fields = {"url": url, "list" if listing else "check": "x"}
|
||
data = urlencode(fields).encode()
|
||
captured: dict = {}
|
||
|
||
def start_response(status, headers):
|
||
captured["status"] = status
|
||
captured["headers"] = headers
|
||
|
||
body = b"".join(
|
||
app.application(
|
||
{
|
||
"PATH_INFO": "/result",
|
||
"REQUEST_METHOD": "POST",
|
||
"CONTENT_LENGTH": str(len(data)),
|
||
"wsgi.input": BytesIO(data),
|
||
"HTTP_X_REAL_IP": client,
|
||
},
|
||
start_response,
|
||
)
|
||
)
|
||
return captured, body.decode()
|
||
|
||
|
||
def test_a_conforming_page_is_listed(service, site, conforming):
|
||
url = site.add("/index.html", conforming())
|
||
captured, body = post(url, listing=True)
|
||
assert captured["status"] == "200 OK"
|
||
assert "Your site is listed" in body
|
||
|
||
handle = db.connect()
|
||
assert [row["domain"] for row in db.listed(handle)] == ["example.test"]
|
||
assert (service / "directory.html").exists()
|
||
assert "example.test" in (service / "directory.html").read_text()
|
||
|
||
|
||
def test_checking_does_not_list(service, site, conforming):
|
||
url = site.add("/index.html", conforming())
|
||
_, body = post(url, listing=False)
|
||
assert "This page conforms" in body
|
||
assert db.listed(db.connect()) == []
|
||
|
||
|
||
def test_a_failing_page_is_not_listed_and_says_why(service, site, fail_page):
|
||
url = site.add("/index.html", fail_page("div"))
|
||
_, body = post(url, listing=True)
|
||
assert "doesn't conform" in body
|
||
assert "Section 4.1" in body
|
||
assert db.listed(db.connect()) == []
|
||
|
||
|
||
def test_every_response_is_a_mews_page(service, site, conforming):
|
||
from mews.lint import validate_bytes
|
||
|
||
url = site.add("/index.html", conforming())
|
||
for listing in (False, True):
|
||
_, body = post(url, listing=listing)
|
||
assert validate_bytes(body.encode()).conforms
|
||
|
||
|
||
def test_no_response_sets_a_cookie(service, site, conforming):
|
||
url = site.add("/index.html", conforming())
|
||
captured, _ = post(url, listing=True)
|
||
assert not [name for name, _ in captured["headers"] if name.lower() == "set-cookie"]
|
||
|
||
|
||
def test_responses_are_not_cached(service, site, conforming):
|
||
url = site.add("/index.html", conforming())
|
||
captured, _ = post(url, listing=True)
|
||
assert ("Cache-Control", "no-store") in captured["headers"]
|
||
|
||
|
||
def test_an_address_that_is_not_a_page_is_refused(service):
|
||
_, body = post("http://10.0.0.1/", listing=True)
|
||
assert "domain name rather than an IP address" in body
|
||
|
||
|
||
def test_an_unreachable_page_is_reported(service, site):
|
||
_, body = post(site.base + "/missing.html", listing=True)
|
||
assert "That page returned 404" in body
|
||
|
||
|
||
def test_resubmitting_a_site_updates_the_same_row(
|
||
service, site, conforming, monkeypatch
|
||
):
|
||
"""One row per registrable domain, however many pages of it are submitted."""
|
||
first = site.add("/a.html", conforming())
|
||
post(first, listing=True)
|
||
# Past the per-domain cooldown, which would otherwise turn this away.
|
||
later = db.now() + timedelta(hours=2)
|
||
monkeypatch.setattr(db, "now", lambda: later)
|
||
second = site.add("/b.html", conforming())
|
||
post(second, listing=True, client="203.0.113.6")
|
||
rows = db.listed(db.connect())
|
||
assert len(rows) == 1
|
||
assert rows[0]["url"] == second
|
||
|
||
|
||
def test_a_blocked_domain_is_refused(service, site, conforming):
|
||
handle = db.connect()
|
||
db.block(handle, "example.test", "spam")
|
||
url = site.add("/index.html", conforming())
|
||
_, body = post(url, listing=True)
|
||
assert "taken out of the directory" in body
|
||
|
||
|
||
def test_the_endpoint_answers_other_paths_and_methods_politely(service):
|
||
from io import BytesIO
|
||
|
||
captured: dict = {}
|
||
|
||
def start_response(status, headers):
|
||
captured["status"] = status
|
||
|
||
body = b"".join(
|
||
app.application(
|
||
{
|
||
"PATH_INFO": "/result",
|
||
"REQUEST_METHOD": "GET",
|
||
"wsgi.input": BytesIO(b""),
|
||
},
|
||
start_response,
|
||
)
|
||
).decode()
|
||
assert captured["status"].startswith("405")
|
||
assert "Nothing to show" in body
|
||
|
||
|
||
def test_rate_limits_stop_a_flood(connection):
|
||
client = db.ip_hash(connection, "203.0.113.9")
|
||
for _ in range(db.IP_LISTINGS_PER_HOUR):
|
||
assert (
|
||
db.rate_limited(connection, client=client, domain="a.example", listing=True)
|
||
is None
|
||
)
|
||
db.record_submission(
|
||
connection,
|
||
url="https://a.example/",
|
||
domain="other.example",
|
||
client=client,
|
||
outcome="listed",
|
||
)
|
||
assert "Try again in an hour" in db.rate_limited(
|
||
connection, client=client, domain="a.example", listing=True
|
||
)
|
||
|
||
|
||
def test_a_domain_has_to_wait_between_submissions(connection):
|
||
client = db.ip_hash(connection, "203.0.113.9")
|
||
db.record_submission(
|
||
connection,
|
||
url="https://a.example/",
|
||
domain="a.example",
|
||
client=client,
|
||
outcome="rejected",
|
||
)
|
||
assert "ten minutes" in db.rate_limited(
|
||
connection, client=client, domain="a.example", listing=True
|
||
)
|
||
|
||
|
||
def test_a_domain_that_keeps_failing_waits_a_day(connection, monkeypatch):
|
||
client = db.ip_hash(connection, "203.0.113.9")
|
||
base = db.now()
|
||
for index in range(db.DOMAIN_REJECTS_BEFORE_SLOWDOWN):
|
||
monkeypatch.setattr(db, "now", lambda i=index: base - timedelta(hours=i + 1))
|
||
db.record_submission(
|
||
connection,
|
||
url="https://a.example/",
|
||
domain="a.example",
|
||
client=client,
|
||
outcome="rejected",
|
||
)
|
||
monkeypatch.setattr(db, "now", lambda: base)
|
||
assert "try again tomorrow" in db.rate_limited(
|
||
connection, client=client, domain="a.example", listing=True
|
||
)
|
||
|
||
|
||
# --- rechecks ----------------------------------------------------------
|
||
|
||
|
||
def _list_site(connection, url):
|
||
return db.upsert_site(
|
||
connection,
|
||
domain="example.test",
|
||
url=url,
|
||
title="Test site",
|
||
description="",
|
||
language="en",
|
||
etag=None,
|
||
last_modified=None,
|
||
)[0]
|
||
|
||
|
||
def test_a_page_that_still_conforms_stays(connection, site, conforming):
|
||
url = site.add("/index.html", conforming(), headers={"ETag": '"v1"'})
|
||
site_id = _list_site(connection, url)
|
||
with Fetcher(allow_loopback=True) as fetcher:
|
||
assert (
|
||
check.recheck(
|
||
connection, db.site(connection, "example.test"), fetcher=fetcher
|
||
)
|
||
== "pass"
|
||
)
|
||
assert db.site(connection, "example.test")["state"] == "listed"
|
||
assert db.site(connection, "example.test")["etag"] == '"v1"'
|
||
assert site_id
|
||
|
||
|
||
def test_a_page_that_stopped_conforming_is_dropped(connection, site, fail_page):
|
||
url = site.add("/index.html", fail_page("div"))
|
||
_list_site(connection, url)
|
||
with Fetcher(allow_loopback=True) as fetcher:
|
||
assert (
|
||
check.recheck(
|
||
connection, db.site(connection, "example.test"), fetcher=fetcher
|
||
)
|
||
== "fail"
|
||
)
|
||
row = db.site(connection, "example.test")
|
||
assert row["state"] == "removed"
|
||
assert "Section 4.1" in row["reason"]
|
||
assert db.listed(connection) == []
|
||
|
||
|
||
def test_a_not_modified_answer_passes_without_rereading(connection, site, conforming):
|
||
url = site.add("/index.html", conforming(), headers={"ETag": '"v1"'})
|
||
_list_site(connection, url)
|
||
site.add("/index.html", b"", status=304, headers={"ETag": '"v1"'})
|
||
connection.execute("UPDATE sites SET etag = '\"v1\"'")
|
||
with Fetcher(allow_loopback=True) as fetcher:
|
||
assert (
|
||
check.recheck(
|
||
connection, db.site(connection, "example.test"), fetcher=fetcher
|
||
)
|
||
== "pass"
|
||
)
|
||
assert db.site(connection, "example.test")["state"] == "listed"
|
||
assert "If-None-Match" in site.requests[-1][1]
|
||
|
||
|
||
def test_downtime_takes_several_tries_to_drop_a_site(connection, site, conforming):
|
||
url = site.add("/index.html", conforming())
|
||
_list_site(connection, url)
|
||
site.routes.clear()
|
||
site.add("/index.html", b"broken", status=503)
|
||
for _ in range(db.TRANSIENT_LIMIT - 1):
|
||
with Fetcher(allow_loopback=True) as fetcher:
|
||
assert (
|
||
check.recheck(
|
||
connection, db.site(connection, "example.test"), fetcher=fetcher
|
||
)
|
||
== "unreachable"
|
||
)
|
||
assert db.site(connection, "example.test")["state"] == "listed"
|
||
with Fetcher(allow_loopback=True) as fetcher:
|
||
assert (
|
||
check.recheck(
|
||
connection, db.site(connection, "example.test"), fetcher=fetcher
|
||
)
|
||
== "fail"
|
||
)
|
||
assert db.site(connection, "example.test")["state"] == "removed"
|
||
|
||
|
||
def test_only_sites_that_are_due_are_rechecked(connection, monkeypatch):
|
||
_list_site(connection, "https://example.test/")
|
||
assert db.due(connection) == []
|
||
later = db.now() + timedelta(days=db.RECHECK_DAYS + 1)
|
||
monkeypatch.setattr(db, "now", lambda: later)
|
||
assert len(db.due(connection)) == 1
|
||
|
||
|
||
# --- the directory page ------------------------------------------------
|
||
|
||
|
||
def test_the_directory_page_conforms(connection, tmp_path):
|
||
from mews.lint import validate_bytes
|
||
|
||
_list_site(connection, "https://example.test/")
|
||
target = tmp_path / "directory.html"
|
||
pages.write_directory(str(target), db.listed(connection))
|
||
assert validate_bytes(target.read_bytes()).conforms
|
||
|
||
|
||
def test_a_hostile_title_is_escaped_and_stripped(connection, tmp_path):
|
||
from mews.lint import validate_bytes
|
||
|
||
connection.execute(
|
||
"INSERT INTO sites (domain, url, title, state, listed_at) "
|
||
"VALUES ('x.test', 'https://x.test/', '<script>oops</script>', "
|
||
"'listed', '2026-01-01T00:00:00+00:00')"
|
||
)
|
||
connection.commit()
|
||
target = tmp_path / "directory.html"
|
||
pages.write_directory(str(target), db.listed(connection))
|
||
text = target.read_text()
|
||
assert "<script>" not in text
|
||
assert "<script>" in text
|
||
assert validate_bytes(target.read_bytes()).conforms
|
||
|
||
|
||
def test_the_directory_page_is_readable_by_the_web_server(connection, tmp_path):
|
||
"""It is served off disk by another user, whatever umask the service has."""
|
||
import os
|
||
import stat
|
||
|
||
target = tmp_path / "directory.html"
|
||
old_umask = os.umask(0o077)
|
||
try:
|
||
pages.write_directory(str(target), db.listed(connection))
|
||
finally:
|
||
os.umask(old_umask)
|
||
assert stat.S_IMODE(target.stat().st_mode) == 0o644
|
||
|
||
|
||
def test_an_empty_directory_still_conforms(connection, tmp_path):
|
||
from mews.lint import validate_bytes
|
||
|
||
target = tmp_path / "directory.html"
|
||
pages.write_directory(str(target), db.listed(connection))
|
||
assert validate_bytes(target.read_bytes()).conforms
|
||
assert "Nothing here yet" in target.read_text()
|
||
|
||
|
||
def test_a_page_that_would_not_conform_leaves_the_old_one_alone(
|
||
connection, tmp_path, monkeypatch
|
||
):
|
||
target = tmp_path / "directory.html"
|
||
pages.write_directory(str(target), db.listed(connection))
|
||
before = target.read_text()
|
||
monkeypatch.setattr(pages, "directory", lambda rows: "<p>not a page</p>")
|
||
with pytest.raises(ValueError, match="doesn't conform"):
|
||
pages.write_directory(str(target), db.listed(connection))
|
||
assert target.read_text() == before
|