Filter the normalized query in strict check_url path filtering

courlan/clean.py

@@ -161,35 +161,55 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str:
     return normalize_part(fragment)
 
 
-def normalize_url(
-    parsed_url: SplitResult | str,
-    strict: bool = False,
-    language: str | None = None,
-    trailing_slash: bool = True,
-) -> str:
-    "Takes a URL string or a parsed URL and returns a normalized URL string"
-    parsed_url = _parse(parsed_url)
-    # lowercase + remove fragments + normalize punycode
-    scheme = parsed_url.scheme.lower()
+def normalize_authority(parsed_url: SplitResult) -> str:
+    "Lower-case the authority, decode punycode and strip the scheme default port."
     netloc = decode_punycode(parsed_url.netloc.lower())
     # port: strip only the scheme's default port (80 for http, 443 for https)
     try:
         port = parsed_url.port
     except ValueError:
         port = None  # port could not be cast to integer value
+    scheme = parsed_url.scheme.lower()
     if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
         # strip the trailing default port (IPv6-safe)
         netloc = netloc.rsplit(":", 1)[0]
-    # path: https://github.com/saintamh/alcazar/blob/master/alcazar/utils/urls.py
+    return netloc
+
+
+def normalize_path(path: str) -> str:
+    "Collapse repeated slashes, drop leading /../ segments, then percent-normalize."
+    # https://github.com/saintamh/alcazar/blob/master/alcazar/utils/urls.py
     # leading /../'s in the path are removed
-    newpath = normalize_part(PATH2.sub("", PATH1.sub("/", parsed_url.path)))
+    return normalize_part(PATH2.sub("", PATH1.sub("/", path)))
+
+
+def normalize_url(
+    parsed_url: SplitResult | str,
+    strict: bool = False,
+    language: str | None = None,
+    trailing_slash: bool = True,
+    netloc: str | None = None,
+    path: str | None = None,
+    query: str | None = None,
+) -> str:
+    """Takes a URL string or a parsed URL and returns a normalized URL string.
+    `netloc`, `path` and `query` skip the corresponding step when the caller has
+    already normalized that part (see check_url)."""
+    parsed_url = _parse(parsed_url)
+    # lowercase + remove fragments + normalize punycode
+    scheme = parsed_url.scheme.lower()
+    if netloc is None:
+        netloc = normalize_authority(parsed_url)
+    if path is None:
+        path = normalize_path(parsed_url.path)
     # strip unwanted query elements
-    newquery = clean_query(parsed_url.query, strict, language)
-    if newquery and not newpath:
-        newpath = "/"
-    elif not trailing_slash and not newquery and newpath.endswith("/"):
-        newpath = newpath.rstrip("/")
+    if query is None:
+        query = clean_query(parsed_url.query, strict, language)
+    if query and not path:
+        path = "/"
+    elif not trailing_slash and not query and path.endswith("/"):
+        path = path.rstrip("/")
     # fragment
     newfragment = "" if strict else normalize_fragment(parsed_url.fragment, language)
     # rebuild
-    return urlunsplit((scheme, netloc, newpath, newquery, newfragment))
+    return urlunsplit((scheme, netloc, path, query, newfragment))

courlan/core.py

@@ -7,7 +7,13 @@
 import re
 from urllib.robotparser import RobotFileParser
 
