fix: close SSRF gaps in MCP OAuth discovery and web_crawl urllib fallback

src/praisonai-agents/praisonaiagents/mcp/oauth_discovery.py

@@ -28,15 +28,35 @@
 
 
 def _require_https(url: str) -> None:
-    """Reject non-HTTPS metadata URLs (loopback HTTP is allowed for testing)."""
+    """Reject OAuth URLs that are not HTTPS or that point at internal hosts.
+
+    Loopback hosts are allowed (over HTTP too) for local testing. Any other
+    host must pass :func:`~praisonaiagents.tools.url_safety.is_safe_http_url`,
+    so private, link-local (cloud metadata) and ambiguous numeric hosts are
+    refused even over HTTPS.
+    """
     parsed = urlparse(url)
-    if parsed.scheme == "https":
-        return
-    if parsed.scheme == "http" and (parsed.hostname or "") in _LOCALHOST_HOSTS:
+    if parsed.scheme in ("http", "https") and (parsed.hostname or "") in _LOCALHOST_HOSTS:
         return
-    raise ValueError(
-        f"Refusing to fetch OAuth metadata over insecure transport: {url!r}"
-    )
+    if parsed.scheme != "https":
+        raise ValueError(
+            f"Refusing OAuth request over insecure transport: {url!r}"
+        )
+    from ..tools.url_safety import is_safe_http_url
+
+    if not is_safe_http_url(url):
+        raise ValueError(
+            f"Refusing OAuth request to a private/internal address: {url!r}"
+        )
+
+
+def _require_success(resp: Any) -> None:
+    """Raise unless ``resp`` is 2xx (credential POSTs don't follow redirects)."""
+    resp.raise_for_status()
+    if not 200 <= resp.status_code < 300:
+        raise ValueError(
+            f"Unexpected HTTP {resp.status_code} from OAuth endpoint {resp.url!r}"
+        )
 
 
 def parse_www_authenticate(header: Optional[str]) -> Dict[str, str]:
@@ -62,30 +82,31 @@ def parse_www_authenticate(header: Optional[str]) -> Dict[str, str]:
     return params
 
 
-def _get_json(url: str, timeout: float = 10.0) -> Dict[str, Any]:
+def _get_json(url: str, timeout: float = 10.0, max_redirects: int = 3) -> Dict[str, Any]:
     """Fetch and parse a JSON metadata document over HTTPS.
 
-    Redirects are followed but every hop — including the final destination — is
-    re-validated with :func:`_require_https` so a redirect cannot escape the
-    HTTPS/loopback guard to an internal or plaintext endpoint (SSRF-safe).
+    Redirects are followed manually and every hop is validated with
+    :func:`_require_https` *before* it is requested, so a redirect cannot
+    reach an internal or plaintext endpoint (SSRF-safe).
     """
-    _require_https(url)
     import requests  # lazy import — optional dependency
 
-    resp = requests.get(
-        url,
-        headers={"Accept": "application/json"},
-        timeout=timeout,
-        allow_redirects=True,
-    )
-    # Re-validate every redirect hop and the final URL: allow_redirects only
-    # checks the initial URL, so a 30x to http:// or an internal host would
-    # otherwise bypass the guard.
-    for previous in getattr(resp, "history", None) or []:
-        _require_https(previous.url)
-    _require_https(resp.url)
-    resp.raise_for_status()
-    return resp.json()
+    current = url
+    for _ in range(max_redirects + 1):
+        _require_https(current)
+        resp = requests.get(
+            current,
+            headers={"Accept": "application/json"},
+            timeout=timeout,
+            allow_redirects=False,
+        )
+        location = resp.headers.get("Location")
+        if resp.status_code in (301, 302, 303, 307, 308) and location:
+            current = urljoin(current, location)
+            continue
+        resp.raise_for_status()
+        return resp.json()
+    raise ValueError(f"Too many redirects fetching OAuth metadata: {url!r}")
 
 
 def _wellknown_candidates(server_url: str) -> List[str]:

src/praisonai-agents/praisonaiagents/mcp/oauth_provider.py

@@ -268,8 +268,11 @@ def _exchange_code(
             data=data,
             headers={"Accept": "application/json"},
             timeout=timeout,
+            allow_redirects=False,  # never re-send credentials to a redirect target
         )
-        resp.raise_for_status()
+        from .oauth_discovery import _require_success
+
+        _require_success(resp)
         return self._normalize_tokens(
             resp.json(), token_endpoint=metadata["token_endpoint"]
         )
@@ -285,7 +288,7 @@ def _try_refresh(self, entry: Dict[str, Any]) -> bool:
             return False
 
         try:
-            from .oauth_discovery import _require_https
+            from .oauth_discovery import _require_https, _require_success
 
             # Defense in depth: never POST a refresh token / client secret to a
             # persisted endpoint that isn't HTTPS (or loopback).
