searxng-use-cli/scripts/common.py

"""Shared utilities for searxng-cli scripts.

This module centralizes code that was previously duplicated across
``search.py`` and ``fetch.py``:

* ``build_auth_headers``  — construct an Authorization header from CLI flags
* ``resolve_auth_basic``  — resolve basic-auth credentials from file/env/CLI
                            (avoids leaving passwords in shell history)
* ``detect_charset``      — guess a response's text encoding
* ``is_retryable_error``  — unified transient-error policy (urllib + requests)
* ``FALLBACK_UAS``        — browser-like User-Agents used when blocked
* retry constants         — ``RETRYABLE_STATUS``, ``RETRY_BACKOFF_BASE``, etc.
* ``setup_logging``       — shared logging configuration (--verbose/--quiet)

Centralizing the retry policy guarantees that both scripts treat 429/5xx
as retryable and connection errors as transient, eliminating the previous
inconsistency where ``search.py`` ignored 5xx.
"""

import logging
import sys
import urllib.error

# Root logger for the searxng-cli package. All modules create child loggers
# via ``logging.getLogger("searxng.<module>")`` so a single setup_logging()
# call controls them all.
_LOG = logging.getLogger("searxng")


def setup_logging(verbose: bool = False, quiet: bool = False) -> None:
    """Configure the ``searxng`` logger hierarchy.

    * Default (no flags): ``INFO`` — progress messages, warnings, retry notices.
      Matches the previous ``print(..., file=sys.stderr)`` behavior so existing
      scripts and agents see no change.
    * ``--verbose`` / ``-v``: ``DEBUG`` — also shows HTTP request URLs, response
      status codes, cache keys, and other diagnostic detail.
    * ``--quiet`` / ``-q``: ``WARNING`` — suppresses progress and retry noise;
      only warnings and errors reach stderr.

    All log output goes to stderr; stdout is reserved for data (JSON/CSV/etc.).
    """
    if verbose:
        level = logging.DEBUG
    elif quiet:
        level = logging.WARNING
    else:
        level = logging.INFO

    _LOG.setLevel(level)
    # Avoid duplicate handlers if setup_logging() is called twice (e.g. tests).
    if not _LOG.handlers:
        handler = logging.StreamHandler(sys.stderr)
        handler.setFormatter(logging.Formatter("%(message)s"))
        _LOG.addHandler(handler)
    # Don't let root logger add its own handler — we own the searxng namespace.
    _LOG.propagate = False

# Retry settings (shared by both scripts)
MAX_RETRIES = 3
RETRY_BACKOFF_BASE = 1.5  # seconds; exponential backoff + jitter

# HTTP status codes that are worth retrying (rate limit + gateway errors)
RETRYABLE_STATUS = frozenset({429, 502, 503, 504})

# Browser-like UA strings for fallback when the searxng-cli UA is blocked
FALLBACK_UAS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
]


def build_auth_headers(bearer_token: str = None, basic_auth: str = None) -> dict:
    """Build an Authorization header dict from CLI auth flags.

    ``bearer_token``: raw Bearer token string.
    ``basic_auth``:   ``"username:password"`` string (base64-encoded).

    If both are provided, Bearer takes precedence (more common for APIs).
    Returns a dict to merge into request headers, or an empty dict.
    """
    import base64

    headers = {}
    if bearer_token:
        headers["Authorization"] = f"Bearer {bearer_token}"
    elif basic_auth:
        encoded = base64.b64encode(basic_auth.encode("utf-8")).decode("ascii")
        headers["Authorization"] = f"Basic {encoded}"
    return headers


def _warn_file_perms(path: str) -> None:
    """Warn if a credentials file is readable by group/other (POSIX only).

    On Windows the Unix permission bits in ``st_mode`` do not reflect the
    actual ACL, so the check is skipped to avoid false alarms.
    """
    import os
    if os.name != "posix":
        return
    log = logging.getLogger("searxng.common")
    try:
        mode = os.stat(path).st_mode & 0o777
        if mode & 0o077:
            log.warning(
                f"Warning: credentials file '{path}' has permissions {oct(mode)} "
                f"(accessible by group/other); recommend 'chmod 600' for security."
            )
    except OSError:
        pass