-from .clean import normalize_url, scrub_url
+from .clean import (
+    clean_query,
+    normalize_authority,
+    normalize_path,
+    normalize_url,
+    scrub_url,
+)
 from .filters import (
     basic_filter,
     domain_filter,
@@ -95,24 +101,42 @@ def check_url(
             LOGGER.debug("rejected, validation test: %s", url)
             raise ValueError
 
+        # the filters below and normalize_url() need the same normalized parts, so
+        # compute each one once and pass them on
+        path = normalize_path(parsed_url.path)
+
         # content filter based on extensions
-        if extension_filter(parsed_url.path) is False:
+        if extension_filter(path) is False:
             LOGGER.debug("rejected, extension filter: %s", url)
             raise ValueError
 
         # unsuitable domain/host name (strip userinfo; domain_filter expects host[/port])
-        host = parsed_url.netloc.rsplit("@", 1)[-1]
+        netloc = normalize_authority(parsed_url)
+        host = netloc.rsplit("@", 1)[-1]
         if not host or domain_filter(host) is False:
             LOGGER.debug("rejected, domain name: %s", url)
             raise ValueError
 
-        # strict content filtering
-        if strict and path_filter(parsed_url.path, parsed_url.query) is False:
-            LOGGER.debug("rejected, path filter: %s", url)
-            raise ValueError
+        # strict content filtering: the query has to be the one that survives
+        # normalization, else a stripped param keeps an index page that
+        # check_url would reject on a second pass
+        query = None
+        if strict:
+            query = clean_query(parsed_url.query, strict, language)
+            if path_filter(path, query) is False:
+                LOGGER.debug("rejected, path filter: %s", url)
+                raise ValueError
 
         # normalize
-        url = normalize_url(parsed_url, strict, language, trailing_slash)
+        url = normalize_url(
+            parsed_url,
+            strict,
+            language,
+            trailing_slash,
+            netloc=netloc,
+            path=path,
+            query=query,
+        )
 
         # domain info: use blacklist in strict mode only
         domain = extract_domain(url, blacklist=BLACKLIST if strict else None)

tests/unit_tests.py

@@ -359,6 +359,57 @@ def test_path_filter():
     assert path_filter("/Datenschutzerklaerung", "") is False
     # assert path_filter("/", "") is False
 
+    # in strict mode an index page must not be kept alive by a query that
+    # normalization strips again: that returned a bare index URL which
+    # check_url itself rejects, so the result was not idempotent
+    for url in (
+        "http://www.case-modder.de/index.php?utm_source=x",  # tracker
+        "http://www.case-modder.de/index.php?sec=artikel",  # non-whitelisted key
+        "http://www.case-modder.de/default/?ref=abc",
+        "http://www.case-modder.de/home?foo=bar",
+    ):
+        assert check_url(url, strict=True) is None
+    # an index page with a surviving (whitelisted) query is still kept, and the
+    # canonical output is stable under a second pass
+    for url in (
+        "http://www.case-modder.de/index.php?id=68&page=1",
+        "http://www.case-modder.de/index.php?p=2",
+    ):
+        checked = check_url(url, strict=True)
+        assert checked is not None
+        assert check_url(checked[0], strict=True) == checked
+    # non-index pages are unaffected by query stripping
+    assert check_url("http://www.case-modder.de/article.html?ref=x", strict=True) == (
+        "http://www.case-modder.de/article.html",
+        "case-modder.de",
+    )
+    # a whitelisted param survives while a co-occurring tracker is stripped, so
+    # the index page is kept with only its meaningful query (also for &)
+    assert check_url(
+        "http://www.case-modder.de/index.php?utm_source=x&id=68", strict=True
+    ) == ("http://www.case-modder.de/index.php?id=68", "case-modder.de")
+    assert check_url(
+        "http://www.case-modder.de/index.php?utm_source=x&id=68", strict=True
+    ) == ("http://www.case-modder.de/index.php?id=68", "case-modder.de")
+    # a language param matching the requested language keeps the index page,
+    # while a mismatching one is filtered out
+    assert (
+        check_url(
+            "http://www.case-modder.de/index.php?lang=en", strict=True, language="en"
+        )
+        is not None
+    )
+    assert (
+        check_url(
+            "http://www.case-modder.de/index.php?lang=fr", strict=True, language="en"
+        )
+        is None
+    )
+    # only strict mode is affected: non-strict still returns the bare index page
+    assert check_url(
+        "http://www.case-modder.de/index.php?utm_source=x", strict=False
+    ) == ("http://www.case-modder.de/index.php", "case-modder.de")
+
 
 def test_lang_filter():
     assert lang_filter("http://test.com/az", "de", trailing_slash=False) is False
@@ -739,6 +790,9 @@ def test_urlcheck_language():
     assert check_url("http://www.example.org/index.html", strict=True) is None
     assert check_url("http://concordia-hagen.de/impressum.html", strict=True) is None
     assert check_url("http://concordia-hagen.de/de/impressum", strict=True) is None
+    # repeated slashes are collapsed before the path filter is applied
+    assert check_url("http://www.example.org/home//", strict=True) is None
+    assert check_url("http://concordia-hagen.de/impressum//", strict=True) is None
     assert (
         check_url("http://parkkralle.de/detail/index/sArticle/2704", strict=True)
         is not None