@@ -306,8 +309,9 @@ def _try_refresh(self, entry: Dict[str, Any]) -> bool:
                 data=data,
                 headers={"Accept": "application/json"},
                 timeout=10.0,
+                allow_redirects=False,  # never re-send credentials to a redirect target
             )
-            resp.raise_for_status()
+            _require_success(resp)
             new_tokens = self._normalize_tokens(
                 resp.json(), token_endpoint=token_endpoint
             )

src/praisonai-agents/praisonaiagents/mcp/oauth_registration.py

@@ -10,25 +10,12 @@
 """
 
 from typing import Any, Dict, List, Optional
-from urllib.parse import urlparse
 
 from praisonaiagents._logging import get_logger
 
-logger = get_logger("mcp-oauth-registration")
-
-_LOCALHOST_HOSTS = {"localhost", "127.0.0.1", "::1"}
-
+from .oauth_discovery import _require_https, _require_success
 
-def _require_https(url: str) -> None:
-    """Reject non-HTTPS registration URLs (loopback HTTP allowed for testing)."""
-    parsed = urlparse(url)
-    if parsed.scheme == "https":
-        return
-    if parsed.scheme == "http" and (parsed.hostname or "") in _LOCALHOST_HOSTS:
-        return
-    raise ValueError(
-        f"Refusing to register client over insecure transport: {url!r}"
-    )
+logger = get_logger("mcp-oauth-registration")
 
 
 def register_client(
@@ -75,8 +62,9 @@ def register_client(
         json=payload,
         headers={"Content-Type": "application/json", "Accept": "application/json"},
         timeout=timeout,
+        allow_redirects=False,  # never re-send credentials to a redirect target
     )
-    resp.raise_for_status()
+    _require_success(resp)
     client_info = resp.json()
 
     if not client_info.get("client_id"):

src/praisonai-agents/praisonaiagents/tools/web_crawl_tools.py

@@ -443,12 +443,32 @@ def _crawl_with_httpx(urls: List[str]) -> List[Dict[str, Any]]:
                     else:
                         raise ValueError("Too many redirects")
             except ImportError:
-                # Fallback to urllib
+                # Fallback to urllib, walking redirects by hand so every hop
+                # is SSRF-validated before it is requested (as above).
+                import urllib.error
                 import urllib.request
-                if not _is_safe_crawl_url(url):
-                    raise ValueError("URL blocked by SSRF policy")
-                with urllib.request.urlopen(url, timeout=30) as response:
-                    content = response.read(_MAX_CRAWL_BYTES).decode('utf-8', errors='ignore')
+
+                class _NoRedirect(urllib.request.HTTPRedirectHandler):
+                    def redirect_request(self, *args, **kwargs):
+                        return None  # surface 30x as HTTPError
+
+                opener = urllib.request.build_opener(_NoRedirect)
+                current = url
+                for _ in range(max_redirects + 1):
+                    if not _is_safe_crawl_url(current):
+                        raise ValueError("Redirect target failed SSRF validation")
+                    try:
+                        with opener.open(current, timeout=30) as response:
+                            content = response.read(_MAX_CRAWL_BYTES).decode('utf-8', errors='ignore')
+                        break
+                    except urllib.error.HTTPError as e:
+                        location = e.headers.get("Location")
+                        if e.code not in redirect_statuses or not location:
+                            raise
+                        current = urllib.parse.urljoin(current, location)
+                else:
+                    raise ValueError("Too many redirects")
+                base_url = current
 
             # Convert HTML to structured Markdown using the standard library so
             # headings, lists, links, and code blocks survive on this path.

src/praisonai-agents/tests/unit/mcp/test_oauth_discovery.py

@@ -7,6 +7,14 @@
 import pytest
 
 
+@pytest.fixture
+def public_dns(monkeypatch):
+    """Treat example.com hosts as public without needing live DNS."""
+    monkeypatch.setattr(
+        "praisonaiagents.tools.url_safety.is_safe_http_url", lambda url: True
+    )
+
+
 class TestParseWWWAuthenticate:
     def test_none_returns_empty(self):
         from praisonaiagents.mcp.oauth_discovery import parse_www_authenticate
@@ -31,7 +39,7 @@ def test_rejects_plain_http_remote(self):
         with pytest.raises(ValueError):
             _require_https("http://evil.example.com/.well-known/x")
 
-    def test_allows_https(self):
+    def test_allows_https(self, public_dns):
         from praisonaiagents.mcp.oauth_discovery import _require_https
         _require_https("https://api.example.com/.well-known/x")
 
@@ -50,7 +58,7 @@ def test_builds_origin_candidates(self):
 
 
 class TestEndpointHttpsValidation:
-    def test_rejects_plaintext_advertised_endpoints(self, monkeypatch):
+    def test_rejects_plaintext_advertised_endpoints(self, monkeypatch, public_dns):
         from praisonaiagents.mcp import oauth_discovery
 
         monkeypatch.setattr(
@@ -64,10 +72,10 @@ def test_rejects_plaintext_advertised_endpoints(self, monkeypatch):
                 "token_endpoint": "http://auth.example.com/token",
             },
         )
-        with pytest.raises(ValueError):
+        with pytest.raises(ValueError, match="insecure transport"):
             oauth_discovery.discover_oauth_metadata("https://api.example.com/mcp")
 
-    def test_accepts_all_https_endpoints(self, monkeypatch):
+    def test_accepts_all_https_endpoints(self, monkeypatch, public_dns):
         from praisonaiagents.mcp import oauth_discovery
 
         monkeypatch.setattr(
@@ -87,32 +95,52 @@ def test_accepts_all_https_endpoints(self, monkeypatch):
 
 
 class TestRedirectRevalidation:
-    def test_rejects_redirect_to_plaintext(self, monkeypatch):
+    def test_rejects_redirect_before_requesting_it(self, monkeypatch):
         from praisonaiagents.mcp import oauth_discovery
 
-        class _Hop:
-            url = "http://internal.local/.well-known/x"
+        requested = []
 
         class _Resp:
-            url = "http://internal.local/.well-known/x"
-            history = [_Hop()]
-
-            def raise_for_status(self):
-                pass
-
-            def json(self):
-                return {}
+            status_code = 302
+            headers = {"Location": "http://internal.local/.well-known/x"}
 
         class _Requests:
             @staticmethod
-            def get(*a, **k):
+            def get(url, **k):
+                requested.append(url)
+                assert k["allow_redirects"] is False
                 return _Resp()
 
         monkeypatch.setitem(
             __import__("sys").modules, "requests", _Requests()
         )
         with pytest.raises(ValueError):
-            oauth_discovery._get_json("https://api.example.com/.well-known/x")
+            oauth_discovery._get_json("http://127.0.0.1:8080/.well-known/x")
+        # The redirect target was refused without ever being fetched.
+        assert requested == ["http://127.0.0.1:8080/.well-known/x"]
+
+
+class TestPrivateHostGuard:
+    @pytest.mark.parametrize("url", [
+        "https://169.254.169.254/latest/meta-data/",
+        "https://10.0.0.5/admin",
+        "https://[::ffff:169.254.169.254]/",
+        "https://0177.0.0.1/",
+    ])
+    def test_rejects_internal_https_hosts(self, url, monkeypatch):
+        from praisonaiagents.mcp.oauth_discovery import _require_https
+        monkeypatch.delenv("ALLOW_LOCAL_CRAWL", raising=False)
+        with pytest.raises(ValueError):
+            _require_https(url)
+
+    def test_registration_rejects_internal_endpoint(self, monkeypatch):
+        from praisonaiagents.mcp.oauth_registration import register_client
+        monkeypatch.delenv("ALLOW_LOCAL_CRAWL", raising=False)
+        with pytest.raises(ValueError):
+            register_client(
+                "https://10.0.0.5/register",
+                redirect_uris=["http://127.0.0.1:19876/mcp/oauth/callback"],
+            )
 
 
 class TestRegistrationGuard:
@@ -145,6 +173,29 @@ def test_returns_valid_stored_token(self, storage):
         assert provider.get_valid_token() == "tok123"
         assert provider.ensure_authenticated() == "tok123"
 
+    def test_refresh_redirect_keeps_saved_tokens(self, storage, monkeypatch, public_dns):
+        import requests
+        from praisonaiagents.mcp.oauth_provider import MCPOAuthProvider
+
+        url = "https://api.example.com/mcp"
+        tokens = {
+            "access_token": "old",
+            "refresh_token": "r1",
+            "token_endpoint": "https://auth.example.com/token",
+        }
+        storage.set_tokens("example", tokens, server_url=url)
+        storage.set_client_info("example", {"client_id": "c1"}, server_url=url)
+
+        resp = requests.Response()
+        resp.status_code = 307
+        resp.url = tokens["token_endpoint"]
+        resp._content = b'{"detail": "redirect"}'
+        monkeypatch.setattr(requests, "post", lambda *a, **k: resp)
+
+        provider = MCPOAuthProvider("example", url, storage=storage)
+        assert provider._try_refresh(storage.get_for_url("example", url)) is False
+        assert storage.get_for_url("example", url)["tokens"]["access_token"] == "old"
+
     def test_expired_token_without_refresh_returns_none(self, storage):
         import time
         from praisonaiagents.mcp.oauth_provider import MCPOAuthProvider