def resolve_auth_basic(cli_value: str = None, file_path: str = None,
                       env_var: str = "SEARXNG_BASIC_AUTH") -> str:
    """Resolve basic-auth credentials without leaking them via shell history.

    Priority (highest wins):
      1. ``cli_value``  — explicit ``--auth-basic "user:pass"`` (convenient
                          but leaks into shell history; discouraged)
      2. ``file_path``  — ``--auth-basic-file FILE``; first non-empty line
                          is read as ``user:pass``. Recommended for shells.
      3. ``env_var``    — ``SEARXNG_BASIC_AUTH`` environment variable.

    Returns ``"user:pass"`` or ``None`` if no source provides credentials.
    Raises ``RuntimeError`` if a file is specified but cannot be read.
    """
    if cli_value:
        return cli_value

    if file_path:
        try:
            from pathlib import Path
            text = Path(file_path).read_text(encoding="utf-8")
            _warn_file_perms(file_path)
            for line in text.splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    return line
            raise RuntimeError(f"auth file '{file_path}' contains no credentials")
        except OSError as e:
            raise RuntimeError(f"cannot read auth file '{file_path}': {e}") from e

    import os
    return os.environ.get(env_var)


def resolve_auth_bearer(cli_value: str = None, file_path: str = None,
                        env_var: str = "SEARXNG_BEARER_TOKEN") -> str:
    """Resolve a Bearer token from CLI flag, file, or environment variable.

    Mirrors :func:`resolve_auth_basic` for token-style auth. Useful for
    long-lived API tokens that should not appear in shell history.
    """
    if cli_value:
        return cli_value

    if file_path:
        try:
            from pathlib import Path
            text = Path(file_path).read_text(encoding="utf-8")
            _warn_file_perms(file_path)
            for line in text.splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    return line
            raise RuntimeError(f"token file '{file_path}' contains no token")
        except OSError as e:
            raise RuntimeError(f"cannot read token file '{file_path}': {e}") from e

    import os
    return os.environ.get(env_var)


def apply_proxy(proxy_url: str) -> None:
    """Configure proxy via environment variables.

    Sets ``HTTP_PROXY`` and ``HTTPS_PROXY`` so both urllib (which reads
    them via :func:`urllib.request.getproxies`) and ``requests`` (which
    honors them when ``trust_env=True``, the default) pick up the proxy
    without any changes to call sites.

    ``NO_PROXY`` is set to ``localhost,127.0.0.1,::1`` (if not already set)
    so local traffic stays direct — matters for self-hosted SearXNG on
    localhost behind a corporate proxy.

    Pass an empty string to clear the proxy env vars (rarely needed; the
    default unset state already means "no proxy").
    """
    import os
    if not proxy_url:
        return
    os.environ["HTTP_PROXY"] = proxy_url
    os.environ["HTTPS_PROXY"] = proxy_url
    # Keep local traffic direct unless the user has explicitly set NO_PROXY
    os.environ.setdefault("NO_PROXY", "localhost,127.0.0.1,::1")


def detect_charset(raw: bytes, content_type: str) -> str:
    """Detect charset from the Content-Type header, then an HTML meta tag.

    Falls back to UTF-8 (with replacement) if nothing reliable is found.
    """
    import re

    # 1. HTTP header
    if "charset=" in content_type:
        charset = content_type.split("charset=")[-1].split(";")[0].strip()
        try:
            raw.decode(charset)
            return charset
        except (UnicodeDecodeError, LookupError):
            pass

    # 2. HTML <meta charset> or <meta http-equiv>
    try:
        head = raw[:4096].decode("ascii", errors="replace")
        m = re.search(r'<meta[^>]+charset=["\']?([a-zA-Z0-9_-]+)', head, re.IGNORECASE)
        if m:
            charset = m.group(1).strip()
            try:
                raw.decode(charset)
                return charset
            except (UnicodeDecodeError, LookupError):
                pass
    except Exception:
        pass

    # 3. Fallback: UTF-8 with replacement
    return "utf-8"


def is_retryable_error(exc: BaseException) -> bool:
    """Return True if ``exc`` is a transient error worth retrying.

    Handles both the stdlib ``urllib`` errors and ``requests`` errors via
    duck-typing (so this module does not need to import ``requests``):

    * ``urllib.error.HTTPError`` → retry iff status in ``RETRYABLE_STATUS``
    * ``urllib.error.URLError`` / ``OSError`` / ``TimeoutError`` → retry
      (connection refused, DNS failure, timeout — all transient)
    * ``requests.exceptions.HTTPError`` → retry iff ``response.status_code``
      is in ``RETRYABLE_STATUS``
    * ``requests`` connection/timeout errors (no ``.response``) → retry
    """
    if isinstance(exc, urllib.error.HTTPError):
        return exc.code in RETRYABLE_STATUS
    if isinstance(exc, (urllib.error.URLError, OSError, TimeoutError)):
        return not isinstance(exc, urllib.error.HTTPError)

    # requests.exceptions.HTTPError / RequestException (duck-typed)
    resp = getattr(exc, "response", None)
    status = getattr(resp, "status_code", None)
    if status is not None:
        return status in RETRYABLE_STATUS
    if resp is None:
        # requests connection/timeout error without a response -> transient
        return True
    return False