Spaces:
Paused
Paused
| """Outbound request filtering using community-maintained blocklists. | |
| Blocks ad networks, trackers, and beacons at the Playwright level to reduce | |
| unnecessary outbound connections and prevent abuse of the rendering service. | |
| Blocklist source: StevenBlack/hosts (MIT license) | |
| https://github.com/StevenBlack/hosts | |
| """ | |
| from __future__ import annotations | |
| import importlib.resources | |
| import logging | |
| from urllib.parse import urlparse | |
| from playwright.async_api import Route | |
| logger = logging.getLogger(__name__) | |
| def load_blocklist(extra_domains: list[str] | None = None) -> frozenset[str]: | |
| """Load the blocklist from the bundled data file. | |
| Args: | |
| extra_domains: Additional domains to block beyond the bundled list. | |
| Returns: | |
| A frozen set of blocked domain names for O(1) lookup. | |
| """ | |
| domains: set[str] = set() | |
| try: | |
| ref = importlib.resources.files("veilrender.data").joinpath("blocklist.txt") | |
| text = ref.read_text(encoding="utf-8") | |
| for line in text.splitlines(): | |
| line = line.strip() | |
| if line and not line.startswith("#"): | |
| domains.add(line.lower()) | |
| except Exception: | |
| logger.warning("Failed to load bundled blocklist, filtering disabled") | |
| if extra_domains: | |
| for d in extra_domains: | |
| d = d.strip().lower() | |
| if d: | |
| domains.add(d) | |
| logger.info("Loaded blocklist with %d domains", len(domains)) | |
| return frozenset(domains) | |
| def should_block(url: str, blocklist: frozenset[str]) -> bool: | |
| """Check if a URL should be blocked based on the blocklist. | |
| Performs suffix matching: for hostname ``a.b.c.com``, checks | |
| ``a.b.c.com``, ``b.c.com``, ``c.com`` against the set. | |
| Args: | |
| url: The request URL to check. | |
| blocklist: Set of blocked domain names. | |
| Returns: | |
| True if the URL's domain matches the blocklist. | |
| """ | |
| try: | |
| hostname = urlparse(url).hostname | |
| if not hostname: | |
| return False | |
| hostname = hostname.lower() | |
| # Walk up the domain hierarchy | |
| parts = hostname.split(".") | |
| for i in range(len(parts)): | |
| candidate = ".".join(parts[i:]) | |
| if candidate in blocklist: | |
| return True | |
| except Exception: | |
| pass | |
| return False | |
| def make_route_handler(blocklist: frozenset[str]): | |
| """Create a Playwright route handler that blocks requests to listed domains. | |
| Args: | |
| blocklist: Set of blocked domain names. | |
| Returns: | |
| An async route handler function for use with ``page.route()``. | |
| """ | |
| async def _handle_route(route: Route) -> None: | |
| url = route.request.url | |
| if should_block(url, blocklist): | |
| logger.info("Blocked: %s", url[:200]) | |
| await route.abort("blockedbyclient") | |
| else: | |
| await route.continue_() | |
| return _handle_route | |