"""Feed parser — fetch and parse RSS, Atom, and JSON feeds.
Returns normalized feed items with: id, title, link, summary, content,
content_link, author, date, and raw data for template access.
"""
import re
import calendar
import hashlib
import html
import ipaddress
import json
import logging
import socket
import threading
from html.parser import HTMLParser
import httpx
import nh3
import feedparser
from datetime import datetime, timezone, timedelta
from urllib.parse import urlencode, urljoin, urlparse
import settings
logger = logging.getLogger("feedecho.feed_parser")
USER_AGENT = "feedecho/1.0 (+https://github.com/yourusername/feedecho)"
MAX_FEED_SIZE = 10 * 1024 * 1024 # 10 MB cap
MAX_REDIRECTS = 5
MAX_IMAGE_SIZE = 10 * 1024 * 1024 # 10 MB cap, matches Mastodon's default limits
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp", "image/avif"}
class SSRFError(ValueError):
"""Raised when a URL points to a private or internal address."""
# 100.64.0.0/10 (RFC 6598 shared address space) is not is_private, is_loopback,
# is_link_local, is_reserved, is_multicast, or is_unspecified, so it slips
# through the explicit checks below — but it is used for internal service
# addressing by cloud providers, Kubernetes CNIs, and Tailscale. Block it.
#
# NOTE: do NOT replace the checks with `not ip.is_global`. On Python 3.14
# is_global returns True for multicast (224.0.0.0/4, ff00::/8) and NAT64
# (64:ff9b::/96), so `not is_global` would ALLOW those ranges that the
# explicit checks below block — a regression. The explicit checks plus the
# CGNAT range below is the strictly-safe union.
_CGNAT_V4 = ipaddress.ip_network("100.64.0.0/10")
def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
"""Check if an IP address is private, loopback, link-local, or reserved."""
return (
ip.is_private
or ip.is_loopback
or ip.is_link_local
or ip.is_reserved
or ip.is_multicast
or ip.is_unspecified
or (ip.version == 4 and ip in _CGNAT_V4)
)
def validate_outbound_url(url: str) -> str:
"""Validate that a URL is safe for server-side fetching (SSRF protection).
Blocks:
- Non-http(s) schemes (file://, gopher://, etc.)
- Userinfo in URL (user:pass@host)
- Hostnames that resolve to private/internal IPs
- Direct IP addresses in private ranges (10.x, 172.16-31.x, 192.168.x,
127.x, 169.254.x, ::1, fc00::, fe80::)
Raises SSRFError if the URL is unsafe. Returns the URL if safe.
NOTE: validation alone is TOCTOU-vulnerable (the DNS answer checked here
is not the connection httpx later makes). Callers that actually fetch
must go through ``ssrf_client()`` so the validated IP is pinned; see
``PinningNetworkBackend``.
"""
parsed = urlparse(url)
if parsed.scheme not in ("http", "https"):
raise SSRFError(f"Blocked: URL scheme '{parsed.scheme}' is not allowed")
if parsed.username or parsed.password:
raise SSRFError("Blocked: URLs with embedded credentials are not allowed")
hostname = parsed.hostname
if not hostname:
raise SSRFError("Blocked: URL has no hostname")
# If it's a literal IP, check directly
is_ip = False
ip = None
try:
ip = ipaddress.ip_address(hostname)
is_ip = True
except ValueError:
pass
if is_ip and ip is not None:
if _is_blocked_ip(ip):
raise SSRFError(f"Blocked: IP address {ip} is in a private/reserved range")
else:
# Not a literal IP — resolve the hostname and check all results
try:
infos = socket.getaddrinfo(hostname, None)
except socket.gaierror:
raise SSRFError(f"Blocked: cannot resolve hostname '{hostname}'")
for family, _, _, _, sockaddr in infos:
addr_str = sockaddr[0]
try:
ip = ipaddress.ip_address(addr_str)
except ValueError:
continue
if _is_blocked_ip(ip):
raise SSRFError(
f"Blocked: hostname '{hostname}' resolves to "
f"private/reserved IP {ip}"
)
return url
# ── DNS-rebinding protection: validated-IP pinning (B2) ─────────────────────
#
# validate_outbound_url checks the DNS answer at call time, but httpx resolves
# the hostname AGAIN when it opens the connection. A low-TTL DNS answer (or an
# attacker-controlled authoritative server) can hand back a different IP
# between the two lookups, and every redirect hop re-resolves too. That gap
# turns the validator into theatre on hostile DNS.
#
# The fix: validate_outbound_url returns the addresses it approved, and the
# fetch path connects to one of those exact addresses. The backend below pins
# per-hostname; TLS is unaffected because httpcore derives SNI and the
# certificate identity from the URL's hostname, not from the IP we dial — a
# pinned-but-hostile IP simply fails certificate verification, and a
# same-CDN address swap to another site behind the same cert is the residual
# risk documented in HANDOFF.
def _hostname_pin_key(hostname: str) -> str:
"""The hostname form httpcore passes to NetworkBackend.connect_tcp.
httpcore dials the IDNA/punycode form of the URL host, while
urlparse().hostname yields the unicode form. Keying the pin map on the
punycode form makes both sides agree (ASCII hostnames are their own
punycode form, so this is a no-op for them).
"""
return hostname.encode("idna").decode("ascii").lower()
def _resolve_addrs(hostname: str) -> list[str]:
"""Resolve a hostname to its IP addresses (both families, best effort)."""
addrs: list[str] = []
try:
infos = socket.getaddrinfo(hostname, None)
except socket.gaierror:
return addrs
for family, _, _, _, sockaddr in infos:
try:
addrs.append(str(ipaddress.ip_address(sockaddr[0])))
except ValueError:
continue
return addrs
def _pick_pinned_addr(addrs: list[str]) -> str:
"""Choose one address to dial: the first IPv4, else the first entry.
Preferring IPv4 keeps the door open for hosts that publish an A record
and an unreachable AAAA record; when only IPv6 exists we take the first.
"""
for addr in addrs:
if isinstance(ipaddress.ip_address(addr), ipaddress.IPv4Address):
return addr
return addrs[0]
class PinningNetworkBackend:
"""httpcore NetworkBackend that dials pre-validated addresses.
Wraps httpcore's default SyncBackend. ``pins`` maps the IDNA hostname to
the exact IP approved by validate_outbound_url; connect_tcp dials the
pinned address instead of resolving DNS again. Unknown hosts (e.g. from
code paths that bypass ssrf_client) fall through to normal resolution —
this backend is defense-in-depth for our fetches, not a system-wide VPN.
"""
def __init__(self) -> None:
import httpcore
self._inner = httpcore.SyncBackend()
self._pins: dict[str, str] = {}
self._lock = threading.Lock()
def set_pin(self, hostname: str, addr: str) -> None:
key = _hostname_pin_key(hostname)
with self._lock:
self._pins[key] = addr
def clear_pins(self) -> None:
with self._lock:
self._pins.clear()
def connect_tcp(self, host, port, timeout=None, local_address=None, socket_options=None):
with self._lock:
pinned = self._pins.get(host)
if not pinned:
# Fail closed: an SSRF guard that silently falls through to live
# DNS on a pin miss is no guard at all. Every outbound host must
# be explicitly pinned via ssrf_client before it is dialed.
raise ConnectionError(
f"SSRF pin miss for {host!r} — host was not pre-validated"
)
# Dial the pinned address; SNI and cert identity still come from the
# URL hostname because httpcore passes them independently of this arg.
return self._inner.connect_tcp(
pinned, port, timeout=timeout,
local_address=local_address, socket_options=socket_options,
)
def connect_unix_socket(self, *args, **kwargs):
return self._inner.connect_unix_socket(*args, **kwargs)
def sleep(self, seconds: float) -> None:
self._inner.sleep(seconds)
def _pins_for_urls(urls: list[str]) -> dict[str, str]:
"""Validate each URL and return {pin_key: dial_addr} for all of them.
Idempotent per URL, so callers can pre-validate the whole redirect set
they are willing to follow, or just the first URL.
"""
pins: dict[str, str] = {}
for url in urls:
parsed = urlparse(url)
hostname = parsed.hostname
if not hostname:
continue
try:
ipaddress.ip_address(hostname)
key = hostname.lower()
except ValueError:
key = _hostname_pin_key(hostname)
addrs = _resolve_addrs(hostname)
if not addrs:
raise SSRFError(f"Blocked: cannot resolve hostname '{hostname}'")
# _pins_for_urls is the SOLE authority on which address gets pinned:
# it resolves and checks independently of validate_outbound_url. Do
# NOT refactor it to trust the validator's result — that reopens the
# TOCTOU window (two independent resolutions, one for validation and
# one for pinning).
for addr in addrs:
if _is_blocked_ip(ipaddress.ip_address(addr)):
raise SSRFError(
f"Blocked: hostname '{hostname}' resolves to "
f"private/reserved IP {addr}"
)
pins[key] = _pick_pinned_addr(addrs)
return pins
def _ssl_context(no_alpn: bool):
"""Build the SSL context for the pinned connection pool.
With ``no_alpn=True`` the context silently ignores ``set_alpn_protocols``.
httpcore always advertises ALPN ``http/1.1`` at connect time
(httpcore/_sync/connection.py), and some edge proxies — notably Tumblr's
custom-domain frontend — drop TLS connections that advertise ALPN
``http/1.1`` (server closes without a response) and reset streams when
HTTP/2 is negotiated instead. Omitting the ALPN extension entirely makes
such edges serve plain HTTP/1.1. Verified safe against Cloudflare,
nginx, and GitHub edges, which all default to HTTP/1.1 without ALPN.
The no-ALPN variant is the httpx context with only its class replaced,
so verify flags, trust store, and trust_env (SSL_CERT_FILE/DIR) handling
stay byte-for-byte identical to the ALPN-advertising path.
"""
from httpx._config import create_ssl_context
ctx = create_ssl_context(verify=True, cert=None, trust_env=True)
if not no_alpn:
return ctx
import ssl
class _NoAlpnSSLContext(ssl.SSLContext):
def set_alpn_protocols(self, alpn_protocols):
return None
ctx.__class__ = _NoAlpnSSLContext
return ctx
def ssrf_client(
urls: list[str],
*,
timeout: float = 30,
no_alpn: bool = False,
) -> tuple[httpx.Client, PinningNetworkBackend]:
"""Build an httpx.Client whose connections are pinned to validated IPs.
Every URL in ``urls`` is validated and its address pinned for the life
of the client. Returns the client and its pinning backend, so callers
(the redirect-validating fetch helper) can pin additional hops as they
validate them. Callers MUST close the client when done (try/finally);
that releases the pool and its connections.
With ``no_alpn=True`` connections are dialed without advertising ALPN;
see ``_ssl_context`` for why some edge proxies require this.
"""
import httpcore
pins = _pins_for_urls(urls)
backend = PinningNetworkBackend()
for key, addr in pins.items():
backend.set_pin(key, addr)
pool = httpcore.ConnectionPool(
ssl_context=_ssl_context(no_alpn),
network_backend=backend,
)
class _PinnedTransport(httpx.BaseTransport):
def handle_request(self, request):
req = httpcore.Request(
method=request.method,
url=httpcore.URL(
scheme=request.url.raw_scheme,
host=request.url.raw_host,
port=request.url.port,
target=request.url.raw_path,
),
headers=request.headers.raw,
content=request.stream,
extensions=request.extensions,
)
with httpx._transports.default.map_httpcore_exceptions():
resp = pool.handle_request(req)
return httpx.Response(
resp.status,
headers=resp.headers,
stream=httpx._transports.default.ResponseStream(resp.stream),
extensions=resp.extensions,
)
def close(self):
# httpx.BaseTransport has no close hook; without this the
# ConnectionPool (and its keep-alive sockets/FDs) leak on every
# client.close() in the long-running scheduler worker.
pool.close()
client = httpx.Client(
transport=_PinnedTransport(),
follow_redirects=False,
timeout=timeout,
)
return client, backend
# Backwards-compatible alias
validate_feed_url = validate_outbound_url
def unpinned_client(*, timeout: float = 30) -> httpx.Client:
"""A plain, unpinned httpx.Client for SINGLE-MODE only.
Single mode has no privilege boundary: the operator owns both the app
and any outbound address (LAN vision servers, local webhook receivers),
so SSRF validation/pinning would only break legitimate self-hosted
setups. Hosted (multi) code must NEVER use this — every outbound request
there goes through pinned_request()/ssrf_client() so the validated IP is
the one dialed.
Kept as one named function so the test that forbids bare ``httpx.Client``
outside feed_parser.py pins the machinery's surface to this file.
"""
return httpx.Client(timeout=timeout, follow_redirects=False)
# Headers that must never follow a cross-origin redirect. httpx strips these
# when the redirect target's origin differs; a hand-rolled loop must too, or a
# hostile homeserver/PDS can bounce a request — and the tenant's Bearer token —
# to an arbitrary public address it chose.
_CREDENTIAL_HEADERS = frozenset({"authorization", "cookie", "proxy-authorization"})
# Headers that describe a request body; dropped when a redirect downgrades the
# method to GET (mirrors httpx's redirect handling).
_ENTITY_HEADERS = frozenset({"content-type", "content-length"})
def _strip_headers(headers: dict | None, drop: frozenset) -> dict | None:
"""Return ``headers`` with the named (case-insensitive) headers removed."""
if not headers:
return headers
return {k: v for k, v in headers.items() if k.lower() not in drop}
def _url_origin(url: str) -> tuple:
u = httpx.URL(url)
return (u.scheme, u.host, u.port)
def pinned_request(
method: str,
url: str,
*,
timeout: float = 30,
headers: dict | None = None,
follow_redirects: bool = False,
**kwargs,
) -> httpx.Response:
"""One outbound HTTP request through the SSRF-pinned transport.
The URL's hostname is resolved once, validated, and its IP pinned for
the life of the request, so the connection dials the exact address that
was approved (closing the DNS-rebinding TOCTOU window that a
validate-then-fetch-anyway call leaves open). The client is created and
closed inside this call.
When ``follow_redirects`` is True, every redirect hop is re-validated
and re-pinned before being followed, so a redirect to an internal
address is refused instead of dialed. Redirect semantics match httpx:
303 -> GET; 301/302 on a POST -> GET with the body dropped; 307/308
preserve method and body. Credential headers (Authorization, Cookie,
Proxy-Authorization) are stripped on cross-origin hops, and entity
headers on GET downgrades, exactly as httpx's own redirect handling.
Raises SSRFError if the URL (or any followed redirect hop) is unsafe or
unresolvable. Returns the final (non-redirect) response.
"""
client, backend = ssrf_client([url], timeout=timeout)
try:
current_url = url
current_method = method.upper()
current_kwargs = dict(kwargs)
hop_headers = headers
for _ in range(MAX_REDIRECTS + 1):
resp = client.request(
current_method, current_url, headers=hop_headers, **current_kwargs
)
if not follow_redirects or not resp.is_redirect:
return resp
location = resp.headers.get("location")
if not location:
# A redirect with no Location is not followable; hand it back
# and let the caller's status handling report it.
return resp
next_url = str(httpx.URL(current_url).join(location))
validate_outbound_url(next_url)
for key, addr in _pins_for_urls([next_url]).items():
backend.set_pin(key, addr)
if _url_origin(current_url) != _url_origin(next_url):
# Cross-origin: drop credentials before re-issuing (from the
# already-stripped hop headers, so a later re-cross keeps
# them stripped).
hop_headers = _strip_headers(hop_headers, _CREDENTIAL_HEADERS)
current_url = next_url
if resp.status_code == 303:
current_method = "GET"
current_kwargs = {}
hop_headers = _strip_headers(hop_headers, _ENTITY_HEADERS)
elif resp.status_code in (301, 302) and current_method == "POST":
current_method = "GET"
current_kwargs = {}
hop_headers = _strip_headers(hop_headers, _ENTITY_HEADERS)
raise SSRFError(f"Too many redirects (max {MAX_REDIRECTS}) for {url}")
finally:
client.close()
def fetch_feed(
url: str,
*,
etag: str | None = None,
last_modified: str | None = None,
) -> dict:
"""Fetch and parse a feed URL. Returns dict with feed metadata and items.
Validates the initial URL and every redirect hop for SSRF protection,
pinning each hop's connection to the IP validated for that hop (B2: the
second DNS lookup a naive fetch performs is the rebinding window).
When ``etag`` / ``last_modified`` are supplied they are sent as
``If-None-Match`` / ``If-Modified-Since``. A 304 then returns
``{"not_modified": True, "items": [], ...}`` with the current validators
echoed back, and a 200 returns the parsed feed with ``etag`` and
``last_modified`` keys carrying the fresh validators (or None). Without
validators the fetch is unconditional and the result carries ``etag`` /
``last_modified`` from the response for the caller to store.
Raises SSRFError if any URL (initial or redirect) points to a
private/internal address.
Raises httpx.HTTPError on network failure.
"""
validate_outbound_url(url)
headers = {"User-Agent": USER_AGENT}
if etag:
headers["If-None-Match"] = etag
if last_modified:
headers["If-Modified-Since"] = last_modified
meta: dict = {}
try:
client, backend = ssrf_client([url])
try:
content, content_type, meta = _fetch_with_redirect_validation(
client, url, headers, MAX_FEED_SIZE, backend=backend
)
finally:
client.close()
except httpx.RemoteProtocolError:
# Some edge proxies (notably Tumblr's custom-domain frontend) drop
# TLS connections that advertise ALPN "http/1.1" — the server closes
# without a response — and reset streams when HTTP/2 is negotiated
# instead. httpcore always advertises ALPN http/1.1, so retry once
# with ALPN suppressed; the edge then serves plain HTTP/1.1.
client, backend = ssrf_client([url], no_alpn=True)
try:
content, content_type, meta = _fetch_with_redirect_validation(
client, url, headers, MAX_FEED_SIZE, backend=backend
)
finally:
client.close()
except httpx.HTTPStatusError as exc:
if exc.response.status_code in (403, 429) and settings.FALLBACK_PROXY_URL:
# The fallback proxy fetches the origin itself; its response
# headers describe the proxy, not the origin, so conditional
# validators must not be sent through it nor stored from it.
proxy_headers = {
k: v for k, v in headers.items()
if k.lower() not in ("if-none-match", "if-modified-since")
}
content, content_type, _ = _fetch_via_fallback_proxy(
url, proxy_headers, MAX_FEED_SIZE
)
meta = {}
else:
raise
# A 304 means the feed is byte-for-byte the representation our stored
# validators already describe: no parsing, no items, keep the validators
# (preferring any fresh ones the response carried).
if meta.get("status") == 304:
return {
"not_modified": True,
"title": "",
"url": url,
"type": "rss",
"items": [],
"etag": meta.get("etag") or etag,
"last_modified": meta.get("last_modified") or last_modified,
}
# JSON Feed. The path (not the full URL) decides: query strings like
# /feed.json?token=... are common on private feeds, and such URLs often
# also omit a JSON content-type, so both signals must use the path.
if "json" in content_type or urlparse(url).path.endswith(".json"):
result = parse_json_feed(json.loads(content))
else:
# RSS/Atom via feedparser
parsed = feedparser.parse(content)
result = parse_rss_feed(parsed, url)
result["etag"] = meta.get("etag")
result["last_modified"] = meta.get("last_modified")
return result
def _fetch_with_redirect_validation(
client: httpx.Client,
url: str,
headers: dict,
max_bytes: int = MAX_FEED_SIZE,
backend: "PinningNetworkBackend | None" = None,
) -> tuple[bytes, str, dict]:
"""Fetch a URL with a hard size cap, validating every redirect hop.
Prevents SSRF via redirect: an attacker can host a public feed that
redirects to an internal IP. This function validates every Location
header before following it, and (when a pinning backend is supplied)
pins each validated hop's hostname to its validated address, closing
the rebinding window on every hop, not just the first.
The body is streamed and abandoned as soon as it exceeds ``max_bytes``.
Checking the size after a buffered read (the previous behaviour) meant a
hostile feed or image could make the worker hold the whole body in memory
before the cap rejected it.
Returns (body, content-type, meta) where meta carries the final hop's
``status``, ``etag``, and ``last_modified`` response headers. Raises
ValueError when the cap is exceeded.
"""
for _ in range(MAX_REDIRECTS + 1):
with client.stream("GET", url, headers=headers) as response:
# httpx treats every 3xx as is_redirect (304 included), but a 304
# has no Location header and is NOT a hop we can follow. It must
# fall through to the meta capture below so the caller can
# short-circuit on "not modified".
if response.is_redirect and response.status_code != 304:
location = response.headers.get("location")
if not location:
raise ValueError("Redirect response had no Location header")
# Resolve relative redirects against the current URL
next_url = str(httpx.URL(url).join(location))
validate_outbound_url(next_url)
if backend is not None:
for key, addr in _pins_for_urls([next_url]).items():
backend.set_pin(key, addr)
url = next_url
continue
# raise_for_status() raises for ANY non-2xx, including 304 (a 3xx
# is not is_success), so skip it explicitly for 304: a not-modified
# response is a success, not an error, and carries no body.
if response.status_code != 304:
response.raise_for_status()
declared = response.headers.get("content-length", "")
if declared.isdigit() and int(declared) > max_bytes:
raise ValueError(
f"Response too large: {declared} bytes (max {max_bytes})"
)
chunks: list[bytes] = []
total = 0
for chunk in response.iter_bytes():
total += len(chunk)
if total > max_bytes:
raise ValueError(
f"Response too large: exceeded {max_bytes} bytes"
)
chunks.append(chunk)
meta = {
"status": response.status_code,
"etag": response.headers.get("etag"),
"last_modified": response.headers.get("last-modified"),
# Feed autodiscovery resolves relative values
# against the FINAL hop's URL, not the requested one; other
# callers simply ignore the extra key.
"final_url": url,
}
return b"".join(chunks), response.headers.get("content-type", ""), meta
raise ValueError(f"Too many redirects (max {MAX_REDIRECTS})")
def _fetch_via_fallback_proxy(
url: str,
headers: dict,
max_bytes: int = MAX_FEED_SIZE,
timeout: float = 30,
) -> tuple[bytes, str, dict]:
"""Fetch an outbound URL through the configured fallback proxy worker.
Triggered when a direct fetch receives HTTP 403 or 429 (e.g. edge WAF blocks
targeting datacenter IP subnets). The proxy URL is contacted via ssrf_client
(so its own host IP is validated/pinned).
"""
proxy_base = settings.FALLBACK_PROXY_URL
if not proxy_base:
raise ValueError("Fallback proxy URL is not configured")
proxy_url = f"{proxy_base.rstrip('/')}/?{urlencode({'url': url})}"
validate_outbound_url(proxy_url)
req_headers = dict(headers)
if settings.FALLBACK_PROXY_SECRET:
req_headers["X-FeedEcho-Proxy-Secret"] = settings.FALLBACK_PROXY_SECRET
client, backend = ssrf_client([proxy_url], timeout=timeout)
try:
return _fetch_with_redirect_validation(
client, proxy_url, req_headers, max_bytes=max_bytes, backend=backend
)
finally:
client.close()
# ── Feed autodiscovery (F4) ──────────────────────────────────────────────────
_FEED_LINK_TYPES = (
"application/rss+xml",
"application/atom+xml",
"application/feed+json",
)
class _DiscoveryHTMLParser(HTMLParser):
"""Collect candidates and the page .
Mirrors the module's small-local-parser style: no dependencies, no error
recovery — a malformed page simply yields what it yields. With
``convert_charrefs=True`` attribute values arrive entity-decoded, so a
href like ``/feed.xml?utm=1&x=2`` is parsed correctly.
"""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.links: list[tuple[str, str, str]] = [] # (rel, type, href)
self.title = ""
self._in_title = False
# The FIRST <title> in the document is the page title; later ones
# (inline SVG <title>, noscript fallbacks) must not append to it.
self._title_done = False
def handle_starttag(self, tag, attrs):
if tag == "link":
attr = dict(attrs)
self.links.append(
(
attr.get("rel") or "",
attr.get("type") or "",
attr.get("href") or "",
)
)
elif tag == "title" and not self._title_done:
self._in_title = True
def handle_endtag(self, tag):
if tag == "title" and self._in_title:
self._in_title = False
self._title_done = True
def handle_data(self, data):
if self._in_title:
self.title += data
def discover_feed_url(url: str, max_page_bytes: int = 2 * 1024 * 1024) -> dict:
"""Resolve a page URL to its feed URL via ``<link rel="alternate">``.
When the input URL itself is a feed (non-HTML content-type), it is
echoed back unchanged. Returns ``{"found", "url", "title", "alternates"}``
— ``alternates`` lists every valid feed link in document order (best
first), de-duplicated by URL with a trailing slash stripped.
Raises SSRFError for unsafe input URLs. SSRF-blocked *candidates* found
on the page are skipped silently, so one hostile link never poisons the
discovery of the legitimate ones. Raises httpx.HTTPError on network
failure — the caller decides whether to fall back to the raw URL.
"""
validate_outbound_url(url)
headers = {"User-Agent": USER_AGENT}
try:
client, backend = ssrf_client([url])
try:
content, content_type, meta = _fetch_with_redirect_validation(
client, url, headers, max_page_bytes, backend=backend
)
finally:
client.close()
except httpx.RemoteProtocolError:
# Same ALPN-hostile-edge retry as fetch_feed: some edge proxies drop
# connections advertising ALPN http/1.1 and only serve plain HTTP/1.1.
client, backend = ssrf_client([url], no_alpn=True)
try:
content, content_type, meta = _fetch_with_redirect_validation(
client, url, headers, max_page_bytes, backend=backend
)
finally:
client.close()
base_url = meta.get("final_url") or url
content_type = content_type.split(";")[0].strip().lower()
if content_type != "text/html":
# Non-HTML response: assume it IS a feed and echo the URL unchanged.
parsed = feedparser.parse(content)
title = ""
try:
title = parsed.feed.get("title", "") or ""
except AttributeError:
pass
found = not parsed.bozo or bool(parsed.version)
return {"found": found, "url": url, "title": title, "alternates": []}
parser = _DiscoveryHTMLParser()
parser.feed(content.decode("utf-8", errors="replace"))
page_title = html.unescape(parser.title).strip()
alternates: list[str] = []
seen: set[str] = set()
for rel, link_type, href in parser.links:
if (rel or "").strip().lower() != "alternate":
continue
mime = (link_type or "").split(";")[0].strip().lower()
if mime not in _FEED_LINK_TYPES and not (
mime == "application/json" and (href or "").lower().endswith(".json")
):
continue
if not href:
continue
resolved = urljoin(base_url, href.strip())
try:
validate_outbound_url(resolved)
except SSRFError:
# A page advertising one 127.0.0.1 link and one real link must
# resolve the real one, not fail the whole discovery.
continue
key = resolved.rstrip("/")
if key in seen:
continue
seen.add(key)
alternates.append(resolved)
return {
"found": bool(alternates),
"url": alternates[0] if alternates else None,
"title": page_title,
"alternates": alternates,
}
def _extract_categories(entry) -> dict[str, str]:
"""Map RSS/Atom category domain (feedparser ``scheme``) → term.
Plain tags without a domain stay only in ``tags`` / ``hashtags``. When a
scheme has a URL fragment (AlboPOP ``#item-category-type``, Dublin Core
taxonomies, …), the fragment is indexed too so templates can use
``{{ categories['item-category-type'] }}`` without repeating the full URI.
Duplicate keys keep the last term (feeds rarely repeat the same domain).
"""
categories: dict[str, str] = {}
for tag in entry.get("tags", []) or []:
if not isinstance(tag, dict):
continue
term = tag.get("term")
if not term:
continue
scheme = (tag.get("scheme") or "").strip()
if not scheme:
continue
categories[scheme] = term
if "#" in scheme:
frag = scheme.rsplit("#", 1)[-1].strip()
if frag:
categories[frag] = term
return categories
def parse_rss_feed(parsed: feedparser.FeedParserDict, url: str) -> dict:
"""Parse an RSS/Atom feed from feedparser output."""
feed_info = parsed.get("feed", {})
items = []
for entry in parsed.get("entries", []):
# Content HTML is read once and reused: {{ content }} is the
# stripped text, {{ content_link }} is the first outbound link
# inside it (recovered before strip_html drops the href).
content_html = entry.get("content", [{}])[0].get("value", "") if entry.get("content") else ""
# Feeds without <content:encoded> (plain WordPress <description>) put
# the full body in summary; use it so the reader text stays structured.
display_html = content_html or entry.get("summary", "")
item = {
"id": _get_item_id(entry),
"title": clean_text(entry.get("title", "")),
"link": entry.get("link", ""),
"summary": clean_text(entry.get("summary", "")),
"content": strip_html(content_html) if entry.get("content") else clean_text(entry.get("summary", "")),
"content_html": prepare_content_html(display_html, entry.get("link") or url) if display_html else "",
"content_text": html_to_text(display_html) if display_html else "",
"content_link": _extract_first_link(content_html, base_url=entry.get("link", "")),
"author": entry.get("author", ""),
"date": _parse_date_struct(entry),
"tags": [tag.get("term", "") for tag in entry.get("tags", []) if tag.get("term")],
"categories": _extract_categories(entry),
"image_url": _extract_rss_image(entry),
"image_alt": _extract_rss_image_alt(entry),
"image_urls": _extract_rss_images(entry),
"enclosure_url": _extract_audio_enclosure(entry),
"raw": {k: v for k, v in entry.items()},
}
items.append(item)
return {
"title": feed_info.get("title", url),
"url": url,
"type": "rss",
"items": items,
}
def parse_json_feed(data: dict) -> dict:
"""Parse a JSON Feed (https://jsonfeed.org/)."""
items = []
for entry in data.get("items", []):
# JSON Feed 1.1 uses "authors" (array), 1.0 uses "author" (object)
author_name = ""
authors = entry.get("authors") or []
if authors and isinstance(authors, list):
author_name = authors[0].get("name", "") if isinstance(authors[0], dict) else ""
elif entry.get("author") and isinstance(entry.get("author"), dict):
author_name = entry["author"].get("name", "")
content_html = entry.get("content_html") or ""
item = {
"id": _get_json_item_id(entry),
"title": entry.get("title", ""),
"link": entry.get("url", ""),
"summary": entry.get("summary", ""),
"content": strip_html(entry.get("content_html") or entry.get("content_text", "")),
"content_html": prepare_content_html(content_html, entry.get("url") or data.get("home_page_url", "")) if content_html else "",
"content_text": html_to_text(entry.get("content_html") or "") or entry.get("content_text", ""),
"content_link": _extract_first_link(content_html, base_url=entry.get("url", "")),
"author": author_name,
"date": _parse_iso_date(entry.get("date_published") or entry.get("date_modified")),
"tags": entry.get("tags", []),
# JSON Feed tags are plain strings — no category domains.
"categories": {},
"image_url": _extract_json_feed_image(entry),
"image_alt": _extract_json_feed_image_alt(entry),
"image_urls": _extract_json_feed_images(entry),
"enclosure_url": _extract_json_feed_audio(entry),
"raw": entry,
}
items.append(item)
return {
"title": data.get("title", "Unknown"),
"url": data.get("feed_url", ""),
"type": "json",
"items": items,
}
def get_new_items(items: list[dict], last_seen_id: str | None) -> list[dict]:
"""Return only items newer than last_seen_id.
If last_seen_id is None, returns empty list (first run — don't post backlog).
If last_seen_id is not found in the feed (scrolled off), returns the newest
item only to prevent backlog spam.
"""
if last_seen_id is None:
return []
if not items:
return []
new_items = []
found_seen = False
for item in items:
if item["id"] == last_seen_id:
found_seen = True
break
new_items.append(item)
if not found_seen:
# Cursor scrolled off the feed — only post the newest item to avoid spam
return items[:1] if items else []
# Items come newest-first in most feeds. Reverse so oldest-new-item posts first.
new_items.reverse()
return new_items
def get_backdated_items(
items: list[dict],
last_seen_id: str | None,
max_days: int,
now: datetime | None = None,
limit: int = 20,
) -> list[dict]:
"""Return items past the cursor whose publish date is within *max_days*.
These are entries that appear positionally OLDER than ``last_seen_id`` (so
:func:`get_new_items` skips them) but carry a publish date inside the
allowed window — e.g. a blog that backdated a post after FeedEcho already
saw a newer one.
Only items **after** the cursor in the feed list are considered. Items
before the cursor (positionally newer) are returned by
:func:`get_new_items` and must never be returned here.
Deduplication is handled downstream by ``_claim_post``'s
``ON CONFLICT(echo_id, item_id)`` upsert, so already-posted items are safe
to return here — they will be silently skipped at delivery time.
A small forward tolerance (up to 1 day ahead of *now*) absorbs publish-date
clock skew between the feed server and this host; anything further in the
future is treated as a broken feed and excluded.
Returns oldest-first (matching ``get_new_items`` ordering within the
backdated subset) so a caller posting chronologically within this set
gets the right order. Positional items from ``get_new_items`` are all
newer than the cursor; backdated items are all older — the two sets
are not interleaved, so concatenating them is not chronologically
monotonic.
"""
if last_seen_id is None or not items or max_days <= 0:
return []
if now is None:
now = datetime.now(timezone.utc)
# Locate the cursor; everything *after* it in the list is positionally
# older and a candidate for backdated delivery.
cursor_idx = None
for i, item in enumerate(items):
if item.get("id") == last_seen_id:
cursor_idx = i
break
if cursor_idx is None:
# Cursor scrolled off the feed. get_new_items returns items[:1] (the
# newest) to prevent backlog spam; the backdated scan must not
# double-handle that item, so start past it.
cursor_idx = 0 # skip items[0]
after_cursor = items[cursor_idx + 1:]
cutoff = now - timedelta(days=max_days)
future_tolerance = now + timedelta(days=1)
backdated: list[dict] = []
for item in after_cursor:
date_str = item.get("date")
if not date_str:
continue
dt = _parse_item_date(date_str)
if dt is None:
continue
if dt < cutoff or dt > future_tolerance:
continue
backdated.append(item)
if not backdated:
return []
# Take the newest *limit* items within the window, then reverse to
# oldest-first so the caller posts chronologically.
backdated.sort(key=lambda it: _parse_item_date(it["date"]), reverse=True)
backdated = backdated[:limit]
backdated.reverse()
return backdated
def _parse_item_date(date_str) -> datetime | None:
"""Parse an item date string (ISO 8601 or RFC 822) to an aware UTC datetime.
Returns None on unparseable or non-string input. Naive datetimes (no tz
suffix) are assumed to be UTC — feed timestamps are almost always
server-local and assuming UTC is safer than treating them as local wall
time.
"""
if not isinstance(date_str, str):
return None
# ISO 8601 (JSON Feed, most modern RSS/Atom).
try:
if date_str.endswith("Z"):
date_str = date_str[:-1] + "+00:00"
dt = datetime.fromisoformat(date_str)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except (ValueError, TypeError):
pass
# RFC 822 (legacy RSS pubDate).
try:
from email.utils import parsedate_to_datetime
dt = parsedate_to_datetime(date_str)
if dt is None:
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except (ValueError, TypeError, OverflowError):
pass
return None
def clean_text(text: str) -> str:
"""Strip HTML tags, decode entities, and normalize whitespace."""
if not text:
return ""
text = re.sub(r"<[^>]+>", "", text)
text = html.unescape(text)
text = re.sub(r"\s+", " ", text).strip()
return text
def strip_html(html_str: str) -> str:
"""Strip all HTML for plain-text output (Mastodon statuses are plain text)."""
if not html_str:
return ""
# Remove script/style blocks
html_str = re.sub(r"<script[^>]*>.*?</script>", "", html_str, flags=re.DOTALL | re.IGNORECASE)
html_str = re.sub(r"<style[^>]*>.*?</style>", "", html_str, flags=re.DOTALL | re.IGNORECASE)
# Strip all tags
html_str = re.sub(r"<[^>]+>", "", html_str)
# Decode entities
html_str = html.unescape(html_str)
# Normalize whitespace
html_str = re.sub(r"\s+", " ", html_str).strip()
return html_str
# Allowlist for {{ content_html }}: readable article markup only. Everything
# else — scripts, styles, event handlers, iframes, forms, custom elements — is
# stripped by nh3. Sanitized AT INGEST (parse time) so every consumer stores
# and forwards the same safe value; consumers never need to remember to
# re-sanitize. href/src are scheme-restricted to http/https by nh3's
# url_schemes.
_HTML_ALLOWED_TAGS = {
"p", "br", "a", "strong", "em", "b", "i", "u", "s",
"code", "pre", "blockquote", "ul", "ol", "li", "img",
# Email-phase additions: structured articles and newsletters carry
# headings, tables, and figure blocks. Still presentational markup only —
# no forms, frames, embeds, or style vectors.
"h1", "h2", "h3", "h4", "h5", "h6", "hr",
"table", "thead", "tbody", "tfoot", "caption", "tr", "td", "th",
"dl", "dt", "dd", "figure", "figcaption", "span",
}
_HTML_ALLOWED_ATTRS = {
"a": {"href", "title"},
"img": {"src", "alt", "title"},
"td": {"colspan", "rowspan"},
"th": {"colspan", "rowspan", "scope"},
}
def _absoluteize_urls(html_str: str, base: str) -> str:
"""Resolve relative href/src values in sanitized HTML against ``base``.
content_html flows into email parts and arbitrary webhook receivers where
a relative URL either breaks or resolves against the wrong origin. The
sanitizer guarantees the only URL-carrying attributes are a[href] and
img[src], so the substitution is anchored to those opening tags — plain
text and code blocks containing `href="..."` prose are never touched.
Runs at ingest, right after sanitization. ``base`` must itself be an
http(s) URL, or nothing is rewritten (urljoin against a hostile scheme
could smuggle non-http URLs past the sanitizer's scheme allowlist).
"""
if not html_str or not base or not base.startswith(("http://", "https://")):
return html_str
def _resolve(match: re.Match) -> str:
prefix, attr, quote, url = (
match.group(1), match.group(2), match.group(3), match.group(4),
)
if not url or url.startswith(("http://", "https://", "#", "mailto:", "cid:")):
return match.group(0)
# prefix already ends with 'href=' / 'src=' — only the value is replaced.
return f'{prefix}{quote}{urljoin(base, url)}{quote}'
return re.sub(
r'(<(?:a|img)\b[^>]*?\b(href|src)=)(["\'])([^"\']*)\3',
_resolve,
html_str,
)
def prepare_content_html(raw, base) -> str:
"""Sanitize then resolve relative URLs — the full ingest pipeline."""
base = base if isinstance(base, str) else ""
return _absoluteize_urls(sanitize_html(raw), base)
def sanitize_html(html_str) -> str:
"""Sanitize untrusted feed HTML down to safe article markup.
The allowlisted output is what {{ content_html }} delivers: formatting
and links survive, scripts/handlers/foreign schemes do not. Links gain
rel="noopener noreferrer" (nh3 default). Returns '' for empty input;
oversized input is already bounded by the feed-fetch byte cap.
"""
if not isinstance(html_str, str) or not html_str:
# feedparser occasionally yields lists/None for content fields; treat
# anything non-string as absent.
return ""
return nh3.clean(
html_str,
tags=_HTML_ALLOWED_TAGS,
attributes=_HTML_ALLOWED_ATTRS,
url_schemes={"http", "https", "mailto"},
)
def html_to_text(html_str: str) -> str:
"""Convert HTML to readable plain text, preserving paragraph/list structure.
Unlike :func:`strip_html` (which collapses everything to one run for
statuses), this keeps a document's shape for the reader: block boundaries
and <br> become newlines, <li> becomes a bullet, and script/style/iframe
blocks are dropped. Output is safe plain text (rendered pre-wrap, never as
HTML), so it needs no sanitizer.
"""
if not html_str:
return ""
html_str = re.sub(
r"<(script|style|iframe|object|embed|form)[^>]*>.*?</\1>",
"", html_str, flags=re.DOTALL | re.IGNORECASE,
)
html_str = re.sub(
r"</(p|div|section|article|blockquote|pre|h[1-6]|li|ul|ol|table|tr|figure|figcaption)>",
"\n", html_str, flags=re.IGNORECASE,
)
html_str = re.sub(r"<(br|hr)\s*/?>", "\n", html_str, flags=re.IGNORECASE)
html_str = re.sub(r"<li[^>]*>", "\n• ", html_str, flags=re.IGNORECASE)
html_str = re.sub(r"<[^>]+>", "", html_str)
text = html.unescape(html_str)
lines = [re.sub(r"[ \t\r\f\v]+", " ", ln).strip() for ln in text.split("\n")]
out = []
blank = True
for ln in lines:
if not ln:
if blank:
continue
out.append("")
blank = True
else:
out.append(ln)
blank = False
while out and out[-1] == "":
out.pop()
return "\n".join(out)
def clean_html(html_str: str) -> str:
"""Light-clean HTML: kept for backwards compat. Use strip_html for Mastodon output."""
return strip_html(html_str)
from utils import truncate_chars
def truncate(text: str, max_len: int = 500, *, preserve_links: bool = True) -> str:
"""Truncate text to max_len chars with ellipsis.
When preserve_links=True (default), preserves trailing URLs (and any trailing
hashtags/suffix) by truncating preceding text instead of severing the link.
"""
return truncate_chars(text, max_len, preserve_links=preserve_links)
def _extract_first_link(html_str: str, base_url: str = "") -> str:
"""First outbound <a href> inside the content HTML, resolved to absolute.
Link-blogs (pika.page, Micro.blog, etc.) wrap the headline in an <a>
pointing at the external article. ``strip_html()`` drops that href while
keeping the anchor text, so ``{{ content }}`` loses the link even though
the headline survives. This recovers it for ``{{ content_link }}``.
Returns the absolute URL, or "" when the content has no link. Relative
hrefs are resolved against ``base_url`` (the item's own link).
"""
if not html_str:
return ""
# Negative lookbehind on the href name so attribute suffixes like
# `data-href` / `x-href` are not mistaken for a real `href`.
match = re.search(r'<a\b[^>]*?(?<![\w-])href\s*=\s*["\']([^"\']+)["\']', html_str, re.IGNORECASE)
if not match:
return ""
href = html.unescape(match.group(1)).strip()
if not href:
return ""
if "://" in href:
return href
if href.startswith("//"):
return "https:" + href
if base_url:
return urljoin(base_url, href)
return href
def _is_image_media(media: dict) -> bool:
"""True when a media_content/media_thumbnail entry is an image."""
medium = str(media.get("medium", "")).lower()
if medium in ("video", "audio"):
return False
m_type = str(media.get("type", "")).lower()
if m_type and not m_type.startswith("image/"):
return False
return True
def _extract_rss_images(entry: dict) -> list[dict]:
"""Extract up to 4 image {url, alt} pairs from an RSS/Atom entry.
Priority order per slot: media_content, media_thumbnail (fallback only
when media_content has no valid images), image enclosures, then <img>
tags in content/summary HTML. Capped at 4 (Mastodon's default image
limit).
"""
images: list[dict] = []
def _add(url: str, alt: str = "") -> None:
if not isinstance(url, str) or not url.strip():
return
url = url.strip()
if url.startswith("data:"):
return
if len(images) >= 4:
return
if any(img["url"] == url for img in images):
return
images.append({"url": url, "alt": (alt or "").strip()})
# Media RSS media_content (may carry media:text captions)
for media in entry.get("media_content") or []:
if not isinstance(media, dict) or not _is_image_media(media):
continue
alt = ""
text = media.get("media_text", "")
if isinstance(text, list) and text:
alt = text[0].get("text", "") if isinstance(text[0], dict) else ""
elif isinstance(text, str):
alt = text
_add(media.get("url", ""), alt)
# media_thumbnail is the thumbnail of media_content: only consult it
# when media_content produced nothing, so a post never ships both the
# full image and its thumbnail.
if not images:
for media in entry.get("media_thumbnail") or []:
if not isinstance(media, dict):
continue
_add(media.get("url", ""))
# RSS enclosures with image MIME types
for enc in entry.get("enclosures", []):
if isinstance(enc, dict) and enc.get("type", "").startswith("image/"):
_add(enc.get("href", ""), enc.get("alt", ""))
# Every <img> in content/summary HTML (src + alt from the same tag)
html_content = ""
if entry.get("content"):
html_content = entry.get("content", [{}])[0].get("value", "")
if not html_content:
html_content = entry.get("summary", "")
if html_content:
for tag in re.findall(r"<img\b[^>]*>", html_content, re.IGNORECASE):
src = re.search(r'\bsrc=["\']([^"\']+)["\']', tag, re.IGNORECASE)
if not src:
continue
alt = re.search(r'\balt=["\']([^"\']*)["\']', tag, re.IGNORECASE)
_add(html.unescape(src.group(1)), html.unescape(alt.group(1)) if alt else "")
return images
def _extract_rss_image(entry: dict) -> str:
"""Extract first image URL from an RSS/Atom entry.
Priority: media_content > media_thumbnail > enclosures > first <img> in content/summary.
Returns "" if no image found.
"""
# media_content / media_thumbnail (Media RSS)
for key in ("media_content", "media_thumbnail"):
media_list = entry.get(key, [])
if media_list and isinstance(media_list, list):
for media in media_list:
if not isinstance(media, dict) or not _is_image_media(media):
continue
url = media.get("url", "")
if url:
return url
# RSS enclosures
for enc in entry.get("enclosures", []):
if isinstance(enc, dict):
ctype = enc.get("type", "")
if ctype.startswith("image/") and enc.get("href"):
return enc["href"]
# First <img> in content or summary HTML
html_content = ""
if entry.get("content"):
html_content = entry.get("content", [{}])[0].get("value", "")
if not html_content:
html_content = entry.get("summary", "")
if html_content:
match = re.search(r'<img[^>]+src=["\']([^"\']+)["\']', html_content, re.IGNORECASE)
if match:
return match.group(1)
return ""
def _extract_rss_image_alt(entry: dict) -> str:
"""Alt text for the image _extract_rss_image picked, or "".
Mirrors its priority order so the caption always belongs to the URL we
chose: Media RSS media:text on the same media object, enclosure alt, or
the alt attribute of the content <img> whose src was chosen.
"""
for key in ("media_content", "media_thumbnail"):
media_list = entry.get(key, [])
if media_list and isinstance(media_list, list):
for media in media_list:
if not isinstance(media, dict) or not _is_image_media(media):
continue
if not media.get("url", ""):
continue
text = media.get("media_text", "")
if isinstance(text, list) and text:
text = text[0].get("text", "") if isinstance(text[0], dict) else ""
if isinstance(text, str) and text.strip():
return text.strip()
return ""
for enc in entry.get("enclosures", []):
if isinstance(enc, dict) and enc.get("type", "").startswith("image/") and enc.get("href"):
return enc.get("alt", "") or ""
html_content = ""
if entry.get("content"):
html_content = entry.get("content", [{}])[0].get("value", "")
if not html_content:
html_content = entry.get("summary", "")
if not html_content:
return ""
# The URL extractor takes the FIRST <img src>; only its own alt applies.
match = re.search(r"<img\b[^>]*>", html_content, re.IGNORECASE)
if not match:
return ""
tag = match.group(0)
src = re.search(r'\bsrc=["\']([^"\']+)["\']', tag, re.IGNORECASE)
if not src:
return ""
alt = re.search(r'\balt=["\']([^"\']*)["\']', tag, re.IGNORECASE)
return alt.group(1).strip() if alt else ""
def _extract_audio_enclosure(entry: dict) -> str:
"""First audio enclosure URL (podcasts), or ''."""
for enc in entry.get("enclosures", []):
if isinstance(enc, dict) and enc.get("type", "").startswith("audio/") and enc.get("href"):
return enc["href"]
return ""
def _extract_json_feed_audio(entry: dict) -> str:
"""First audio attachment URL from a JSON Feed item, or ''."""
for att in entry.get("attachments") or []:
if isinstance(att, dict) and str(att.get("mime_type", "")).startswith("audio/") and att.get("url"):
return att["url"]
return ""
def _extract_json_feed_image(entry: dict) -> str:
"""Extract first image URL from a JSON Feed entry.
Priority: image > banner_image > first <img> in content_html.
JSON Feed 1.1 allows "image" as an object {url, caption} (or a string).
Returns "" if no image found.
"""
image = entry.get("image")
if isinstance(image, dict) and image.get("url"):
return image["url"]
if isinstance(image, str) and image:
return image
banner = entry.get("banner_image")
if isinstance(banner, dict) and banner.get("url"):
return banner["url"]
if isinstance(banner, str) and banner:
return banner
html_content = entry.get("content_html", "")
if html_content:
match = re.search(r'<img[^>]+src=["\']([^"\']+)["\']', html_content, re.IGNORECASE)
if match:
return match.group(1)
return ""
def _extract_json_feed_images(entry: dict) -> list[dict]:
"""Extract up to 4 image {url, alt} pairs from a JSON Feed entry.
JSON Feed 1.1 priority: the item's `image` field is primary,
`attachments` are supplementary, `banner_image` is a site-level header
(fallback only when the item has no image of its own). <img> tags in
content_html come last. Capped at 4.
"""
images: list[dict] = []
def _add(url: str, alt: str = "") -> None:
if not isinstance(url, str) or not url.strip():
return
url = url.strip()
if url.startswith("data:"):
return
if len(images) >= 4:
return
if any(img["url"] == url for img in images):
return
images.append({"url": url, "alt": (alt or "").strip()})
image = entry.get("image")
if isinstance(image, dict) and image.get("url"):
_add(image["url"], image.get("caption", ""))
elif isinstance(image, str):
_add(image)
for att in entry.get("attachments") or []:
if isinstance(att, dict) and str(att.get("mime_type", "")).startswith("image/"):
_add(att.get("url", ""), att.get("title", ""))
# banner_image is the site header, not the article image: only use it
# when the item itself had no image at all.
if not images:
banner = entry.get("banner_image")
if isinstance(banner, dict) and banner.get("url"):
_add(banner["url"], banner.get("caption", ""))
elif isinstance(banner, str):
_add(banner)
content_html = entry.get("content_html", "")
if content_html:
for tag in re.findall(r"<img\b[^>]*>", content_html, re.IGNORECASE):
src = re.search(r'\bsrc=["\']([^"\']+)["\']', tag, re.IGNORECASE)
if not src:
continue
alt = re.search(r'\balt=["\']([^"\']*)["\']', tag, re.IGNORECASE)
_add(html.unescape(src.group(1)), html.unescape(alt.group(1)) if alt else "")
return images
def _extract_json_feed_image_alt(entry: dict) -> str:
"""Alt text for the image _extract_json_feed_image picked, or "".
JSON Feed 1.1 allows "image" as an object {url, caption} (or a string);
caption is the official alt-text slot. Falls back to the content <img>
alt attribute when the URL came from content_html.
"""
image = entry.get("image")
if isinstance(image, dict) and image.get("url"):
return (image.get("caption") or "").strip()
if isinstance(image, str) and image:
return ""
banner = entry.get("banner_image")
if isinstance(banner, dict) and banner.get("url"):
return (banner.get("caption") or "").strip()
if isinstance(banner, str) and banner:
return ""
html_content = entry.get("content_html", "")
if not html_content:
return ""
# Mirror the URL extractor: the FIRST <img> with a src is the chosen one,
# so only its own alt applies.
match = re.search(r"<img\b[^>]*>", html_content, re.IGNORECASE)
while match:
tag = match.group(0)
if re.search(r'\bsrc=["\']([^"\']+)["\']', tag, re.IGNORECASE):
alt = re.search(r'\balt=["\']([^"\']*)["\']', tag, re.IGNORECASE)
return alt.group(1).strip() if alt else ""
match = re.search(r"<img\b[^>]*>", html_content[match.end():], re.IGNORECASE)
return ""
def fetch_image(url: str) -> tuple[bytes, str] | None:
"""Fetch an image for Mastodon media upload.
Validates URL for SSRF, caps size, and returns (content_bytes, content_type).
Returns None if the fetch fails, the URL is unsafe, or the content is not an image.
"""
try:
validate_outbound_url(url)
except SSRFError:
return None
headers = {"User-Agent": USER_AGENT}
try:
try:
client, backend = ssrf_client([url])
try:
content, raw_type, _meta = _fetch_with_redirect_validation(
client, url, headers, MAX_IMAGE_SIZE, backend=backend
)
finally:
client.close()
except httpx.HTTPStatusError as exc:
if exc.response.status_code in (403, 429) and settings.FALLBACK_PROXY_URL:
content, raw_type, _meta = _fetch_via_fallback_proxy(url, headers, MAX_IMAGE_SIZE)
else:
raise
content_type = raw_type.split(";")[0].strip()
if content_type not in ALLOWED_IMAGE_TYPES:
return None
return content, content_type
except Exception:
# A single dead link or blocked host is routine (dead feeds,
# geo-blocked images) and not worth a log line above DEBUG — but a
# *systemic* failure (misconfigured FALLBACK_PROXY_URL, an outbound
# network policy change) previously had zero trace anywhere, since
# this caught network errors, redirect-loop/oversize ValueErrors,
# and fallback-proxy failures alike with no logging at all.
# Error-handling audit finding 5.3a (docs/reviews/2026-09-08-error-handling-audit.md).
logger.debug("Image fetch failed for %s", url, exc_info=True)
return None
# ── Link-card page metadata (Bluesky external embeds) ────────────────────────
MAX_PAGE_METADATA_BYTES = 2_000_000 # cap the HTML we pull for og: tags
_TITLE_TAG_RE = re.compile(r"<title\b[^>]*>(.*?)", re.I | re.S)
_META_TAG_RE = re.compile(r"]*>", re.I)
_META_KEY_RE = re.compile(r"""(?:property|name)\s*=\s*["']([^"']+)["']""", re.I)
# Paired quotes: [^"']* would stop a double-quoted value at an apostrophe
# ("Today's Headline" -> "Today"), so each alternative matches its own
# closing delimiter.
_META_CONTENT_RE = re.compile(r"""content\s*=\s*(?:"([^"]*)"|'([^']*)')""", re.I)
def _extract_page_metadata(page_html: str, base_url: str = "") -> dict:
"""Pull og:*/twitter:* card metadata out of an HTML string.
og: wins over twitter:; and meta description are the fallbacks
when neither card format is present. Values are HTML-unescaped (markup
in the <title> fallback is stripped, not shown raw) and a relative
image URL is resolved against base_url. Only the head region matters,
but the search runs over whatever the caller passes
(fetch_page_metadata caps the body size upstream).
"""
og: dict[str, str] = {}
twitter: dict[str, str] = {}
plain_description = ""
for tag in _META_TAG_RE.findall(page_html):
key_match = _META_KEY_RE.search(tag)
content_match = _META_CONTENT_RE.search(tag)
if not key_match or not content_match:
continue
key = key_match.group(1).strip().lower()
value = html.unescape(
content_match.group(1)
if content_match.group(1) is not None
else content_match.group(2)
).strip()
if not value:
continue
if key.startswith("og:") and key.count(":") == 1:
og.setdefault(key[3:], value)
elif key.startswith("twitter:"):
rest = key[len("twitter:"):]
if rest == "image:src": # legacy Twitter card field
rest = "image"
if ":" not in rest:
twitter.setdefault(rest, value)
elif key == "description" and not plain_description:
plain_description = value
title_match = _TITLE_TAG_RE.search(page_html)
title = (
og.get("title")
or twitter.get("title")
or (
html.unescape(re.sub(r"<[^>]+>", "", title_match.group(1))).strip()
if title_match
else ""
)
)
description = og.get("description") or twitter.get("description") or plain_description
image = og.get("image") or twitter.get("image") or ""
if image and base_url:
image = urljoin(base_url, image)
return {"title": title, "description": description, "image": image}
def fetch_page_metadata(url: str) -> dict | None:
"""Fetch a page and extract Open Graph / Twitter card metadata.
Used for Bluesky link cards (app.bsky.embed.external): returns
{"title", "description", "image"} where image is the og:image URL
string, or None when the fetch fails, the URL is unsafe, or the
response is not HTML. Never raises — a dead link must degrade to a
cardless post, not fail the echo.
"""
try:
validate_outbound_url(url)
except SSRFError as exc:
# A refused or unresolvable URL is silent otherwise, and the reason
# (private address, DNS failure) is the whole diagnosis. The Bluesky
# caller logs the INFO-level "no link card" line and points here.
logger.debug("Page metadata skipped for %s: %s", url, exc)
return None
headers = {"User-Agent": USER_AGENT}
try:
client, backend = ssrf_client([url])
try:
content, raw_type, _meta = _fetch_with_redirect_validation(
client, url, headers, MAX_PAGE_METADATA_BYTES, backend=backend
)
finally:
client.close()
except httpx.HTTPStatusError as exc:
status = exc.response.status_code
if status in (403, 429) and settings.FALLBACK_PROXY_URL:
try:
content, raw_type, _meta = _fetch_via_fallback_proxy(
url, headers, MAX_PAGE_METADATA_BYTES
)
except Exception:
logger.debug(
"Page metadata fallback fetch failed for %s (origin returned HTTP %s)",
url,
status,
exc_info=True,
)
return None
else:
# The status is what tells a WAF block (403/429) from a dead link.
logger.debug("Page metadata fetch failed for %s: HTTP %s", url, status)
return None
except Exception:
logger.debug("Page metadata fetch failed for %s", url, exc_info=True)
return None
content_type = raw_type.split(";")[0].strip().lower()
if content_type and "html" not in content_type and "xml" not in content_type:
logger.debug(
"Page metadata skipped for %s: content type %s is not HTML",
url,
content_type,
)
return None
text = content.decode("utf-8", errors="replace")
return _extract_page_metadata(text, base_url=url)
def _get_item_id(entry: dict) -> str:
"""Get a stable item ID, synthesizing one if the feed lacks guid/link."""
item_id = entry.get("id") or entry.get("guid") or entry.get("link", "")
if item_id:
return item_id
# Synthesize from title + date + content hash
seed = (entry.get("title", "") + entry.get("published", "") + entry.get("summary", "")).encode()
return hashlib.sha256(seed).hexdigest()[:16]
def _get_json_item_id(entry: dict) -> str:
"""Get a stable item ID for JSON Feed entries."""
item_id = entry.get("id") or entry.get("url", "")
if item_id:
return item_id
seed = (entry.get("title", "") + entry.get("date_published", "") + entry.get("content_text", "")).encode()
return hashlib.sha256(seed).hexdigest()[:16]
def _parse_date_struct(entry: dict) -> str | None:
"""Parse date from feedparser's struct_time fields (already parsed)."""
for field in ("published_parsed", "updated_parsed"):
parsed = entry.get(field)
if parsed:
try:
# timegm, not mktime: feedparser's *_parsed struct_times are
# already UTC, and mktime interprets them as local wall time,
# so every item date was shifted by the host's UTC offset.
dt = datetime.fromtimestamp(calendar.timegm(parsed), tz=timezone.utc)
return dt.isoformat()
except (ValueError, OverflowError):
pass
# Fallback to raw string
return entry.get("published") or entry.get("updated")
def _parse_iso_date(date_str: str | None) -> str | None:
"""Parse an ISO 8601 date string."""
if not date_str:
return None
try:
dt = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
return dt.isoformat()
except (ValueError, TypeError):
return date_str