path: root/webmentions_ssg/tasks/sender.py

import re
import uuid
from datetime import datetime, timezone
from urllib.parse import urljoin

import httpx
from bs4 import BeautifulSoup
from flask import current_app

from .. import APP_NAME, VERSION
from .. import DATABASE as db
from .. import HUEY as huey
from ..models import SentWebmention, SentWebmentionStatus
from ..url_security import AddressResolutionError, is_http_url, is_public_url

LINK_SPLIT = re.compile(r",\s*(?=<)")

DISCOVERY_HEADERS = {"Accept": "text/html, application/xhtml+xml;q=0.9"}


class SenderError(Exception):
    """A Webmention cannot be sent."""


class TemporarySenderError(SenderError):
    """A Webmention send failed for a potentially temporary reason."""


class PermanentSenderError(SenderError):
    """A Webmention send failed permanently for this source revision."""


def temporary_http_status(status: int) -> bool:
    """Return whether an HTTP status should be retried."""
    return status in {408, 425, 429} or 500 <= status <= 599


def ensure_public_request(request: httpx.Request) -> None:
    """Prevent requests to non-public network addresses."""
    try:
        if not is_public_url(str(request.url)):
            raise PermanentSenderError("Request resolves to a non-public address")
    except AddressResolutionError as exc:
        raise TemporarySenderError("Request hostname could not be resolved") from exc


def resolve_endpoint(response: httpx.Response, href: str) -> str:
    """Resolve and validate a discovered Webmention endpoint."""
    endpoint = urljoin(str(response.url), href.strip())

    if not is_http_url(endpoint):
        raise PermanentSenderError(f"Invalid Webmention endpoint: {endpoint!r}")

    return endpoint


def parse_link_value(value: str) -> tuple[str, set[str]] | None:
    """Parse a Link header value into its target and relations."""
    value = value.strip()

    if not value.startswith("<"):
        return None

    href, separator, parameters = value[1:].partition(">")

    if not separator:
        return None

    relations: set[str] = set()

    for parameter in parameters.split(";"):
        name, separator, value = parameter.partition("=")

        if separator and name.strip().lower() == "rel":
            relations.update(
                relation.lower() for relation in value.strip().strip("\"'").split()
            )

    return href, relations


def endpoint_from_headers(response: httpx.Response) -> str | None:
    """Return the first Webmention endpoint advertised by HTTP Link."""
    for header in response.headers.get_list("Link"):
        for value in LINK_SPLIT.split(header):
            if (link := parse_link_value(value)) is None:
                continue

            href, relations = link

            if "webmention" in relations:
                return resolve_endpoint(response, href)

    return None


def endpoint_from_html(response: httpx.Response, body: bytes) -> str | None:
    """Return the first HTML Webmention endpoint in document order."""
    document = BeautifulSoup(body, "html5lib")

    for element in document.find_all(["link", "a"], href=True):
        relations = element.get("rel")

        if isinstance(relations, str):
            relations = relations.split()

        if not relations:
            continue

        if not any(
            isinstance(relation, str) and relation.lower() == "webmention"
            for relation in relations
        ):
            continue

        href = element.get("href")

        if not isinstance(href, str):
            continue

        return resolve_endpoint(response, href)

    return None


def read_target_body(response: httpx.Response) -> bytes:
    """Read a target document up to the configured size limit."""
    max_bytes = current_app.config.get("WEBMENTIONS_SSG_MAX_TARGET_BYTES", 1_000_000)

    if (content_length := response.headers.get("Content-Length")) is not None:
        try:
            if int(content_length) > max_bytes:
                raise PermanentSenderError("Target document is too large")
        except ValueError:
            pass

    body = bytearray()

    for chunk in response.iter_bytes(chunk_size=64 * 1024):
        body.extend(chunk)

        if len(body) > max_bytes:
            raise PermanentSenderError("Target document is too large")

    return bytes(body)


def discover_webmention_endpoint(client: httpx.Client, target: str) -> str | None:
    """Discover the Webmention endpoint advertised by a target."""
    head_response = client.head(target, headers=DISCOVERY_HEADERS)

    if endpoint := endpoint_from_headers(head_response):
        return endpoint

    with client.stream("GET", target, headers=DISCOVERY_HEADERS) as response:
        if endpoint := endpoint_from_headers(response):
            return endpoint

        if temporary_http_status(response.status_code):
            raise TemporarySenderError(f"Target returned HTTP {response.status_code}")

        if not response.is_success:
            raise PermanentSenderError(f"Target returned HTTP {response.status_code}")

        media_type = (
            response.headers.get("Content-Type", "").partition(";")[0].strip().lower()
        )

        if media_type not in {"text/html", "application/xhtml+xml"}:
            return None

        return endpoint_from_html(response, read_target_body(response))


