web.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/tools/web.py
 
"""
Web tools for URL fetching and DuckDuckGo search.
 
Purpose
-------
Provides :func:`fetch_url` (cleaned text extraction via
``trafilatura``) and :func:`web_search` (DuckDuckGo instant answers).
Falls back to clear error messages when optional dependencies are
missing so the agent can decide whether to ask the user to install
or skip the web step.
 
Responsibilities
----------------
- :func:`fetch_url` — HTTP GET with redirect-follow, then strip
  boilerplate via ``trafilatura.extract`` (or fall back to raw HTML
  when extraction is unavailable).
- :func:`web_search` — DuckDuckGo instant-answer JSON API; returns
  the abstract + related topics.
- ``TOOLS`` and ``SCHEMAS`` for auto-registration.
 
Requires: ``pip install httpx trafilatura``
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Missing optional deps return ``{"error": "<package> not installed.
  Run: pip install <package>"}`` rather than raising.
- ``fetch_url`` honours ``max_chars`` (default 10000) and signals
  truncation in the return shape.
- HTTP failures are caught and surfaced as ``{"error": "..."}``
  rather than propagating.
- All network I/O has a finite timeout (default 15 s).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.cancellation (lazy; tool_cancel_requested - skips the request once a Ctrl+C has arrived)
- oc_chat.core.egress_guard (lazy; validate_url_egress, EgressDenied and MAX_REDIRECTS - the public-host egress check on every hop)
- oc_chat.core.external_content_gate (lazy; the external-content trust gate for each fetch hop and the search source)
- oc_chat.core.resource_caps (lazy; clamp_bytes, clamp_results and MAX_BYTES - the size and result-count ceilings)
- oc_chat.tools._local_fetch (lazy; local-mode admission, its named refusal and the bounded response headers)
- oc_chat.tools._sandbox (lazy; redact_secrets - redacts the search query and error text)
- httpx (lazy; the HTTP client for both tools, optional)
- trafilatura (lazy; boilerplate stripping for fetch_url, optional)
 
See Also
--------
oc_chat/docs/conventions/CONTRACTS.md §3 Tool registry contract; §4 Tool return-
value conventions.
"""
 
from oc_diagnostics import _dbg
 
 
def _validate_hop(url: str, *, source_label: str) -> bool:
    """Validate ONE fetch hop and return ``True`` if it was admitted as LOCAL.
 
    The single place the per-hop destination rule lives, so the initial URL and
    every redirect target are judged by exactly the same code:
 
    - a **local candidate** (``http``/``https`` on ``localhost`` / ``127.0.0.1``
      / ``::1``) must be admitted by :func:`~oc_chat.tools._local_fetch.admit_local_url`
      — the ``permissive`` profile plus an explicit, allowlisted port — or it
      raises :class:`~oc_chat.tools._local_fetch.LocalFetchRefused`. A candidate is
      NEVER re-judged as a public target, so its refusal names the real reason
      (plan 232 P4) instead of the generic non-public denial;
    - everything else goes through the unchanged M8.0 WU-4 egress guard
      (§5.1.3, F2/F2a), which raises :class:`~oc_chat.core.egress_guard.EgressDenied`
      for a non-``https`` scheme, URL credentials, or any host resolving to a
      non-public address — ``127.0.0.2``, RFC1918, link-local and cloud-metadata
      included, whatever the port allowlist says.
    """
    func = "_validate_hop"
    from oc_chat.core.egress_guard import validate_url_egress
    from oc_chat.tools._local_fetch import admit_local_url, is_local_target
    if is_local_target(url):
        admit_local_url(url)
        _dbg("OC_CHAT.TOOLS", func, f"{source_label}: local mode {url}", 2)
        return True
    validate_url_egress(url, source_label=source_label)
    return False
 
 
