Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ jobs:
- os: windows-latest
python-version: "3.13"
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v5

# Python and uv setup
- name: Set up Python ${{ matrix.python-version }}
Expand Down
4 changes: 2 additions & 2 deletions courlan/clean.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@
)

MIDDLE_URL = re.compile(r"https?://.+?(https?://.+?)(?:https?://|$)")
NETLOC_RE = re.compile(r"(?<=\w):(?:80|443)")

# path
PATH1 = re.compile(r"/+")
Expand Down Expand Up @@ -182,7 +181,8 @@ def normalize_url(
except ValueError:
port = None # port could not be cast to integer value
if (scheme == "http" and port == 80) or (scheme == "https" and port == 443):
netloc = NETLOC_RE.sub("", netloc)
# strip the trailing default port (IPv6-safe)
netloc = netloc.rsplit(":", 1)[0]
# path: 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)))
Expand Down
25 changes: 17 additions & 8 deletions courlan/filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,13 +142,17 @@ def basic_filter(url: str) -> bool:

def domain_filter(domain: str) -> bool:
"Find invalid domain/host names."
# no valid FQDN exceeds the DNS length limit
if len(domain) > 253:
return False
# IPv4 or IPv6
if set(domain) <= IP_SET:
if all(c in IP_SET for c in domain):
try:
ip_address(domain)
return True
except ValueError:
return False
return True
# hex-only string that is not an IP (e.g. "abc.de") → keep validating
pass

# malformed domains
if not VALID_DOMAIN_PORT.match(domain):
Expand All @@ -163,13 +167,13 @@ def domain_filter(domain: str) -> bool:
return False

# extensions
extension_match = EXTENSION_REGEX.search(domain)
extension_match = EXTENSION_REGEX.search(domain.lower())
return not extension_match or extension_match[0] not in WHITELISTED_EXTENSIONS


def extension_filter(urlpath: str) -> bool:
"Filter based on file extension."
extension_match = EXTENSION_REGEX.search(urlpath)
extension_match = EXTENSION_REGEX.search(urlpath.lower())
return not extension_match or extension_match[0] in WHITELISTED_EXTENSIONS


Expand Down Expand Up @@ -254,11 +258,16 @@ def validate_url(url: str | None) -> tuple[bool, SplitResult | None]:
except ValueError:
return False, None

if not parsed_url.scheme or parsed_url.scheme not in PROTOCOLS:
if parsed_url.scheme not in PROTOCOLS:
return False, None

if len(parsed_url.netloc) < 5 or (
parsed_url.netloc.startswith("www.") and len(parsed_url.netloc) < 8
# plausibility checks on the network location (case-insensitive for "www.")
netloc = parsed_url.netloc
if (
len(netloc) < 4
or (netloc.lower().startswith("www.") and len(netloc) < 8)
# reject dotless/colonless hosts (e.g. "1234", "localhost")
or ("." not in netloc and ":" not in netloc)
):
return False, None

Expand Down
3 changes: 2 additions & 1 deletion courlan/network.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,8 @@
import urllib3

LOGGER = logging.getLogger(__name__)
urllib3.disable_warnings()
# only silence the warning triggered by cert_reqs="CERT_NONE" triggers
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


RETRY_STRATEGY = urllib3.util.Retry(
Expand Down
14 changes: 9 additions & 5 deletions courlan/urlstore.py
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,10 @@ def path(self) -> str:


class UrlStore:
"Defines a class to store domain-classified URLs and perform checks against it."
"""Defines a class to store domain-classified URLs and perform checks against it.

Thread-safety: readers are safe and writers are serialized, but a logical
write is not globally atomic — drive mutations from a single writer thread."""

__slots__ = (
"compressed",
Expand Down Expand Up @@ -226,10 +229,11 @@ def _store_urls(
domain = candidate
elif domain.startswith("https://"):
candidate = "http" + domain[5:]
# replace entry
if candidate in self.urldict:
self.urldict[domain] = self.urldict[candidate]
del self.urldict[candidate]
# replace entry: check-and-swap must be atomic against other writers
with self._lock:
if candidate in self.urldict:
self.urldict[domain] = self.urldict[candidate]
del self.urldict[candidate]

# load URLs or create entry
if domain in self.urldict:
Expand Down
3 changes: 2 additions & 1 deletion courlan/urlutils.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,8 @@


def get_tldinfo(url: str, fast: bool = False) -> tuple[str | None, str | None]:
"""Cached function to extract top-level domain info"""
"""Extract domain info, returning a ``(domain, full_domain)`` tuple.
With ``fast=True`` a regex shortcut is tried before the ``tld`` library."""
if not url or not isinstance(url, str):
return None, None
if fast:
Expand Down
29 changes: 29 additions & 0 deletions tests/unit_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -230,6 +230,10 @@ def test_extension_filter():
assert extension_filter(parsed_url.path) is True
_, parsed_url = validate_url("http://www.example.org/test.php6")
assert extension_filter(parsed_url.path) is True
# uppercase extensions are treated like their lowercase forms
assert extension_filter("/photo.JPG") is False
assert extension_filter("/page.HTML") is True
assert extension_filter("/index.PHP") is True


def test_spam_filter():
Expand Down Expand Up @@ -478,6 +482,18 @@ def test_validate():
assert not is_valid_url("http://www.test[.org/test")
assert is_valid_url("http://test.org/test")

# verdict no longer flips on port/userinfo/case; short valid domains accepted
assert is_valid_url("http://t.co/")
assert is_valid_url("http://t.co:80/")
assert is_valid_url("http://user@t.co/")
assert is_valid_url("http://g.co/")
assert not is_valid_url("http://WWW.a.b/")
assert not is_valid_url("http://www.a.b/")
# dotless/colonless and too-short hosts stay rejected
assert not is_valid_url("http://1234")
assert not is_valid_url("http://localhost/")
assert not is_valid_url("http://a.b/")


def test_normalization():
assert normalize_url("HTTPS://WWW.DWDS.DE/") == "https://www.dwds.de/"
Expand Down Expand Up @@ -505,6 +521,12 @@ def test_normalization():
normalize_url("https://hanxiao.io//404.html") == "https://hanxiao.io/404.html"
)

# IPv6: default port stripped (was missed by the old \w-lookbehind regex)
assert normalize_url("http://[::1]:80/") == "http://[::1]/"
assert normalize_url("https://[::1]:443/") == "https://[::1]/"
# non-default port preserved
assert normalize_url("http://[::1]:8080/") == "http://[::1]:8080/"

# punycode
assert normalize_url("http://xn--Mnchen-3ya.de") == "http://münchen.de"
assert normalize_url("http://Mnchen-3ya.de") == "http://mnchen-3ya.de"
Expand Down Expand Up @@ -719,6 +741,7 @@ def test_urlcheck_port():
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
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
Expand All @@ -739,10 +762,16 @@ def test_domain_filter():
assert domain_filter("https:") is False

assert domain_filter("127.0.0.1") is True
assert domain_filter("::1") is True
assert domain_filter("900.200.100.75") is False
assert domain_filter("111.111.111") is False
assert domain_filter("0127.0.0.1") is False

# hex-only strings that are not IPs must still be validated as domains
assert domain_filter("abc.de") is True
assert domain_filter("aced.de") is True
assert domain_filter("dead.beef") is True

assert domain_filter("example.jpg") is False
assert domain_filter("example.html") is False
assert domain_filter("0.gravatar.com") is False
Expand Down
Loading