92 lines
3.8 KiB
Python
92 lines
3.8 KiB
Python
|
"""Fail-closed, allowlisted HTTP JSON search provider for criteria-first discovery."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
from urllib.request import Request, urlopen
|
||
|
|
|
||
|
|
from .website_scanner import validate_url
|
||
|
|
|
||
|
|
MAX_RESULTS = 50
|
||
|
|
MAX_RESPONSE_BYTES = 64 * 1024
|
||
|
|
TIMEOUT_SECONDS = 8
|
||
|
|
|
||
|
|
|
||
|
|
class ProviderConfigError(ValueError):
|
||
|
|
"""Provider is absent or configured in a way that is unsafe to call."""
|
||
|
|
|
||
|
|
|
||
|
|
def _config() -> tuple[str, set[str], str]:
|
||
|
|
endpoint = os.environ.get("SEARCH_PROVIDER_URL", "").strip()
|
||
|
|
allowed = {x.strip().lower().rstrip(".") for x in os.environ.get("SEARCH_PROVIDER_ALLOWED_HOSTS", "").split(",") if x.strip()}
|
||
|
|
api_key = os.environ.get("SEARCH_PROVIDER_API_KEY", "").strip()
|
||
|
|
return endpoint, allowed, api_key
|
||
|
|
|
||
|
|
|
||
|
|
def _validated_endpoint() -> tuple[str, str, str]:
|
||
|
|
endpoint, allowed, api_key = _config()
|
||
|
|
if not endpoint:
|
||
|
|
raise ProviderConfigError("not_configured")
|
||
|
|
parsed = urlparse(endpoint)
|
||
|
|
host = (parsed.hostname or "").lower().rstrip(".")
|
||
|
|
if parsed.scheme != "https" or not host or parsed.username or parsed.password or parsed.fragment or host not in allowed:
|
||
|
|
raise ProviderConfigError("unsafe_provider")
|
||
|
|
return endpoint, host, api_key
|
||
|
|
|
||
|
|
|
||
|
|
def provider_status() -> dict[str, object]:
|
||
|
|
endpoint = os.environ.get("SEARCH_PROVIDER_URL", "").strip()
|
||
|
|
if not endpoint:
|
||
|
|
return {"provider": "generic_http_json", "status": "not_configured", "configured": False, "network_enabled": False}
|
||
|
|
try:
|
||
|
|
_, host, _ = _validated_endpoint()
|
||
|
|
except ProviderConfigError as exc:
|
||
|
|
return {"provider": "generic_http_json", "status": "unsafe_configured", "configured": False, "network_enabled": False, "error": str(exc)}
|
||
|
|
return {"provider": "generic_http_json", "status": "ready", "configured": True, "network_enabled": True, "host": host, "max_results": MAX_RESULTS}
|
||
|
|
|
||
|
|
|
||
|
|
def _result_urls(payload: object, limit: int) -> list[str]:
|
||
|
|
items = payload.get("results", payload.get("items", [])) if isinstance(payload, dict) else []
|
||
|
|
if not isinstance(items, list):
|
||
|
|
raise ProviderConfigError("invalid_provider_response")
|
||
|
|
urls: list[str] = []
|
||
|
|
for item in items[:limit]:
|
||
|
|
raw = item.get("url", item.get("website", item.get("link", ""))) if isinstance(item, dict) else item
|
||
|
|
if not isinstance(raw, str) or not raw.strip():
|
||
|
|
continue
|
||
|
|
if urlparse(raw.strip()).scheme != "https":
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
safe = validate_url(raw.strip())
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
continue
|
||
|
|
if safe not in urls:
|
||
|
|
urls.append(safe)
|
||
|
|
return urls
|
||
|
|
|
||
|
|
|
||
|
|
def search(criteria: dict, limit: int) -> list[str]:
|
||
|
|
endpoint, _, api_key = _validated_endpoint()
|
||
|
|
try:
|
||
|
|
bounded = max(1, min(int(limit), MAX_RESULTS))
|
||
|
|
except (TypeError, ValueError) as exc:
|
||
|
|
raise ProviderConfigError("invalid_limits") from exc
|
||
|
|
body = json.dumps({"criteria": criteria, "limit": bounded}, separators=(",", ":"), ensure_ascii=False).encode()
|
||
|
|
headers = {"Content-Type": "application/json", "Accept": "application/json"}
|
||
|
|
if api_key:
|
||
|
|
headers["Authorization"] = "Bearer " + api_key
|
||
|
|
request = Request(endpoint, data=body, headers=headers, method="POST")
|
||
|
|
try:
|
||
|
|
with urlopen(request, timeout=TIMEOUT_SECONDS) as response:
|
||
|
|
raw = response.read(MAX_RESPONSE_BYTES + 1)
|
||
|
|
except Exception as exc:
|
||
|
|
raise ProviderConfigError("provider_unavailable") from exc
|
||
|
|
if len(raw) > MAX_RESPONSE_BYTES:
|
||
|
|
raise ProviderConfigError("provider_response_too_large")
|
||
|
|
try:
|
||
|
|
payload = json.loads(raw.decode("utf-8"))
|
||
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
||
|
|
raise ProviderConfigError("invalid_provider_response") from exc
|
||
|
|
return _result_urls(payload, bounded)
|