def fetch_url(url: str, max_chars: int = 10_000) -> dict:
    """Fetch a URL and return its cleaned text content.
 
    M8.0 WU-4 (§5.1.3, F2/F2a): ``https``-only; URL credentials refused; the host
    is resolved and refused if it maps to a non-public address (loopback / RFC1918
    / link-local / cloud-metadata); redirects are followed MANUALLY with
    ``follow_redirects=False`` and each hop re-validated; ``trust_env=False`` so a
    poisoned ``HTTP(S)_PROXY`` / ``SSL_CERT_*`` cannot redirect or MITM the fetch.
 
    Plan 232 / WEB-01 adds the ONE operator-ruled exception: a **local mode** for
    a delivery check against a dev server. A hop on ``localhost`` / ``127.0.0.1``
    / ``::1`` is admitted only under the ``permissive`` profile and only when its
    explicit port is listed in the visible per-project file
    ``.oc_chat/local_fetch.json`` (:mod:`oc_chat.tools._local_fetch`); the
    admitted result additionally carries ``status``, a bounded ``headers``
    allowlist and ``local: True``, and a non-2xx status is returned as DATA
    (a 404 is exactly what a delivery check needs to see). Nothing else changes:
    every other address stays refused, the public path stays ``https``-only, and
    a local candidate that is not admitted refuses BY NAME rather than falling
    through to the public path.
    """
    func = "fetch_url"
    from oc_chat.core.resource_caps import clamp_bytes
    max_chars = clamp_bytes(max_chars)  # M8.0 WU-9a: cannot exceed the 5 MB ceiling
    _dbg("OC_CHAT.TOOLS", func, f"url={url} max_chars={max_chars}", 3)
    try:
        import httpx
    except ImportError:
        return {"error": "httpx not installed. Run: pip install httpx"}
 
    # Cooperative cancel: bail before issuing the request if a Ctrl+C arrived.
    # (The request itself is bounded by its timeout below.)
    from oc_chat.core.cancellation import tool_cancel_requested
    if tool_cancel_requested():
        return {"cancelled": True, "url": url}
 
    from urllib.parse import urljoin
    from oc_chat.core.egress_guard import EgressDenied, MAX_REDIRECTS
    from oc_chat.core.external_content_gate import (
        external_fetch_allowed, make_external_gate_block, normalize_source,
    )
    from oc_chat.tools._local_fetch import LocalFetchRefused, bounded_response_headers
    from oc_chat.tools._sandbox import redact_secrets
 
    current = url
    local = False
    status: "int | None" = None
    headers: dict = {}
    try:
        with httpx.Client(trust_env=False, follow_redirects=False, timeout=15) as client:
            for _hop in range(MAX_REDIRECTS + 1):
                # SSRF (F2/F2a): deny an internal/cloud-metadata target outright on
                # EVERY hop — never offered for confirmation. Plan 232 (WEB-01) adds
                # the single exception: a localhost/127.0.0.1/::1 hop is a LOCAL
                # CANDIDATE, admitted only by the port allowlist under `permissive`
                # and otherwise refused BY NAME — it never reaches the public path.
                local = _validate_hop(current, source_label=func)
                # M8.0 F67 (§7.3): gate the CONTENT SOURCE of THIS hop. A redirect to
                # a new host is re-gated — trusting/allowing the initial source does
                # NOT extend to a host it redirects to (egress is checked first so an
                # internal redirect target is denied outright, not offered to confirm).
                # A LOCAL hop is gated here too (plan 232): local mode narrows WHICH
                # addresses may be reached, never whether a human approves pulling
                # their content into the deciding context.
                hop_source = normalize_source(current)
                if not external_fetch_allowed(hop_source):
                    _dbg("OC_CHAT.TOOLS", func, f"external-content gate: {hop_source!r} not trusted", 2)
                    return make_external_gate_block("fetch_url", hop_source, current)
                response = client.get(current)
                if response.is_redirect and response.headers.get("location"):
                    current = urljoin(current, response.headers["location"])
                    continue
                if local:
                    # Plan 232 P3: on a LOCAL hop a non-2xx status is DATA, not an
                    # error — seeing the 404 is the whole point of a delivery check.
                    # The public path keeps raise_for_status() unchanged.
                    status = response.status_code
                    headers = bounded_response_headers(response.headers)
                else:
                    response.raise_for_status()
                raw_html = response.text
                # M8.0 WU-9a (Appendix B / F72): cap the body that enters context at
                # 5 MB (truncate + note) so a huge page cannot flood the transcript.
                from oc_chat.core.resource_caps import MAX_BYTES
                if len(raw_html) > MAX_BYTES:
                    raw_html = raw_html[:MAX_BYTES] + "\n...[body capped at 5MB]..."
                break
            else:
                return {"error": f"Too many redirects (> {MAX_REDIRECTS})"}
    except LocalFetchRefused as e:
        # Plan 232 P4: a named refusal. Never a silent fallthrough to the public
        # path, so the operator always learns WHICH rule refused: the profile,
        # the port, or the allowlist file.
        _dbg("OC_CHAT.TOOLS", func, f"local fetch refused: {e}", 2)
        return {"error": f"Local fetch refused: {e}"}
    except EgressDenied as e:
        _dbg("OC_CHAT.TOOLS", func, f"egress denied: {e}", 2)
        return {"error": f"Egress denied: {e}"}
    except Exception as e:
        return {"error": redact_secrets(str(e))}
 
    # Try trafilatura for clean text extraction; fall back to raw HTML excerpt
    try:
        import trafilatura
        text = trafilatura.extract(raw_html) or raw_html
    except ImportError:
        text = raw_html
 
    if len(text) > max_chars:
        text = text[:max_chars] + "\n...[truncated]..."
 
    result = {"url": url, "content": text, "truncated": len(text) >= max_chars}
    if local:
        # Plan 232 P3: the delivery-check shape. `status` is the integer as served
        # (a 404 is data), `headers` is the bounded allowlist — never cookies or
        # auth headers — and `local` marks which rule admitted the final hop.
        result["status"] = status
        result["headers"] = headers
        result["local"] = True
    return result
 
 
