1 files changed, 397 insertions, 0 deletions
diff --git a/webmentions_ssg/tasks/scanner.py b/webmentions_ssg/tasks/scanner.py
new file mode 100644
index 0000000..f80e5f4
--- /dev/null
+++ b/webmentions_ssg/tasks/scanner.py
@@ -0,0 +1,397 @@
+from collections.abc import Iterator
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from fnmatch import fnmatchcase
+from itertools import chain
+from pathlib import Path
+from typing import Any
+from urllib.parse import quote, urldefrag, urljoin, urlsplit
+
+import mf2py
+import sqlalchemy as sa
+import xxhash
+from bs4 import BeautifulSoup, Tag
+from flask import current_app
+from sqlalchemy.orm import selectinload
+
+from .. import DATABASE as db
+from .. import HUEY as huey
+from ..models import SentWebmention, Source
+from ..url_security import is_http_url
+from .sender import send_webmention
+
+REACTION_PROPERTIES = ("in-reply-to", "like-of", "repost-of", "bookmark-of")
+
+
+class SourceScanError(Exception):
+    """A source document cannot safely be processed."""
+
+
+@dataclass(frozen=True)
+class ScannedSource:
+    path: str
+    url: str
+    content_hash: str
+    targets: frozenset[str]
+
+
+def source_url_for_path(relative_path: Path, base_url: str) -> str:
+    """Derive the public source URL from its relative filesystem path."""
+    directory = relative_path.parent.as_posix()
+
+    return urljoin(base_url, f"{quote(directory, safe='/')}/")
+
+
+def canonical_url(
+    document: BeautifulSoup, *, relative_path: Path, base_url: str | None
+) -> str | None:
+    """Return the canonical URL declared by the document."""
+    if (link := document.select_one('link[rel~="canonical"][href]')) is not None:
+        href = link.get("href")
+
+        if not isinstance(href, str) or not (href := href.strip()):
+            raise SourceScanError("Document contains an empty canonical URL")
+
+        if is_http_url(href):
+            return href
+
+        if base_url is None:
+            raise SourceScanError(
+                "Document contains a relative canonical URL, but "
+                "WEBMENTIONS_SSG_SOURCE_BASE_URL is not configured"
+            )
+
+        resolved = urljoin(source_url_for_path(relative_path, base_url), href)
+
+        if not is_http_url(resolved):
+            raise SourceScanError(
+                f"Canonical URL is not a valid HTTP or HTTPS URL: {href!r}"
+            )
+
+        return resolved
+
+
+def property_urls(entry: dict[str, Any], property_name: str) -> Iterator[str]:
+    """Yield URL values from a microformats property."""
+    properties = entry.get("properties")
+
+    if not isinstance(properties, dict):
+        return
+
+    values = properties.get(property_name)
+
+    if not isinstance(values, list):
+        return
+
+    yield from (value for value in values if isinstance(value, str))
+
+
+def parse_entry(element: Tag, base_url: str) -> dict[str, Any]:
+    """Parse the source h-entry with mf2py."""
+    parsed = mf2py.parse(doc=str(element), url=base_url)
+
+    if not isinstance(parsed, dict):
+        raise SourceScanError("Microformats parser did not return a document object")
+
+    items = parsed.get("items")
+
+    if not isinstance(items, list) or len(items) != 1:
+        raise SourceScanError("Expected exactly one parsed h-entry")
+
+    entry = items[0]
+
+    if not isinstance(entry, dict):
+        raise SourceScanError("Parsed h-entry is not an object")
+
+    types = entry.get("type")
+
+    if not isinstance(types, list) or "h-entry" not in types:
+        raise SourceScanError("Parsed microformats item is not an h-entry")
+
+    return entry
+
+
+def primary_entry(
+    document: BeautifulSoup, source_url: str
+) -> tuple[Tag, dict[str, Any]]:
+    """Return and validate the source h-entry."""
+    entries = document.find_all(class_="h-entry")
+
+    if len(entries) != 1:
+        raise SourceScanError(f"Expected exactly one h-entry, found {len(entries)}")
+
+    entry_element = entries[0]
+    mf2_entry = parse_entry(entry_element, source_url)
+
+    urls = tuple(property_urls(mf2_entry, "url"))
+
+    if source_url not in urls:
+        raise SourceScanError(
+            f"The h-entry u-url does not match the source URL {source_url!r}: {urls!r}"
+        )
+
+    return entry_element, mf2_entry
+
+
+def content_element(entry: Tag) -> Tag:
+    """Return the source h-entry's e-content element."""
+    contents = entry.find_all(class_="e-content")
+
+    if len(contents) != 1:
+        raise SourceScanError(f"Expected exactly one e-content, found {len(contents)}")
+
+    return contents[0]
+
+
+def normalize_target(value: str, *, base_url: str, source_url: str) -> str | None:
+    """Resolve and validate a possible Webmention target URL."""
+    value = value.strip()
+
+    if not value:
+        return None
+
+    target = urljoin(base_url, value)
+
+    if not is_http_url(target):
+        return None
+
+    if urldefrag(target)[0] == urldefrag(source_url)[0]:
+        return None
+
+    return target
+
+
+def iter_targets(
+    content: Tag,
+    mf2_entry: dict[str, Any],
+    *,
+    base_url: str,
+    source_url: str,
+    ignored_hostnames: tuple[str, ...] = (),
+) -> Iterator[str]:
+    """Yield outgoing Webmention targets from an h-entry."""
+    hrefs = (
+        href
+        for element in content.find_all("a", href=True)
+        if isinstance(href := element.get("href"), str)
+    )
+
+    reactions = chain.from_iterable(
+        property_urls(mf2_entry, property_name) for property_name in REACTION_PROPERTIES
+    )
+
+    for value in chain(hrefs, reactions):
+        target = normalize_target(value, base_url=base_url, source_url=source_url)
+        if target is not None:
+            hostname = urlsplit(target).hostname or ""
+            if not any(fnmatchcase(hostname, pattern) for pattern in ignored_hostnames):
+                yield target
+
+
+def scan_source_file(
+    path: Path,
+    *,
+    root: Path,
+    base_url: str | None,
+    ignored_hostnames: tuple[str, ...] = (),
+) -> ScannedSource:
+    """Parse one generated source document."""
+    document = BeautifulSoup(path.read_bytes(), "html5lib")
+
+    relative_path = path.relative_to(root)
+
+    source_url = canonical_url(document, relative_path=relative_path, base_url=base_url)
+
+    if source_url is None:
+        if base_url is None:
+            raise SourceScanError(
+                "Document has no canonical URL and "
+                "WEBMENTIONS_SSG_SOURCE_BASE_URL is not configured"
+            )
+
+        source_url = source_url_for_path(relative_path, base_url)
+
+    entry_element, mf2_entry = primary_entry(document, source_url)
+
+    content = content_element(entry_element)
+
+    return ScannedSource(
+        path=relative_path.as_posix(),
+        url=source_url,
+        content_hash=xxhash.xxh3_128_hexdigest(str(entry_element).encode("utf-8")),
+        targets=frozenset(
+            iter_targets(
+                content,
+                mf2_entry,
+                base_url=source_url,
+                source_url=source_url,
+                ignored_hostnames=ignored_hostnames,
+            )
+        ),
+    )
+
+
+def create_source(scanned: ScannedSource, scan_time: datetime) -> Source:
+    """Create a source from a newly discovered document."""
+    source = Source(
+        path=scanned.path,
+        url=scanned.url,
+        content_hash=scanned.content_hash,
+        revision=1,
+        last_seen_at=scan_time,
+        revised_at=scan_time,
+        deleted_at=None,
+    )
+
+    for target in scanned.targets:
+        source.sent_webmentions.append(
+            SentWebmention(target=target, active=True, desired_revision=1)
+        )
+
+    return source
+
+
+def update_source(
+    source: Source, scanned: ScannedSource, scan_time: datetime
+) -> Source:
+    """Update a source from a newly scanned revision."""
+    if source.url != scanned.url:
+        raise SourceScanError(
+            f"Source path {source.path!r} changed public URL "
+            f"from {source.url!r} to {scanned.url!r}"
+        )
+
+    source.last_seen_at = scan_time
+
+    if source.deleted_at is None and source.content_hash == scanned.content_hash:
+        return source
+
+    source.revision += 1
+    source.content_hash = scanned.content_hash
+    source.revised_at = scan_time
+    source.deleted_at = None
+
+    existing = {webmention.target: webmention for webmention in source.sent_webmentions}
+
+    for webmention in existing.values():
+        webmention.active = webmention.target in scanned.targets
+        webmention.desired_revision = source.revision
+
+        if not webmention.active and webmention.sent_revision is None:
+            webmention.processed_revision = source.revision
+
+    for target in scanned.targets:
+        if target not in existing:
+            source.sent_webmentions.append(
+                SentWebmention(
+                    target=target, active=True, desired_revision=source.revision
+                )
+            )
+
+    return source
+
+
+@huey.lock_task("scan-webmention-sources")
+def scan_sources() -> None:
+    """Scan generated source documents and queue pending Webmentions."""
+    directory = current_app.config.get("WEBMENTIONS_SSG_SOURCE_DIRECTORY")
+
+    if directory is None:
+        return
+
+    root = Path(directory).expanduser().resolve(strict=True)
+
+    if not root.is_dir():
+        raise RuntimeError(f"Source directory is not a directory: {root}")
+
+    base_url = current_app.config.get("WEBMENTIONS_SSG_SOURCE_BASE_URL")
+
+    if base_url is not None:
+        if not is_http_url(base_url):
+            raise RuntimeError(
+                "WEBMENTIONS_SSG_SOURCE_BASE_URL must be an absolute HTTP or HTTPS URL"
+            )
+        base_url = f"{base_url.rstrip('/')}/"
+
+    ignored_hostnames = tuple(
+        pattern.lower().rstrip(".")
+        for pattern in current_app.config.get("WEBMENTIONS_SSG_IGNORED_HOSTNAMES", ())
+    )
+
+    sources_by_path = {
+        source.path: source
+        for source in db.session.scalars(
+            sa.select(Source).options(selectinload(Source.sent_webmentions))
+        )
+    }
+
+    seen_paths: set[str] = set()
+    scanned_count = 0
+    scan_time = datetime.now(timezone.utc)
+
+    for index_path in root.rglob("index.html"):
+        relative_path = index_path.relative_to(root)
+
+        if relative_path.parent == Path("."):
+            continue
+
+        seen_paths.add(relative_path.as_posix())
+
+        try:
+            scanned = scan_source_file(
+                index_path,
+                root=root,
+                base_url=base_url,
+                ignored_hostnames=ignored_hostnames,
+            )
+
+            if (source := sources_by_path.get(scanned.path)) is None:
+                source = create_source(scanned, scan_time)
+                db.session.add(source)
+            else:
+                source = update_source(source, scanned, scan_time)
+
+            sources_by_path[source.path] = source
+            scanned_count += 1
+
+        except (OSError, SourceScanError) as exc:
+            current_app.logger.warning("Could not scan source %s: %s", index_path, exc)
+
+    for path, source in sources_by_path.items():
+        if source.deleted_at is not None or path in seen_paths:
+            continue
+
+        source.revision += 1
+        source.revised_at = scan_time
+        source.deleted_at = scan_time
+
+        for webmention in source.sent_webmentions:
+            webmention.active = False
+            webmention.desired_revision = source.revision
+
+            if webmention.sent_revision is None:
+                webmention.processed_revision = source.revision
+
+    # Persist the desired state before queueing any work.
+    db.session.commit()
+
+    pending_count = 0
+
+    for identifier in db.session.scalars(
+        sa.select(SentWebmention.uuid)
+        .where(
+            sa.or_(
+                SentWebmention.processed_revision.is_(None),
+                SentWebmention.processed_revision < SentWebmention.desired_revision,
+            )
+        )
+        .order_by(SentWebmention.uuid)
+    ):
+        send_webmention(identifier)
+        pending_count += 1
+
+    current_app.logger.info(
+        "Webmention source scan complete: %d sources scanned, %d pending sends",
+        scanned_count,
+        pending_count,
+    )