convert.py

import json
import sys
from collections import OrderedDict, defaultdict
from datetime import date
from pathlib import Path
from typing import Dict, List, Set


class DomainBlocklistConverter:

    INPUT_FILE = "pihole-google.txt"
    PIHOLE_FILE = "google-domains"
    UNBOUND_FILE = "pihole-google-unbound.conf"
    DNSMASQ_FILE = "pihole-google-dnsmasq.conf"
    ADGUARD_FILE = "pihole-google-adguard.txt"
    ADGUARD_IMPORTANT_FILE = "pihole-google-adguard-important.txt"
    WILDCARDS_FILE = "wildcards-domains"
    RPZ_FILE = "pihole-google-rpz.txt"
    CATEGORIES_PATH = "categories"

    BLOCKLIST_ABOUT = "This blocklist helps to restrict access to Google and its domains. Contribute at https://github.com/nickspaargaren/no-google"

    def __init__(self):
        self.data: Dict[List] = OrderedDict()
        self.timestamp: str = date.today().strftime("%Y-%m-%d")

    def read(self):
        """
        Read input file into `self.data`, a dictionary mapping category names to lists of member items.
        """
        with open(self.INPUT_FILE, "r") as f:
            category = None
            for line in f:
                line = line.strip()
                if line.startswith("#"):
                    category = line.lstrip("# ")
                    self.data.setdefault(category, [])
                else:
                    if category is None:
                        raise ValueError("Unable to store item without category")
                    self.data[category].append(line)

    def dump(self):
        """
        Output data in JSON format on STDOUT.
        """
        print(json.dumps(self.data, indent=4))

    def pihole(self):
        """
        Produce blocklist for the Pi-hole.
        """
        with open(self.PIHOLE_FILE, "w") as f:
            f.write(f"# {self.BLOCKLIST_ABOUT}\n")
            f.write(f"# Last updated: {self.timestamp}\n")
            for category, entries in self.data.items():
                f.write(f"# {category}\n")
                for entry in entries:
                    if entry != "":
                        f.write(f"0.0.0.0 {entry}\n")

    def unbound(self):
        """
        Produce blocklist for the Unbound DNS server.

        https://github.com/nickspaargaren/no-google/issues/67
        """
        with open(self.UNBOUND_FILE, "w") as f:
            f.write(f"# {self.BLOCKLIST_ABOUT}\n")
            f.write(f"# Last updated: {self.timestamp}\n")
            for category, entries in self.data.items():
                f.write(f"\n# Category: {category}\n")
                for entry in entries:
                    if entry != "":
                        f.write(f'local-zone: "{entry}" always_refuse\n')

    def dnsmasq(self):
        """
        Produce blocklist for DNSMasq v2.86+.

        Uses the local=/domain/ syntax which is supported in DNSMasq v2.86+.
        This format is used by OpenWrt and other systems.
        https://github.com/nickspaargaren/no-google/issues/196
        """
        with open(self.DNSMASQ_FILE, "w") as f:
            f.write(f"# {self.BLOCKLIST_ABOUT}\n")
            f.write(f"# Last updated: {self.timestamp}\n")
            f.write(f"# DNSMasq v2.86+ format\n")
            for category, entries in self.data.items():
                f.write(f"\n# Category: {category}\n")
                for entry in entries:
                    if entry != "":
                        f.write(f"local=/{entry}/\n")

    def rpz(self):
        """
        Produce blocklist in RPZ (Response Policy Zone) format.

        RPZ is a DNS zone file format used by BIND, Knot, Unbound and other DNS servers
        for implementing DNS firewalls and blocking policies.
        https://github.com/nickspaargaren/no-google/issues/256
        """
        # Generate serial number in YYYYMMDD01 format
        serial = date.today().strftime("%Y%m%d") + "01"

        with open(self.RPZ_FILE, "w") as f:
            # Write RPZ zone file header
            f.write("$TTL 18000\n")
            f.write(
                f"@ IN SOA localhost. root.localhost. {serial} 25200 3600 2592000 18000\n"
            )
            f.write("  IN NS  localhost.\n")
            f.write(";\n")
            f.write("; Title: No Google RPZ\n")
            f.write(f"; Description: {self.BLOCKLIST_ABOUT}\n")
            f.write("; Homepage: https://github.com/nickspaargaren/no-google\n")
            f.write(f"; Last modified: {self.timestamp}\n")
            f.write(";\n")

            # Write domain blocking entries
            for category, entries in self.data.items():
                f.write(f"\n; {category}\n")
                filtered_entries = self._filter_redundant_subdomains(entries)
                for entry in filtered_entries:
                    if entry != "":
                        # QNAME trigger with NXDOMAIN action
                        # Both apex and wildcard entries are required to block domain and all subdomains
                        f.write(f"{entry} CNAME .\n")
                        f.write(f"*.{entry} CNAME .\n")

    def _filter_redundant_subdomains(self, entries: List[str]) -> List[str]:
        """
        Remove redundant subdomain entries where parent domain already exists.

        In AdGuard syntax, ||domain.com^ blocks domain.com and all subdomains.
        This method removes entries like ||sub.domain.com^ when ||domain.com^ exists.
        """
        # Filter out empty entries and wildcards
        valid_entries = [e for e in entries if e and not e.startswith("*")]

        # Build set of all domains (lowercase for comparison)
        all_domains: Set[str] = {domain.lower() for domain in valid_entries}

        # Identify redundant subdomains
        redundant: Set[str] = set()
        for subdomain in all_domains:
            # Check if any parent domain exists
            for parent in all_domains:
                if subdomain != parent and subdomain.endswith("." + parent):
                    redundant.add(subdomain)
                    break

        # Return filtered list preserving original case and order
        return [e for e in valid_entries if e.lower() not in redundant]

    def adguard(self):
        """
        Produce blocklist for AdGuard.
        Automatically removes redundant subdomain rules.
        """
        with open(self.ADGUARD_FILE, "w") as f:
            f.write(f"! {self.BLOCKLIST_ABOUT}\n")
            f.write(f"! Last updated: {self.timestamp}\n")
            for category, entries in self.data.items():
                f.write(f"! {category}\n")
                filtered_entries = self._filter_redundant_subdomains(entries)
                for entry in filtered_entries:
                    if entry != "":
                        f.write(f"||{entry}^\n")

    def adguard_important(self):
        """
        Produce blocklist for AdGuard including important syntax.
        Automatically removes redundant subdomain rules.
        """
        with open(self.ADGUARD_IMPORTANT_FILE, "w") as f:
            f.write(f"! {self.BLOCKLIST_ABOUT}\n")
            f.write(f"! Last updated: {self.timestamp}\n")
            for category, entries in self.data.items():
                f.write(f"! {category}\n")
                filtered_entries = self._filter_redundant_subdomains(entries)
                for entry in filtered_entries:
                    if entry != "":
                        f.write(f"||{entry}^$important\n")

    def _get_wildcard_domains(self, entries: List[str]) -> List[str]:
        """
        Extract base domains (e.g., 'example.com' from 'sub.example.com').
        """
        # Common second-level domain prefixes used in multi-part top-level domains
        second_level_prefixes = {"co", "com", "net", "org", "gov", "ac", "sch", "edu"}

        base_domains: Set[str] = set()
        for entry in entries:
            if entry:
                parts = entry.lower().split(".")
                if len(parts) >= 2:
                    # Check if it uses a multi-part top-level domain
                    if len(parts) >= 3 and parts[-2] in second_level_prefixes:
                        # For multi-part top-level domains, take last 3 parts (e.g., google.co.uk)
                        base_domains.add(".".join(parts[-3:]))
                    else:
                        # For regular top-level domains, take last 2 parts (e.g., google.com)
                        base_domains.add(".".join(parts[-2:]))
        return sorted(base_domains)

    def wildcards(self):
        """
        Produce wildcards-domains file with only base domains for wildcard blocking.
        """
        with open(self.WILDCARDS_FILE, "w") as f:
            f.write(
                f"#This blocklist tries to shorten down the original pihole-google list by keepin only the main domains in the list, making all the linked other domains hosted on those, be blocked as well.\n"
            )
            f.write(f"#Last updated: {self.timestamp}\n")

            for category, entries in self.data.items():
                # Filter to get only parent domains (remove all subdomains)
                filtered_entries = self._get_wildcard_domains(entries)

                # Only write category if it has entries
                if filtered_entries:
                    f.write(f"\n# {category}\n")
                    for entry in filtered_entries:
                        if entry != "":
                            f.write(f"{entry}\n")

    def categories(self):
        """
        Produce individual per-category blocklist files.
        """

        def write_file(path, category, entries, line_prefix=""):
            """
            Generic function to write per-category file in both flavours.
            """
            with open(path, "w") as f:
                f.write(f"# {self.BLOCKLIST_ABOUT}\n")
                f.write(f"# Last updated: {self.timestamp}\n")
                f.write(f"# {category}\n")
                f.write(f"\n")
                for entry in entries:
                    if entry != "":
                        f.write(f"{line_prefix}{entry}\n")

        for category, entries in self.data.items():

            # Compute file names.
            filename = category.replace(" ", "").lower()
            filepath = Path(self.CATEGORIES_PATH).joinpath(filename)
            text_file = filepath.with_suffix(".txt")
            parsed_file = str(filepath) + "parsed"

            # Write two flavours of per-category file.
            write_file(text_file, category, entries, line_prefix="0.0.0.0 ")
            write_file(parsed_file, category, entries)

    def duplicates(self):
        """
        Find duplicates in main source file.
        """
        hashes = defaultdict(int)
        for category, entries in self.data.items():
            for entry in entries:
                hashes[hash(entry)] += 1
        for category, entries in self.data.items():
            for entry in entries:
                hashvalue = hash(entry)
                if hashvalue in hashes:
                    count = hashes[hashvalue]
                    if count > 1:
                        print(
                            f"Domain {entry} found {count} times, please remove duplicate domains."
                        )
                        hashes[hashvalue] = 0


def run(action: str):
    """
    Invoke different actions on converter engine.
    """

    # Create converter instance and read input file.
    converter = DomainBlocklistConverter()
    converter.read()

    # Invoke special action "json".
    if action == "json":
        converter.dump()
        sys.exit()

    # Either invoke specific action, or expand to all actions.
    if action == "all":
        subcommands = action_candidates
    else:
        subcommands = [action]

    # Invoke all actions subsequently.
    for action in subcommands:
        print(f"Invoking subcommand '{action}'")
        method = getattr(converter, action)
        method()


if __name__ == "__main__":

    # Read subcommand from command line, with error handling.
    action_candidates = [
        "pihole",
        "unbound",
        "dnsmasq",
        "rpz",
        "adguard",
        "adguard_important",
        "wildcards",
        "categories",
    ]
    special_candidates = ["all", "duplicates", "json"]
    subcommand = None
    try:
        subcommand = sys.argv[1]
    except:
        pass
    if subcommand not in action_candidates + special_candidates:
        print(
            f"ERROR: Subcommand not given or invalid, please use one of {action_candidates + special_candidates}"
        )
        sys.exit(1)

    # Invoke subcommand.
    run(subcommand)