fix: code hardening in URL parsing and UrlStore

courlan/clean.py

@@ -14,9 +14,7 @@
 
 # parsing
 PROTOCOLS = re.compile(r"https?://")
-SELECTION = re.compile(
-    r'(https?://[^">&? ]+?)(?:https?://)|(?:https?://[^/]+?/[^/]+?[&?]u(rl)?=)(https?://[^"> ]+)'
-)
+SELECTION = re.compile(r'(https?://[^">&? ]+?)(?:https?://)')
 
 MIDDLE_URL = re.compile(r"https?://.+?(https?://.+?)(?:https?://|$)")
 
@@ -36,7 +34,7 @@
 TRACKERS_RE = re.compile(
     r"^(?:dc|fbc|gc|twc|yc|ysc)lid|"
     r"^(?:click|gbra|msclk|igsh|partner|wbra)id|"
-    r"^(?:ads?|mc|ga|gs|itm|mc|mkt|ml|mtm|oly|pk|utm|vero)_|"
+    r"^(?:ads?|mc|ga|gs|itm|mkt|ml|mtm|oly|pk|utm|vero)_|"
     r"(?:\b|_)(?:aff|affi|affiliate|campaign|cl?id|eid|ga|gl|"
     r"kwd|keyword|medium|ref|referr?er|session|source|uid|xtor)"
 )
@@ -89,7 +87,6 @@ def scrub_url(url: str) -> str:
         url = match[1]
     if len(url) > 500:  # arbitrary choice
         LOGGER.debug("invalid-looking link %s of length %d", url[:50] + "…", len(url))
-
     # trailing slashes in URLs without path or in embedded URLs
     if url.count("/") == 3 or url.count("://") > 1:
         url = url.rstrip("/")

courlan/core.py

