diff --git a/src/common/core/blacklist/jobs/blacklist-download.py b/src/common/core/blacklist/jobs/blacklist-download.py index be58d7d210..c9d2068560 100644 --- a/src/common/core/blacklist/jobs/blacklist-download.py +++ b/src/common/core/blacklist/jobs/blacklist-download.py @@ -54,6 +54,47 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: return False, b"" +def sanitize_validator(value: str) -> str: + """Keep a validator only when it is a single line.""" + value = value.strip() + if "\n" in value or "\r" in value: + return "" + return value + + +def parse_http_metadata(cached_data: bytes) -> dict: + """Extract ETag and Last-Modified stored in cache file comment headers.""" + meta = {"etag": "", "last_modified": ""} + for line in cached_data.split(b"\n"): + if not line.startswith(b"# "): + break + if line.startswith(b"# ETag: "): + meta["etag"] = sanitize_validator(line[8:].decode("utf-8", errors="replace")) + elif line.startswith(b"# Last-Modified: "): + meta["last_modified"] = sanitize_validator(line[17:].decode("utf-8", errors="replace")) + return meta + + +def iter_cached_entries(cached_data: bytes): + """Yield the entries of a per-URL cache file, skipping its comment header lines.""" + for line in cached_data.split(b"\n"): + line = line.strip() + if line and not line.startswith(b"#"): + yield line + + +def build_cache_header(url: str, etag: str = "", last_modified: str = "") -> bytes: + """Build cache file header bytes with URL and optional HTTP metadata.""" + header = b"# Downloaded from " + url.encode("utf-8") + b"\n" + etag = sanitize_validator(etag) + last_modified = sanitize_validator(last_modified) + if etag: + header += b"# ETag: " + etag.encode("utf-8") + b"\n" + if last_modified: + header += b"# Last-Modified: " + last_modified.encode("utf-8") + b"\n" + return header + + LOGGER = getLogger("BLACKLIST") status = 0 @@ -168,11 +209,11 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: deleted, err = JOB.del_cache(file, service_id=file.parent.name) if not deleted: - LOGGER.warning(f"Couldn't delete file {file} from cache : {err}") + LOGGER.error(f"Couldn't delete file {file} from cache : {err}") sys_exit(0) urls = set() - failed_urls = set() + failed_urls = set() # (kind, url) pairs: a failure while handling one kind must not skip another kind processed_urls = set() # Track which URLs have been processed globally # Initialize aggregation per kind with service tracking aggregated_recap = { @@ -181,6 +222,7 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: "total_urls": 0, "downloaded_urls": 0, "skipped_urls": 0, + "not_modified": 0, "failed_count": 0, "total_lines": 0, } @@ -195,7 +237,7 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: LOGGER.warning(f"{service} blacklist for {kind} is cached but no URL is configured, removing from cache...") deleted, err = JOB.del_cache(f"{kind}.list", service_id=service) if not deleted: - LOGGER.warning(f"Couldn't delete {service} {kind}.list from cache : {err}") + LOGGER.error(f"Couldn't delete {service} {kind}.list from cache : {err}") continue # Track that this service provided URLs for the current kind @@ -205,7 +247,10 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: unique_entries = set() content = b"" for url in urls_list: - url_file = f"{bytes_hash(url, algorithm='sha1')}.list" + # Key by kind too: cached entries are already transformed by check_line for that kind + # SHA-256 instead of SHA-1; TODO: switch the other list downloaders (greylist, whitelist, + # realip, dnsbl, robotstxt) to the same SHA-256 url_file key, they still use SHA-1 + url_file = f"{bytes_hash(f'{kind}:{url}', algorithm='sha256')}.list" cached_url = JOB.get_cache(url_file, with_info=True, with_data=True) try: # Only count URLs that haven't been processed globally @@ -213,7 +258,7 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: aggregated_recap[kind]["total_urls"] += 1 # If the URL has recently been downloaded, use cache - if url in failed_urls: + if (kind, url) in failed_urls: if url not in processed_urls: aggregated_recap[kind]["failed_count"] += 1 elif ( @@ -225,18 +270,17 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: if url not in processed_urls: aggregated_recap[kind]["skipped_urls"] += 1 # Process cached data and add to unique_entries - cached_data = cached_url.get("data", b"") - if cached_data: - # Skip first line (URL comment) and process entries - for line in cached_data.split(b"\n")[1:]: - line = line.strip() - if line: - unique_entries.add(line) + unique_entries.update(iter_cached_entries(cached_url.get("data") or b"")) else: failed = False + handle_304 = False + old_cached_data = cached_url.get("data", b"") if isinstance(cached_url, dict) else b"" + etag = "" + last_modified = "" LOGGER.info(f"Downloading blacklist data from {url} ...") download_url, url_filter = split_list_url(url) content_type = "" + body = b"" if download_url.startswith("file://"): try: body = Path(normpath(download_url[7:])).read_bytes() @@ -244,16 +288,26 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: status = 2 LOGGER.debug(format_exc()) LOGGER.error(f"Error while opening file {download_url[7:]} : {e}") - failed_urls.add(url) + failed_urls.add((kind, url)) if url_file not in urls: aggregated_recap[kind]["failed_count"] += 1 failed = True else: + # Build conditional request headers from stored metadata + req_headers = {} + if old_cached_data: + meta = parse_http_metadata(old_cached_data) + if meta["etag"]: + req_headers["If-None-Match"] = meta["etag"] + if meta["last_modified"]: + req_headers["If-Modified-Since"] = meta["last_modified"] + max_retries = 3 retry_count = 0 - while retry_count < max_retries: + # Leaves only by break (resp assigned) or by re-raising on the last attempt + while True: try: - resp = get(download_url, stream=True, timeout=10) + resp = get(download_url, stream=True, timeout=10, headers=req_headers or None) break except ConnectionError as e: retry_count += 1 @@ -262,49 +316,69 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: LOGGER.warning(f"Connection refused, retrying in 3 seconds... ({retry_count}/{max_retries})") sleep(3) - if resp.status_code != 200: + if resp.status_code == 304 and old_cached_data: + resp.close() + handle_304 = True + elif resp.status_code != 200: + # Also catches a 304 with nothing cached to fall back on status = 2 - LOGGER.warning(f"Got status code {resp.status_code}, skipping...") - failed_urls.add(url) + LOGGER.error(f"Got status code {resp.status_code}, skipping...") + resp.close() + failed_urls.add((kind, url)) if url_file not in urls: aggregated_recap[kind]["failed_count"] += 1 failed = True else: - body = resp.content - content_type = resp.headers.get("Content-Type", "") + try: + etag = resp.headers.get("ETag", "") + last_modified = resp.headers.get("Last-Modified", "") + content_type = resp.headers.get("Content-Type", "") + body = resp.content + finally: + resp.close() if not failed: - if url not in processed_urls: - aggregated_recap[kind]["downloaded_urls"] += 1 - - url_content = b"" - count_lines = 0 - for line in iter_list_entries( - body, - content_type=content_type, - url_filter=url_filter, - ip_list=kind in ("IP", "IGNORE_IP"), - whole_line=kind in ("USER_AGENT", "IGNORE_USER_AGENT"), - logger=LOGGER, - ): - ok, data = check_line(kind, line) - if ok: - unique_entries.add(data) - url_content += data + b"\n" - count_lines += 1 - if not count_lines: - LOGGER.warning(f"No valid {kind} entry found in {url} (Content-Type: {content_type or 'unknown'})") - if url not in processed_urls: - aggregated_recap[kind]["total_lines"] += count_lines - - cached, err = JOB.cache_file(url_file, b"# Downloaded from " + url.encode("utf-8") + b"\n" + url_content) - if not cached: - LOGGER.error(f"Error while caching url content for {url}: {err}") + if handle_304: + LOGGER.info(f"URL {url} returned 304 Not Modified, using cached data.") + if url not in processed_urls: + aggregated_recap[kind]["not_modified"] += 1 + # Re-save to bump last_update timestamp so the 1-hour check stays fresh + cached, err = JOB.cache_file(url_file, old_cached_data) + if not cached: + LOGGER.error(f"Error while refreshing cache for {url}: {err}") + unique_entries.update(iter_cached_entries(old_cached_data)) + else: + if url not in processed_urls: + aggregated_recap[kind]["downloaded_urls"] += 1 + + url_content = b"" + count_lines = 0 + for line in iter_list_entries( + body, + content_type=content_type, + url_filter=url_filter, + ip_list=kind in ("IP", "IGNORE_IP"), + whole_line=kind in ("USER_AGENT", "IGNORE_USER_AGENT"), + logger=LOGGER, + ): + ok, data = check_line(kind, line) + if ok: + unique_entries.add(data) + url_content += data + b"\n" + count_lines += 1 + if not count_lines: + LOGGER.warning(f"No valid {kind} entry found in {url} (Content-Type: {content_type or 'unknown'})") + if url not in processed_urls: + aggregated_recap[kind]["total_lines"] += count_lines + + cached, err = JOB.cache_file(url_file, build_cache_header(url, etag, last_modified) + url_content) + if not cached: + LOGGER.error(f"Error while caching url content for {url}: {err}") except BaseException as e: status = 2 LOGGER.debug(format_exc()) LOGGER.error(f"Exception while getting {service} blacklist from {url} :\n{e}") - failed_urls.add(url) + failed_urls.add((kind, url)) if url not in processed_urls: aggregated_recap[kind]["failed_count"] += 1 finally: @@ -312,6 +386,11 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: processed_urls.add(url) urls.add(url_file) + # Keep a failed URL's last downloaded entries, so the list is not rebuilt without them + if (kind, url) in failed_urls and isinstance(cached_url, dict) and cached_url.get("data"): + LOGGER.warning(f"Using the last cached {kind} entries from {url} until it can be downloaded again") + unique_entries.update(iter_cached_entries(cached_url["data"])) + # Build final content from unique entries, sorted for consistency content = b"\n".join(sorted(unique_entries)) + b"\n" if unique_entries else b"" @@ -345,11 +424,12 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: continue successful = recap["downloaded_urls"] skipped = recap["skipped_urls"] + not_modified = recap["not_modified"] failed = recap["failed_count"] total_lines = recap["total_lines"] LOGGER.info( f"Recap for {kind} urls: Total Services: {service_count}, Successful: {successful}, " - f"Skipped (cached): {skipped}, Failed: {failed}, Total Lines: {total_lines}" + f"Skipped (cached): {skipped}, Not modified: {not_modified}, Failed: {failed}, Total Lines: {total_lines}" ) # Remove old files @@ -359,7 +439,7 @@ def check_line(kind: str, line: bytes) -> Tuple[bool, bytes]: LOGGER.warning(f"Removing no longer used url file {url_file} ...") deleted, err = JOB.del_cache(url_file) if not deleted: - LOGGER.warning(f"Couldn't delete url file {url_file} from cache : {err}") + LOGGER.error(f"Couldn't delete url file {url_file} from cache : {err}") except SystemExit as e: status = e.code except BaseException as e: diff --git a/tests/unit/jobs/test_blacklist_conditional_download.py b/tests/unit/jobs/test_blacklist_conditional_download.py new file mode 100644 index 0000000000..65275beb86 --- /dev/null +++ b/tests/unit/jobs/test_blacklist_conditional_download.py @@ -0,0 +1,256 @@ +"""Conditional GET (ETag / Last-Modified) in the blacklist download job, against a local server and a cache that can age.""" + +import logging +import os +import runpy +import sys +from hashlib import sha1, sha256 +from pathlib import Path +from time import time +from types import SimpleNamespace + +import pytest +import requests + +from common_utils import bytes_hash + +ROOT = Path(os.environ.get("BW_BLACKLIST_JOB_ROOT", Path(__file__).resolve().parents[3])) +JOB_SCRIPT = ROOT / "src/common/core/blacklist/jobs/blacklist-download.py" +ETAG = '"v1"' +LAST_MODIFIED = "Wed, 21 Oct 2026 07:28:00 GMT" +ENTRIES = b"192.0.2.1\n198.51.100.0/24\n" + + +class Cache: + """Job cache stand-in that keeps last_update per file, so a test can push it past the 1-hour skip.""" + + def __init__(self, path): + self.job_path = path + self.data = {} + self.updated = {} + + def get_cache(self, name, service_id="", **kwargs): + data = self.data.get((service_id, name)) + return {"last_update": self.updated[service_id, name], "data": data} if data is not None else None + + def cache_file(self, name, data, service_id="", **kwargs): + self.data[service_id, name] = data + self.updated[service_id, name] = time() + path = self.job_path / service_id / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + return True, "" + + def cache_hash(self, name, service_id=""): + data = self.data.get((service_id, name)) + return bytes_hash(data) if data is not None else None + + def del_cache(self, name, service_id=""): + path = name if isinstance(name, Path) else self.job_path / service_id / name + self.data.pop((service_id, path.name), None) + self.updated.pop((service_id, path.name), None) + path.unlink(missing_ok=True) + return True, "" + + def url_files(self): + return {name: data for (service_id, name), data in self.data.items() if not service_id} + + def age(self): + for key in self.updated: + self.updated[key] -= 7200 + + +def run_job(cache, monkeypatch, **urls): + for key in list(os.environ): + if key.startswith("BLACKLIST_"): + monkeypatch.delenv(key) + for key, value in {"SERVER_NAME": "lists.example", "MULTISITE": "no", "USE_BLACKLIST": "yes", "BLACKLIST_COMMUNITY_LISTS": "", **urls}.items(): + monkeypatch.setenv(key, value) + monkeypatch.setitem(sys.modules, "jobs", SimpleNamespace(Job=lambda *args: cache)) + monkeypatch.setitem(sys.modules, "logger", SimpleNamespace(getLogger=logging.getLogger)) + with pytest.raises(SystemExit) as result: + runpy.run_path(str(JOB_SCRIPT)) + return result.value.code + + +def conditional_route(seen, etag=ETAG, body=ENTRIES): + """Answer 304 when the client replays ``etag``, else 200 with validators; record what was sent.""" + + def route(headers): + seen.append((headers.get("If-None-Match"), headers.get("If-Modified-Since"))) + if headers.get("If-None-Match") == etag: + return 304, b"", {"ETag": etag} + return 200, body, {"ETag": etag, "Last-Modified": LAST_MODIFIED} + + return route + + +@pytest.mark.parametrize("etag", (pytest.param(ETAG, id="strong-etag"), pytest.param('W/"v1"', id="weak-etag"))) +def test_validators_stored_then_replayed_and_304_reuses_entries(etag, url_server, tmp_path, monkeypatch, caplog): + caplog.set_level(logging.INFO) + seen = [] + url_server.routes["/ips"] = conditional_route(seen, etag=etag) + url = url_server.http("/ips") + cache = Cache(tmp_path) + + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 1 + assert seen == [(None, None)] + assert list(cache.url_files().values()) == [f"# Downloaded from {url}\n# ETag: {etag}\n# Last-Modified: {LAST_MODIFIED}\n".encode() + ENTRIES] + + # Within the hour: no request, and the header lines are not read back as entries + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 0 + assert len(seen) == 1 + assert cache.data["lists.example", "IP.list"] == ENTRIES + + # Past the hour: both validators replayed as stored, 304 keeps the entries and restarts the hour + cache.age() + caplog.clear() + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 0 + assert seen[-1] == (etag, LAST_MODIFIED) + assert cache.data["lists.example", "IP.list"] == ENTRIES + assert all(time() - updated < 60 for (service_id, _), updated in cache.updated.items() if not service_id) + assert "Not modified: 1" in caplog.text + + +def test_304_on_header_only_cache_writes_no_list(url_server, tmp_path, monkeypatch): + seen = [] + url_server.routes["/ips"] = conditional_route(seen, body=b"# no entries yet\n") + url = url_server.http("/ips") + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 0 + assert list(cache.url_files().values()) == [f"# Downloaded from {url}\n# ETag: {ETAG}\n# Last-Modified: {LAST_MODIFIED}\n".encode()] + + cache.age() + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 0 + assert seen[-1] == (ETAG, LAST_MODIFIED) + assert ("lists.example", "IP.list") not in cache.data + + +def test_failed_url_keeps_its_last_cached_entries(url_server, tmp_path, monkeypatch, caplog): + url_server.routes["/a"] = (200, b"192.0.2.1\n", {}) + url_server.routes["/b"] = (200, b"198.51.100.1\n", {}) + urls = f"{url_server.http('/a')} {url_server.http('/b')}" + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=urls) == 1 + + # /b goes down and /a changes: /a's update lands, /b's last good entries stay + url_server.routes["/a"] = (200, b"192.0.2.2\n", {}) + url_server.routes["/b"] = (503, b"down", {}) + cache.age() + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=urls) == 2 + assert cache.data["lists.example", "IP.list"] == b"192.0.2.2\n198.51.100.1\n" + assert "Using the last cached IP entries" in caplog.text + + +def test_changed_validator_downloads_new_entries(url_server, tmp_path, monkeypatch): + seen = [] + url_server.routes["/ips"] = conditional_route(seen) + url = url_server.http("/ips") + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 1 + + url_server.routes["/ips"] = conditional_route(seen, etag='"v2"', body=b"203.0.113.7\n") + cache.age() + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 1 + assert seen[-1] == (ETAG, LAST_MODIFIED) + assert cache.data["lists.example", "IP.list"] == b"203.0.113.7\n" + assert b'# ETag: "v2"\n' in next(iter(cache.url_files().values())) + + +def test_unsolicited_304_without_cache_is_a_failure(url_server, tmp_path, monkeypatch): + url_server.routes["/ips"] = (304, b"", {}) + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url_server.http("/ips")) == 2 + assert ("lists.example", "IP.list") not in cache.data + assert not cache.url_files() + + +def test_connection_refused_retries_three_times_then_fails(tmp_path, monkeypatch): + attempts, waits = [], [] + + def refuse(*args, **kwargs): + attempts.append(args[0]) + raise requests.exceptions.ConnectionError("refused") + + # The job binds both names at import time, so patch them before runpy executes it + monkeypatch.setattr("requests.get", refuse) + monkeypatch.setattr("time.sleep", waits.append) + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS="http://127.0.0.1:9/ips") == 2 + assert len(attempts) == 3 + assert waits == [3, 3] + assert not cache.url_files() + + +@pytest.mark.parametrize( + "header", + ( + pytest.param(b"", id="legacy-cache-without-validators"), + pytest.param(b'# ETag: "a\rb"\n', id="validator-with-carriage-return"), + ), +) +def test_cache_without_usable_validator_gets_plain_download(header, url_server, tmp_path, monkeypatch): + seen = [] + url_server.routes["/ips"] = conditional_route(seen) + url = url_server.http("/ips") + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 1 + + [url_file] = cache.url_files() + cache.cache_file(url_file, f"# Downloaded from {url}\n".encode() + header + ENTRIES) + cache.age() + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 0 + assert seen[-1] == (None, None) + assert f"# ETag: {ETAG}\n".encode() in cache.url_files()[url_file] + + +def test_url_cache_file_named_by_sha256_of_kind_and_url(url_server, tmp_path, monkeypatch): + url_server.routes["/ips"] = (200, ENTRIES, {}) + url = url_server.http("/ips") + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 1 + assert list(cache.url_files()) == [f"{sha256(f'IP:{url}'.encode()).hexdigest()}.list"] + + +def test_legacy_sha1_url_file_is_removed(url_server, tmp_path, monkeypatch): + url_server.routes["/ips"] = (200, ENTRIES, {}) + url = url_server.http("/ips") + cache = Cache(tmp_path) + legacy = f"{sha1(url.encode()).hexdigest()}.list" + cache.cache_file(legacy, f"# Downloaded from {url}\n".encode() + ENTRIES) + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url) == 1 + assert legacy not in cache.url_files() + assert not (tmp_path / legacy).exists() + + +def test_failure_for_one_kind_does_not_skip_another_kind(url_server, tmp_path, monkeypatch): + url_server.routes["/mixed"] = (200, b"/admin\n", {}) + url = url_server.http("/mixed") + ip_url_file = f"{sha256(f'IP:{url}'.encode()).hexdigest()}.list" + + class FailingIpCache(Cache): + def cache_file(self, name, data, service_id="", **kwargs): + if name == ip_url_file: + raise OSError("disk full") + return super().cache_file(name, data, service_id, **kwargs) + + cache = FailingIpCache(tmp_path) + # IP runs first (KINDS order): its cache write fails, URI must still download and publish its list + assert run_job(cache, monkeypatch, BLACKLIST_IP_URLS=url, BLACKLIST_URI_URLS=url) == 2 + assert cache.data["lists.example", "URI.list"] == b"/admin\n" + + +def test_same_url_under_two_kinds_keeps_entries_apart(url_server, tmp_path, monkeypatch): + seen = [] + url_server.routes["/mixed"] = conditional_route(seen, body=b"/admin\n") + url = url_server.http("/mixed") + cache = Cache(tmp_path) + assert run_job(cache, monkeypatch, BLACKLIST_USER_AGENT_URLS=url, BLACKLIST_URI_URLS=url) == 1 + assert cache.data["lists.example", "USER_AGENT.list"] == b"(?:\\b)/admin(?:\\b)\n" + assert cache.data["lists.example", "URI.list"] == b"/admin\n" + + # The 304 path must keep them apart too + cache.age() + assert run_job(cache, monkeypatch, BLACKLIST_USER_AGENT_URLS=url, BLACKLIST_URI_URLS=url) == 0 + assert seen[-2:] == [(ETAG, LAST_MODIFIED)] * 2 + assert cache.data["lists.example", "URI.list"] == b"/admin\n"