app/tasks/link_health_task.py

"""Author RAG Chatbot SaaS — Link Health Check Task.

Runs daily. Validates all stored purchase/preview URLs.
Marks broken URLs in the database and optionally alerts authors.
"""

import structlog
import httpx
from app.tasks.celery_app import celery_app

logger = structlog.get_logger(__name__)


@celery_app.task(name="app.tasks.link_health_task.check_all_links")
def check_all_links():
    """Check all stored book links for HTTP 200 response."""
    import asyncio
    asyncio.run(_run())


async def _run():
    from datetime import datetime, timezone
    from sqlalchemy import select, update
    from app.dependencies import _get_session_factory
    from app.models.link import Link

    now = datetime.now(timezone.utc)
    async with _get_session_factory()() as db:
        result = await db.execute(
            select(Link).where(Link.purchase_url != None)
        )
        links = result.scalars().all()

        async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client:
            for link in links:
                purchase_ok = None
                preview_ok = None

                if link.purchase_url:
                    purchase_ok = await _check_url(client, link.purchase_url)
                if link.preview_url:
                    preview_ok = await _check_url(client, link.preview_url)

                await db.execute(
                    update(Link).where(Link.id == link.id).values(
                        purchase_url_ok=purchase_ok,
                        preview_url_ok=preview_ok,
                        last_health_check=now,
                    )
                )

        await db.commit()
    logger.info("Link health check complete", checked=len(links))


async def _check_url(client: httpx.AsyncClient, url: str) -> bool:
    """Check if a URL returns an OK response.

    Includes SSRF protection — refuses to fetch internal/private IPs (R-130).

    Args:
        client: httpx async client.
        url: URL to check.

    Returns:
        True if URL is reachable (2xx/3xx), False otherwise.
    """
    # R-130: SSRF protection — block internal/private IPs before fetching.
    # BUG-7 fix: _is_internal_url() calls socket.gethostbyname() which is a
    # blocking DNS call — offload to thread pool to avoid freezing the event loop.
    import asyncio
    is_internal = await asyncio.to_thread(_is_internal_url, url)
    if is_internal:
        logger.warning("Link health: blocked internal URL", url=url[:100])
        return False
    try:
        response = await client.head(url)
        return response.status_code < 400
    except Exception:
        return False


def _is_internal_url(url: str) -> bool:
    """Return True if the URL resolves to an internal/private/loopback IP."""
    import ipaddress
    import socket
    from urllib.parse import urlparse

    try:
        hostname = urlparse(url).hostname
        if not hostname:
            return True
        ip = ipaddress.ip_address(socket.gethostbyname(hostname))
        return ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved
    except Exception:
        return True  # Fail safe — block if we can't resolve