Astrologue-Website/tools/qa-crawl.py

#!/usr/bin/env python3
"""QA crawl: serve the site locally and fetch every subresource of every page.

Asserts that every request made while crawling goes to 127.0.0.1 — no CDN,
no Google Fonts, no picsum leftovers. Pages are discovered by following
local links from / so the crawl covers whatever is linked. CSS url(...)
references (e.g. @font-face src) are fetched too.

data:/mailto:/tel: URIs are self-contained by definition — never fetched,
never counted as external. Fragment-only links are same-page anchors.

Usage: python3 tools/qa-crawl.py [repo-root]
"""

import http.server
import os
import pathlib
import re
import socketserver
import sys
import threading
import urllib.parse
import urllib.request
from html.parser import HTMLParser

ROOT = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else ".").resolve()
PORT = int(os.environ.get("QA_CRAWL_PORT", "0"))  # 0 = ephemeral

RESOURCE_ATTRS = {"link": ["href"], "script": ["src"], "img": ["src", "srcset"],
                  "source": ["src", "srcset"], "video": ["src", "poster"],
                  "audio": ["src"], "track": ["src"], "embed": ["src"],
                  "object": ["data"], "use": ["href", "xlink:href"]}
NAV_ATTRS = {"a": ["href"], "area": ["href"], "form": ["action"]}

SELF_CONTAINED = ("data:", "mailto:", "tel:", "javascript:")


class Collector(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.resources, self.links = [], []

    def handle_starttag(self, tag, attrs):
        d = dict(attrs)
        if tag in RESOURCE_ATTRS:
            for a in RESOURCE_ATTRS[tag]:
                v = d.get(a)
                if v:
                    if a == "srcset":
                        for part in v.split(","):
                            self.resources.append(part.strip().split()[0])
                    else:
                        self.resources.append(v)
        if tag in NAV_ATTRS:
            v = d.get(NAV_ATTRS[tag][0])
            if v:
                self.links.append(v)


class QuietHandler(http.server.SimpleHTTPRequestHandler):
    def log_message(self, fmt, *args):
        pass  # the crawler records requests itself


def fetch(url):
    req = urllib.request.Request(url, headers={"User-Agent": "qa-crawl/1.0"})
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.status, r.read(), r.geturl()


def norm(base, ref):
    """Resolve ref against base; return (abs_url, fetchable_local)."""
    if ref.startswith(SELF_CONTAINED):
        return ref, True  # self-contained, nothing to fetch
    p = urllib.parse.urlparse(ref)
    if p.scheme or p.netloc:  # absolute — the thing we are hunting
        return ref, False
    return urllib.parse.urljoin(base, ref), True


def defrag(u):
    return u.split("#", 1)[0]


seen_pages, seen_res, external, failures = set(), set(), [], []


def fetch_resource(absu, referer):
    if absu in seen_res:
        return
    seen_res.add(absu)
    if absu.startswith(SELF_CONTAINED):
        return  # self-contained, nothing to fetch
    try:
        st, body, fin = fetch(absu)
        if st != 200:
            failures.append(f"asset {absu} (on {referer}): HTTP {st}")
            return
        if absu.endswith(".css"):
            # fetch url(...) references inside the stylesheet (fonts etc.);
            # strip data: URIs first (respecting the quote style actually
            # used) so nested url(#frag) refs inside them are not mistaken
            # for stylesheet references
            css = body.decode("utf-8", "replace")
            css = re.sub(r'url\(\s*"(?:[^"]*?data:[^"]*?)"\s*\)', "url()", css)
            css = re.sub(r"url\(\s*'(?:[^']*?data:[^']*?)'\s*\)", "url()", css)
            css = re.sub(r"url\(\s*data:[^'\")]*\s*\)", "url()", css)
            for ref in re.findall(r"url\(\s*['\"]?([^'\")]+)['\"]?\s*\)", css):
                if ref.startswith(SELF_CONTAINED) or ref.startswith(("#", "%23")):
                    continue
                a, local = norm(absu, ref)
                if not local:
                    external.append(f"css url() {ref} (in {absu})")
                    continue
                fetch_resource(a, absu)
    except Exception as e:
        failures.append(f"asset {absu} (on {referer}): {e}")


def crawl(start):
    queue = [start]
    while queue:
        url = defrag(queue.pop(0))
        if not url or url in seen_pages:
            continue
        try:
            status, body, final = fetch(url)
        except Exception as e:
            failures.append(f"page {url}: {e}")
            continue
        if status != 200:
            failures.append(f"page {url}: HTTP {status}")
            continue
        seen_pages.add(final)
        final = defrag(final)
        if not final.endswith((".html", ".htm", "/")):
            continue  # non-HTML subresource, nothing to parse
        text = body.decode("utf-8", "replace")
        c = Collector()
        c.feed(text)
        for ref in c.resources:
            absu, local = norm(final, ref)
            if not local:
                external.append(f"resource {ref} (on {final})")
                continue
            fetch_resource(absu, final)
        for ref in c.links:
            absu, local = norm(final, ref)
            if not local:
                external.append(f"link {ref} (on {final})")
                continue
            queue.append(absu)

    print(f"pages crawled : {len(seen_pages)}")
    for u in sorted(seen_pages):
        print(f"  {u}")
    print(f"assets fetched: {len(seen_res)}")
    for u in sorted(seen_res):
        print(f"  {u}")
    if external:
        print("\nNON-LOCAL REFERENCES:")
        for e in external:
            print(f"  {e}")
    if failures:
        print("\nFAILURES:")
        for f in failures:
            print(f"  {f}")
    if external or failures:
        sys.exit(1)
    print("\nOK: every request hit 127.0.0.1 only; all assets resolved 200")


def main():
    handler = lambda *a, **kw: QuietHandler(*a, directory=str(ROOT), **kw)
    socketserver.TCPServer.allow_reuse_address = True
    with socketserver.TCPServer(("127.0.0.1", PORT), handler) as httpd:
        port = httpd.server_address[1]  # resolves the ephemeral port
        threading.Thread(target=httpd.serve_forever, daemon=True).start()
        try:
            crawl(f"http://127.0.0.1:{port}/index.html")
        finally:
            httpd.shutdown()


if __name__ == "__main__":
    main()