def post_webmention(
    client: httpx.Client, *, endpoint: str, source: str, target: str
) -> tuple[int, str | None]:
    """POST a Webmention and return its status code and status URL."""
    with client.stream(
        "POST", endpoint, data={"source": source, "target": target}
    ) as response:
        status_url = None

        if response.status_code == 201 and (
            location := response.headers.get("Location")
        ):
            status_url = urljoin(str(response.url), location)

        return (response.status_code, status_url)


def attempt_is_current(webmention: SentWebmention, revision: int) -> bool:
    """Return whether an attempt still represents the desired revision."""
    db.session.refresh(webmention)

    return webmention.desired_revision == revision and (
        webmention.processed_revision is None
        or webmention.processed_revision < revision
    )


@huey.task(retries=2, retry_delay=50)
def send_webmention(webmention_uuid: uuid.UUID) -> None:
    """Discover a receiver endpoint and send one Webmention."""
    webmention = db.session.get(SentWebmention, webmention_uuid)

    if webmention is None:
        current_app.logger.warning(
            "Cannot send unknown SentWebmention %s", webmention_uuid
        )
        return

    if (
        webmention.processed_revision is not None
        and webmention.processed_revision >= webmention.desired_revision
    ):
        return

    revision = webmention.desired_revision
    source = webmention.source.url
    target = webmention.target

    webmention.last_attempted_at = datetime.now(timezone.utc)
    webmention.endpoint = None
    webmention.response_status = None
    webmention.status_url = None

    db.session.commit()

    endpoint: str | None = None
    response_status: int | None = None

    try:
        with httpx.Client(
            headers={"User-Agent": (f"{APP_NAME}/{VERSION} SentWebmention")},
            timeout=httpx.Timeout(
                current_app.config.get("WEBMENTIONS_SSG_REQUEST_TIMEOUT", 5.0)
            ),
            follow_redirects=True,
            max_redirects=current_app.config.get("WEBMENTIONS_SSG_MAX_REDIRECTS", 20),
            trust_env=False,
            event_hooks={"request": [ensure_public_request]},
        ) as client:
            endpoint = discover_webmention_endpoint(client, target)

            if endpoint is None:
                if not attempt_is_current(webmention, revision):
                    return

                webmention.processed_revision = revision
                webmention.status = SentWebmentionStatus.UNSUPPORTED
                webmention.failure_reason = "No Webmention endpoint discovered"
                webmention.endpoint = None
                webmention.response_status = None

                db.session.commit()
                return

            (response_status, status_url) = post_webmention(
                client, endpoint=endpoint, source=source, target=target
            )

            if not (200 <= response_status <= 299):
                message = f"Webmention endpoint returned HTTP {response_status}"

                if temporary_http_status(response_status):
                    raise TemporarySenderError(message)

                raise PermanentSenderError(message)

    except (TemporarySenderError, httpx.RequestError) as exc:
        if not attempt_is_current(webmention, revision):
            return

        webmention.status = SentWebmentionStatus.FAILED
        webmention.failure_reason = str(exc) or "Webmention request failed"
        webmention.endpoint = endpoint
        webmention.response_status = response_status

        db.session.commit()

        # Leave processed_revision unchanged. Huey may retry the
        # attempt, and the periodic scanner will also rediscover
        # outstanding work.
        raise

    except PermanentSenderError as exc:
        if not attempt_is_current(webmention, revision):
            return

        webmention.processed_revision = revision
        webmention.status = SentWebmentionStatus.FAILED
        webmention.failure_reason = str(exc)
        webmention.endpoint = endpoint
        webmention.response_status = response_status

        db.session.commit()
        return

    if not attempt_is_current(webmention, revision):
        return

    webmention.processed_revision = revision
    webmention.sent_revision = revision

    webmention.status = SentWebmentionStatus.SENT
    webmention.failure_reason = None

    webmention.endpoint = endpoint
    webmention.response_status = response_status
    webmention.status_url = status_url

    webmention.last_sent_at = datetime.now(timezone.utc)

    db.session.commit()