@@ -178,13 +178,14 @@ def extract_links(
 
     # extract links
     for link in (m[0] for m in FIND_LINKS_REGEX.finditer(pagecontent)):
-        if "rel" in link and "nofollow" in link:
+        if "rel=" in link and "nofollow" in link:
             continue
         # https://en.wikipedia.org/wiki/Hreflang
         if no_filter is False and language is not None and "hreflang" in link:
             langmatch = HREFLANG_REGEX.search(link)
             if langmatch and (
-                langmatch[1].startswith(language) or langmatch[1] == "x-default"
+                (lang := langmatch[1].lower()).startswith(language)
+                or lang == "x-default"
             ):
                 linkmatch = LINK_REGEX.search(link)
                 if linkmatch:
@@ -214,7 +215,7 @@ def extract_links(
                 continue
             link = checked[0]
             # external/internal links
-            if external_bool != is_external(
+            if reference and external_bool != is_external(
                 url=link, reference=reference, ignore_suffix=True
             ):
                 continue

courlan/sampling.py

@@ -21,11 +21,7 @@ def _make_sample(
     "Iterate through the hosts in store and draw samples."
     output_urls = []
     for domain in urlstore.urldict:  # key=cmp_to_key(locale.strcoll)
-        urlpaths = [
-            p.path()
-            for p in urlstore._load_urls(domain)
-            if p.urlpath not in (b"/", None)
-        ]
+        urlpaths = [p.path() for p in urlstore._load_urls(domain) if p.urlpath != b"/"]
         # too few or too many URLs
         if (
             not urlpaths

courlan/urlstore.py

@@ -88,11 +88,11 @@ class DomainEntry:
 
     def __init__(self, state: State = State.OPEN) -> None:
         self.count: int = 0
-        self.rules: RobotFileParser | None = None
+        self.rules: bytes | RobotFileParser | None = None
         self.state: State = state
         self.timestamp: datetime | None = None
         self.total: int = 0
-        self.tuples: deque[UrlPathTuple] = deque()
+        self.tuples: bytes | deque[UrlPathTuple] = deque()
 
 
 class UrlPathTuple:
@@ -203,11 +203,12 @@ def _buffer_urls(
         return inputdict
 
     def _load_urls(self, domain: str) -> deque[UrlPathTuple]:
-        if domain in self.urldict:
-            if self.compressed:
-                return COMPRESSOR.decompress(self.urldict[domain].tuples)  # type: ignore
-            return self.urldict[domain].tuples
-        return deque()
+        if domain not in self.urldict:
+            return deque()
+        raw = self.urldict[domain].tuples
+        if isinstance(raw, bytes):  # compressed
+            return COMPRESSOR.decompress(raw)
+        return raw
 
     def _set_done(self) -> None:
         if not self.done and all(v.state != State.OPEN for v in self.urldict.values()):
@@ -220,6 +221,7 @@ def _store_urls(
         to_right: deque[UrlPathTuple] | None = None,
         timestamp: datetime | None = None,
         to_left: deque[UrlPathTuple] | None = None,
+        replace: bool = False,
     ) -> None:
         # http/https switch
         if domain.startswith("http://"):
@@ -236,21 +238,24 @@ def _store_urls(
                     del self.urldict[candidate]
 
         # load URLs or create entry
-        if domain in self.urldict:
-            # discard if busted
-            if self.urldict[domain].state is State.BUSTED:
-                return
-            urls = self._load_urls(domain)
-            known = {u.path() for u in urls}
+        if domain in self.urldict and self.urldict[domain].state is State.BUSTED:
+            return
+        if replace and to_right is not None:
+            urls = to_right  # skip dedup: store caller's already-mutated deque
         else:
-            urls = deque()
-            known = set()
-
-        # check if the link or its variants are known
-        if to_right is not None:
-            urls.extend(t for t in to_right if not is_known_link(t.path(), known))
-        if to_left is not None:
-            urls.extendleft(t for t in to_left if not is_known_link(t.path(), known))
+            if domain in self.urldict:
+                urls = self._load_urls(domain)
+                known = {u.path() for u in urls}
+            else:
+                urls = deque()
+                known = set()
+
+            if to_right is not None:
+                urls.extend(t for t in to_right if not is_known_link(t.path(), known))
+            if to_left is not None:
+                urls.extendleft(
+                    t for t in to_left if not is_known_link(t.path(), known)
+                )
 
         with self._lock:
             if self.compressed:
@@ -414,7 +419,9 @@ def get_url(self, domain: str, as_visited: bool = True) -> str | None:
                         url.visited = True
                         with self._lock:
                             self.urldict[domain].count += 1
-                        self._store_urls(domain, url_tuples, timestamp=datetime.now())
+                        self._store_urls(
+                            domain, url_tuples, timestamp=datetime.now(), replace=True
+                        )
                     return domain + url.path()
         # nothing to draw from
         with self._lock:
@@ -447,7 +454,7 @@ def get_download_urls(
 
     def establish_download_schedule(
         self, max_urls: int = 100, time_limit: int = 10
-    ) -> list[str]:
+    ) -> list[tuple[float, str]]:
         """Get up to the specified number of URLs along with a suitable
         backoff schedule (in seconds)."""
         # see which domains are free
@@ -492,10 +499,10 @@ def establish_download_schedule(
             # calculate difference and offset last addition
             total_diff = now + timedelta(0, schedule_secs - time_limit)
             # store new info
-            self._store_urls(domain, url_tuples, timestamp=total_diff)
+            self._store_urls(domain, url_tuples, timestamp=total_diff, replace=True)
         # sort by first tuple element (time in secs)
         self._set_done()
-        return sorted(targets, key=itemgetter(1))  # type: ignore[arg-type]
+        return sorted(targets, key=itemgetter(0))
 
     # CRAWLING
 
@@ -507,22 +514,20 @@ def store_rules(self, website: str, rules: RobotFileParser | None) -> None:
 
     def get_rules(self, website: str) -> RobotFileParser | None:
         "Return the stored crawling rules for the given website."
-        if website in self.urldict:
-            if self.compressed:
-                return COMPRESSOR.decompress(self.urldict[website].rules)  # type: ignore
-            return self.urldict[website].rules
-        return None
+        if website not in self.urldict:
+            return None
+        raw = self.urldict[website].rules
+        if isinstance(raw, bytes):  # compressed
+            return COMPRESSOR.decompress(raw)
+        return raw
 
     def get_crawl_delay(self, website: str, default: float = 5) -> float:
         "Return the delay as extracted from robots.txt, or a given default."
-        delay = None
         rules = self.get_rules(website)
-        try:
-            delay = rules.crawl_delay("*")  # type: ignore[union-attr]
-        except AttributeError:  # no rules or no crawl delay
-            pass
-        # backup
-        return delay or default  # type: ignore[return-value]
+        if rules is None:
+            return default
+        delay = rules.crawl_delay("*")
+        return float(delay) if delay is not None else default
 
     # GENERAL INFO
 

courlan/urlutils.py

@@ -6,7 +6,7 @@
 from html import unescape
 from urllib.parse import SplitResult, urljoin, urlsplit, urlunsplit
 
-from tld import get_tld
+from tld import Result, get_tld
 
 DOMAIN_REGEX = re.compile(
     r"(?:(?:f|ht)tp)s?://"  # protocols
@@ -36,10 +36,10 @@ def get_tldinfo(url: str, fast: bool = False) -> tuple[str | None, str | None]:
                 return clean_match, full_domain
     # fallback
     tldinfo = get_tld(url, as_object=True, fail_silently=True)
-    if tldinfo is None:
+    if not isinstance(tldinfo, Result):
         return None, None
     # this step is necessary to standardize output
-    return tldinfo.domain, CLEAN_FLD_REGEX.sub("", tldinfo.fld)  # type: ignore[union-attr]
+    return tldinfo.domain, CLEAN_FLD_REGEX.sub("", tldinfo.fld)
 
 
 def extract_domain(
@@ -108,13 +108,14 @@ def fix_relative_urls(baseurl: str, url: str) -> str:
     if url.startswith("{"):
         return url
 
-    base_netloc = urlsplit(baseurl).netloc
+    parsed_base = urlsplit(baseurl)
+    base_netloc = parsed_base.netloc
     split_url = urlsplit(url)
 
     if split_url.netloc not in (base_netloc, ""):
         if split_url.scheme:
             return url
-        return urlunsplit(split_url._replace(scheme="http"))
+        return urlunsplit(split_url._replace(scheme=parsed_base.scheme or "http"))
 
     return urljoin(baseurl, url)
 

tests/unit_tests.py

@@ -125,7 +125,14 @@ def test_fix_relative():
     )
     assert (
         fix_relative_urls("https://example.org", "//www.eff.org")
-        == "http://www.eff.org"
+        == "https://www.eff.org"
+    )
+    assert (
+        fix_relative_urls("http://example.org", "//www.eff.org") == "http://www.eff.org"
+    )
+    assert (
+        fix_relative_urls("http://example.org", "https://www.eff.org")
+        == "https://www.eff.org"
     )
     # looks like an absolute URL but is actually a valid relative URL
     assert (
@@ -755,6 +762,11 @@ def test_domain_filter():
     "Test filters related to domain and hostnames."
     assert domain_filter("") is False
     assert domain_filter("a" * 254 + ".com") is False  # exceeds DNS length limit
+    d_ok = "a." * 125 + "abc"  # 253 chars — at the DNS length limit
+    d_long = "a." * 125 + "abcd"  # 254 chars — over
+    assert len(d_ok) == 253 and len(d_long) == 254
+    assert domain_filter(d_ok) is True
+    assert domain_filter(d_long) is False
     assert domain_filter("too-long" + "g" * 60 + ".org") is False
     assert domain_filter("long" + "g" * 50 + ".org") is True
     assert domain_filter("example.-com") is False
@@ -815,6 +827,13 @@ def _fake_head(status, location):
         # transport failure: redirection_test raises ValueError, check_url returns None
         mock_request.side_effect = Exception("unreachable")
         assert check_url("https://www.ht.or", with_redirects=True) is None
+        # geturl() returns None in urllib3 2.x
+        mock_request.side_effect = None
+        mock_request.return_value = _fake_head(200, None)
+        assert check_url("http://example.org/page", with_redirects=True) == (
+            "http://example.org/page",
+            "example.org",
+        )
 
 
 def test_redirection(httpserver):
@@ -964,6 +983,8 @@ def test_extraction():
     # nofollow
     pagecontent = '<html><a href="https://test.com/example" rel="nofollow ugc"/></html>'
     assert not extract_links(pagecontent, "https://test.com/", False)
+    pagecontent = '<html><a href="https://test.com/rel/nofollow-guide"/></html>'
+    assert len(extract_links(pagecontent, "https://test.com/", False)) == 1
     # language
     pagecontent = '<html><a href="https://test.com/example" hreflang="de-DE"/></html>'
     assert len(extract_links(pagecontent, "https://test.com/", False)) == 1
@@ -986,6 +1007,17 @@ def test_extraction():
     assert (
         len(extract_links(pagecontent, "https://test.com/", False, language="en")) == 1
     )
+    pagecontent = '<html><a href="https://test.com/example" hreflang="DE-DE"/></html>'
+    assert (
+        len(extract_links(pagecontent, "https://test.com/", False, language="de")) == 1
+    )
+    assert not extract_links(pagecontent, "https://test.com/", False, language="en")
+    pagecontent = (
+        '<html><a href="https://test.com/example" hreflang="X-DEFAULT"/></html>'
+    )
+    assert (
+        len(extract_links(pagecontent, "https://test.com/", False, language="de")) == 1
+    )
     # language + content
     pagecontent = '<html><a hreflang="de-DE" href="https://test.com/example"/><a href="https://test.com/example2"/><a href="https://test.com/example2 ADDITIONAL"/></html>'
     links = extract_links(pagecontent, "https://test.com/", external_bool=False)
@@ -1116,6 +1148,10 @@ def test_extraction_navigation():
     assert extract_links(pagecontent, "https://knoema.com/", external_bool=True) == {
         "https://knoema.recruitee.com"
     }
+    # without url, external_bool must not filter (no reference to compare)
+    pagecontent = '<html><a href="https://example.com/page"/><a href="https://other.org/post"/></html>'
+    assert len(extract_links(pagecontent)) == 2
+    assert len(extract_links(pagecontent, external_bool=True)) == 2
     # return all links without filters
     pagecontent = '<html><a hreflang="de-DE" href="https://test.com/example"/><a href="/page/2"/><a href="https://example.com/gallery/"/></html>'
     result = extract_links(
@@ -1296,6 +1332,9 @@ def test_sample():
     assert len(sample_urls(mylist, 1, verbose=True)) == 2
     assert not sample_urls(mylist, 1, exclude_min=10, verbose=True)
     assert len(sample_urls(mylist, 1, exclude_max=1, verbose=True)) == 1
+    bound_urls = [f"http://bound.org/{i}" for i in range(3)]
+    assert len(sample_urls(bound_urls, 10, exclude_min=3)) == 3
+    assert not sample_urls(bound_urls, 10, exclude_min=4)
 
     test_urls = [f"https://test.org/{a}" for a in range(1000)]
     example_urls = [f"https://www.example.org/{a}" for a in range(100)]

tests/urlstore_tests.py

@@ -540,6 +540,55 @@ def test_urlstore_open_but_all_visited():
     assert entry.state == State.ALL_VISITED
 
 
+def test_urlstore_compressed_defaults(robots_rules):
+    "Compressed store handles unset rules/tuples and discard gracefully."
+    s = UrlStore(compressed=True)
+    s.add_urls(["http://x.com/a"])
+
+    assert s.get_rules("http://x.com") is None
+    assert s.get_crawl_delay("http://x.com") == 5
+
+    s.discard(["http://x.com"])
+    assert s.dump_urls() == []
+
+    zero_rules = RobotFileParser()
+    zero_rules.parse(["User-agent: *", "Disallow:", "Crawl-delay: 0"])
+    delay_store = UrlStore()
+    delay_store.add_urls(["http://x.com/a"])
+    delay_store.store_rules("http://x.com", zero_rules)
+    assert delay_store.get_crawl_delay("http://x.com") == 0
+
+
+def test_schedule_time_ordering():
+    "establish_download_schedule returns entries sorted by scheduled time, not URL string."
+    s = UrlStore()
+    s.add_urls(["http://x.com/z", "http://x.com/a"])  # /z first: deque ≠ alpha order
+    schedule = s.establish_download_schedule(max_urls=10, time_limit=10)
+    assert len(schedule) == 2
+    assert schedule[0][0] <= schedule[1][0], "entries must be time-ordered"
+    assert schedule[0][1].endswith("/z")  # /z inserted first → time 0.0
+
+
+def test_urlstore_compressed_crawl():
+    "get_url and establish_download_schedule persist visited flags in compressed mode."
+    s = UrlStore(compressed=True)
+    s.add_urls(["http://x.com/a", "http://x.com/b"])
+    u1 = s.get_url("http://x.com")
+    u2 = s.get_url("http://x.com")
+    u3 = s.get_url("http://x.com")
+    assert u1 != u2
+    assert u3 is None
+    assert s.is_exhausted_domain("http://x.com")
+    assert s.urldict["http://x.com"].count == 2
+
+    s2 = UrlStore(compressed=True)
+    s2.add_urls(["http://y.com/a", "http://y.com/b"])
+    run1 = s2.establish_download_schedule(max_urls=10, time_limit=10)
+    run2 = s2.establish_download_schedule(max_urls=10, time_limit=10)
+    assert len(run1) == 2
+    assert len(run2) == 0
+
+
 def test_urlstore_store_non_http_domain():
     "_store_urls tolerates a domain without an http(s) scheme (defensive fall-through)."
     store = UrlStore()