def web_search(query: str, max_results: int = 5) -> dict:
    """Search the web using DuckDuckGo and return top result snippets (no API key required)."""
    func = "web_search"
    from oc_chat.core.resource_caps import clamp_results
    max_results = clamp_results(max_results)  # M8.0 WU-9a: cannot exceed 100
    try:
        import httpx
    except ImportError:
        return {"error": "httpx not installed. Run: pip install httpx"}
 
    from oc_chat.core.cancellation import tool_cancel_requested
    if tool_cancel_requested():
        return {"cancelled": True, "query": query}
 
    # M8.0 WU-4 (§5.1.3): pinned host; trust_env=False (a poisoned proxy/cert env
    # cannot MITM or move the host); follow_redirects=False (a regional-mirror 3xx
    # cannot move egress off the pinned host); the model-controlled query is
    # secret-pattern redacted before it leaves the process (F51).
    from oc_chat.tools._sandbox import redact_secrets
    safe_query = redact_secrets(query)
 
    # M8.0 F67 (§7.3): web_search egress is the model-controlled query leaving the
    # process AND untrusted results re-entering. Confirm before the search unless
    # the (fixed) search source is session-trusted; headless fails closed.
    from oc_chat.core.external_content_gate import (
        external_fetch_allowed, make_external_gate_block,
    )
    _SEARCH_SOURCE = "api.duckduckgo.com"
    if not external_fetch_allowed(_SEARCH_SOURCE):
        _dbg("OC_CHAT.TOOLS", func, f"external-content gate: {_SEARCH_SOURCE} not trusted", 2)
        return make_external_gate_block("web_search", _SEARCH_SOURCE, f"search: {safe_query}")
 
    try:
        resp = httpx.get(
            "https://api.duckduckgo.com/",
            params={"q": safe_query, "format": "json", "no_html": "1", "skip_disambig": "1"},
            timeout=10,
            trust_env=False,
            follow_redirects=False,
        )
        resp.raise_for_status()
        data = resp.json()
    except Exception as e:
        return {"error": redact_secrets(str(e))}
 
    results = []
 
    if data.get("AbstractText"):
        results.append({"title": data.get("Heading", ""), "snippet": data["AbstractText"], "url": data.get("AbstractURL", "")})
 
    for item in data.get("RelatedTopics", []):
        if len(results) >= max_results:
            break
        if isinstance(item, dict) and item.get("Text"):
            results.append({"title": "", "snippet": item["Text"], "url": item.get("FirstURL", "")})
 
    return {"query": safe_query, "results": results}
 
 
TOOLS = {
    "fetch_url":  fetch_url,
    "web_search": web_search,
}
 
SCHEMAS = [
    {
        "type": "function",
        "function": {
            "name": "fetch_url",
            "description": "Fetch a URL and return its cleaned text content.",
            "parameters": {
                "type": "object",
                "properties": {
                    "url":       {"type": "string",  "description": "URL to fetch"},
                    "max_chars": {"type": "integer", "description": "Max characters to return (default 10000)"},
                },
                "required": ["url"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "web_search",
            "description": "Search the web using DuckDuckGo and return top result snippets.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query":       {"type": "string",  "description": "Search query"},
                    "max_results": {"type": "integer", "description": "Number of results (default 5)"},
                },
                "required": ["query"],
            },
        },
    },
]