import logging import re import uuid from datetime import datetime, timedelta, timezone from html.parser import HTMLParser import httpx from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.database import async_session_factory from app.models import LinkPreview from app.services.ssrf import UnsafeUrlError, validate_target_url from app.storage import ALLOWED_IMAGE_CONTENT_TYPES from app.ws.broadcaster import Broadcaster logger = logging.getLogger(__name__) # Deliberately not a full URL regex (e.g. no IPv6-literal-host support) -- # just enough to find "a URL-shaped thing" in a chat message the same way # the frontend's Markdown renderer autolinks bare URLs. Trailing punctuation # a sentence would naturally have after a URL (a period, closing paren from # "(see https://example.com)", etc.) is trimmed off separately below. _URL_RE = re.compile(r"https?://[^\s<>\"]+") _TRAILING_PUNCTUATION = ".,;:!?)'\">" _FETCH_TIMEOUT_SECONDS = 5.0 _MAX_BYTES = 512 * 1024 _MAX_REDIRECTS = 3 # #70: was 7 days -- confirmed live as too long for how this app actually # gets used: re-posting a URL whose title/content had genuinely changed # kept showing the stale first-fetch preview for up to a week. Short # enough that it's effectively "always fresh" for any realistic human # posting cadence, while still doing the one thing a cache here is # actually for -- collapsing a burst of near-simultaneous fetches of the # same URL (several people pasting the same link within moments of each # other, or the same person's message history being loaded repeatedly) # into one, and not hammering a URL that just failed on every message # that references it. _USER_AGENT = "ds-chat-link-preview/1.0" _CACHE_TTL = timedelta(minutes=5) def extract_first_url(content: str | None) -> str | None: if not content: return None match = _URL_RE.search(content) if not match: return None return match.group(0).rstrip(_TRAILING_PUNCTUATION) or None class _OpenGraphParser(HTMLParser): """Pulls og:title/og:description/og:image/og:site_name meta tags, falling back to -- stdlib html.parser is enough for meta-tag scraping, no reason to add a full HTML parsing dependency (e.g. BeautifulSoup) just for this.""" def __init__(self) -> None: super().__init__(convert_charrefs=True) self.og: dict[str, str] = {} self.title: str | None = None self._in_title = False def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if tag == "meta": attr_dict = dict(attrs) prop = attr_dict.get("property") or attr_dict.get("name") content = attr_dict.get("content") if prop and content and prop.startswith("og:"): self.og.setdefault(prop, content) elif tag == "title": self._in_title = True def handle_endtag(self, tag: str) -> None: if tag == "title": self._in_title = False def handle_data(self, data: str) -> None: if self._in_title and self.title is None: self.title = data.strip() async def _fetch_preview_data(url: str) -> dict | None: """SSRF-safe fetch: validates (scheme + resolved-IP allowlist check, see ssrf.py) before *every* hop, following redirects manually rather than via httpx's own follow_redirects -- that would connect to each intermediate hop before any of them got validated, defeating the point. Still subject to the DNS-rebinding gap documented in ssrf.py (a window between validating a hostname and httpx independently resolving it to connect); accepted for the same reason it's accepted there. """ current_url = url async with httpx.AsyncClient(timeout=_FETCH_TIMEOUT_SECONDS, follow_redirects=False) as client: for _ in range(_MAX_REDIRECTS + 1): try: validate_target_url(current_url) except UnsafeUrlError: return None try: async with client.stream( "GET", current_url, headers={"User-Agent": _USER_AGENT} ) as response: if response.is_redirect: location = response.headers.get("location") if not location: return None current_url = str(httpx.URL(current_url).join(location)) continue if response.status_code >= 400: return None content_type = response.headers.get("content-type", "").split(";")[0].strip() # A direct link to an image file -- render the image # itself rather than trying to scrape og: tags from # nonexistent HTML. No need to read the body at all; # the URL (now fully resolved through any redirects, # each already SSRF-validated above) *is* the preview. if content_type in ALLOWED_IMAGE_CONTENT_TYPES: return { "title": None, "description": None, "image_url": current_url, "site_name": None, "is_image": True, } if content_type != "text/html": return None body = b"" async for chunk in response.aiter_bytes(): body += chunk if len(body) >= _MAX_BYTES: break break except httpx.HTTPError: return None else: return None parser = _OpenGraphParser() try: parser.feed(body.decode("utf-8", errors="replace")) except Exception: return None title = parser.og.get("og:title") or parser.title if not title: return None return { "title": title.strip()[:500], "description": (parser.og.get("og:description") or "").strip()[:1000] or None, "image_url": parser.og.get("og:image") or None, "site_name": (parser.og.get("og:site_name") or "").strip()[:200] or None, "is_image": False, } async def _get_or_fetch(db: AsyncSession, url: str) -> LinkPreview | None: """Returns None only when there's genuinely nothing to show (a fresh fetch failed, or a cached row says an earlier one did) -- callers don't need to distinguish "still fetching" from "never going to have one".""" existing = (await db.execute(select(LinkPreview).where(LinkPreview.url == url))).scalar_one_or_none() if existing is not None and datetime.now(timezone.utc) - existing.fetched_at < _CACHE_TTL: return None if existing.fetch_failed else existing data = await _fetch_preview_data(url) if existing is not None: existing.fetch_failed = data is None if data: existing.title = data["title"] existing.description = data["description"] existing.image_url = data["image_url"] existing.site_name = data["site_name"] existing.is_image = data["is_image"] existing.fetched_at = datetime.now(timezone.utc) await db.commit() return None if data is None else existing row = LinkPreview( url=url, fetch_failed=data is None, **( data or { "title": None, "description": None, "image_url": None, "site_name": None, "is_image": False, } ), ) db.add(row) await db.commit() return None if data is None else row async def fetch_and_broadcast_link_preview( broadcaster: Broadcaster, room_id: uuid.UUID, message_id: uuid.UUID, url: str ) -> None: """Entry point for a fire-and-forget asyncio.create_task from broadcast_new_message -- runs on its own DB session (see push_service.py's send_push_to_user docstring for why a background task must never share the caller's session) so a slow/hanging fetch can never delay message delivery to anyone actually online. """ try: async with async_session_factory() as db: preview = await _get_or_fetch(db, url) except Exception: logger.warning("Link preview fetch failed for %s", url, exc_info=True) return if preview is None: return await broadcaster.publish( room_id, { "type": "link_preview", "room_id": str(room_id), "id": str(message_id), "url": preview.url, "title": preview.title, "description": preview.description, "image_url": preview.image_url, "site_name": preview.site_name, "is_image": preview.is_image, }, ) async def get_link_previews_for_urls(db: AsyncSession, urls: list[str]) -> dict[str, LinkPreview]: """Batch lookup for message history -- mirrors message_service. get_reactions_for_messages's shape (one query, zipped back onto results by the caller) rather than an ORM relationship, since the join key is a plain string column, not a FK.""" if not urls: return {} result = await db.execute(select(LinkPreview).where(LinkPreview.url.in_(urls), LinkPreview.fetch_failed.is_(False))) return {row.url: row for row in result.scalars().all()}