fix(security): add SSRF validation to manage_endpoints agent tool (#6327)

The manage_endpoints agent tool accepts arbitrary user-supplied base_urls
and writes them to the database without calling check_outbound_url(),
while the HTTP route (POST /api/model-endpoints) enforces full SSRF
hardening. A prompt-injection payload smuggled into a skill, note, fetched
page, or email could instruct the model to register a cloud-metadata or
internal-network endpoint via this tool path.

Add the same check_outbound_url() guard used by the HTTP route to the
agent tool's 'add' action. Also sanitize the error message in the except
block to avoid leaking raw exception text.

Add 8 regression tests covering: cloud metadata, non-HTTP schemes,
loopback (local-first), strict mode, safe URLs, empty URLs, gopher
scheme, and error message sanitization.

Co-authored-by: isharak7m <192635824+isharak7m@users.noreply.github.com>

src/agent_tools/admin_tools.py

@@ -38,11 +38,25 @@ async def do_manage_endpoints(content: str, owner: Optional[str] = None) -> Dict
 
         elif action == "add":
             import uuid as _uuid
+            from src.url_safety import check_outbound_url
             name = args.get("name", "")
             base_url = args.get("base_url", "")
             api_key = args.get("api_key", "")
             if not base_url:
                 return {"error": "base_url is required", "exit_code": 1}
+            # SSRF hardening: validate the URL before storing it. Every entry
+            # point that accepts an external base_url must run this guard.
+            # The HTTP route (POST /api/model-endpoints) gets the same guard
+            # in #6316; this tool bypasses that route entirely, so it needs
+            # its own validation. A prompt-injection payload smuggled into a
+            # skill, note, fetched page, or email could instruct the model to
+            # register a cloud-metadata or internal-network endpoint.
+            ok, reason = check_outbound_url(
+                base_url,
+                block_private=os.getenv("MODELENDPOINT_BLOCK_PRIVATE_IPS", "false").lower() == "true",
+            )
+            if not ok:
+                return {"error": f"Rejected endpoint URL: {reason}", "exit_code": 1}
             eid = str(_uuid.uuid4())[:8]
             from datetime import datetime
             ep = ModelEndpoint(id=eid, name=name or base_url, base_url=base_url,
@@ -75,7 +89,7 @@ async def do_manage_endpoints(content: str, owner: Optional[str] = None) -> Dict
             return {"error": f"Unknown action: {action}", "exit_code": 1}
     except Exception as e:
         logger.error(f"manage_endpoints error: {e}")
-        return {"error": str(e), "exit_code": 1}
+        return {"error": "An internal error occurred while managing endpoints", "exit_code": 1}
     finally:
         db.close()