saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
55 lines
1.7 KiB
Python
55 lines
1.7 KiB
Python
"""robots.txt compliance check - see SECURITY.md rule 1."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from urllib.parse import urljoin, urlparse
|
|
from urllib.robotparser import RobotFileParser
|
|
|
|
from app.core.config import Settings, get_settings
|
|
from app.core.http import SsrfBlockedError, safe_fetch
|
|
from app.core.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
_CACHE_TTL_SECONDS = 3600
|
|
_cache: dict[str, tuple[float, RobotFileParser]] = {}
|
|
|
|
|
|
async def _get_parser(base_url: str, settings: Settings) -> RobotFileParser:
|
|
parsed = urlparse(base_url)
|
|
origin = f"{parsed.scheme}://{parsed.netloc}"
|
|
cached = _cache.get(origin)
|
|
now = time.monotonic()
|
|
if cached and now - cached[0] < _CACHE_TTL_SECONDS:
|
|
return cached[1]
|
|
|
|
parser = RobotFileParser()
|
|
robots_url = urljoin(origin, "/robots.txt")
|
|
try:
|
|
result = await safe_fetch(robots_url, settings=settings)
|
|
if result.status_code == 200:
|
|
parser.parse(result.text.splitlines())
|
|
else:
|
|
# No robots.txt or inaccessible -> "allow all" per convention.
|
|
parser.parse([])
|
|
except SsrfBlockedError:
|
|
parser.parse([])
|
|
except Exception as exc: # pragma: no cover - defensive
|
|
logger.warning("robots_txt_fetch_failed", url=robots_url, error=str(exc))
|
|
parser.parse([])
|
|
|
|
_cache[origin] = (now, parser)
|
|
return parser
|
|
|
|
|
|
async def is_allowed(url: str, *, settings: Settings | None = None) -> bool:
|
|
settings = settings or get_settings()
|
|
parser = await _get_parser(url, settings)
|
|
return parser.can_fetch(settings.scraper_user_agent, url)
|
|
|
|
|
|
def clear_cache() -> None:
|
|
"""Test helper - the module-level cache would otherwise leak between tests."""
|
|
_cache.clear()
|