diff --git a/agent/skills/service.py b/agent/skills/service.py
index a34a546f..95cfb9bb 100644
--- a/agent/skills/service.py
+++ b/agent/skills/service.py
@@ -34,6 +34,27 @@ class SkillService:
         """
         self.manager = skill_manager
 
+    def _safe_skill_dir(self, name: str) -> str:
+        """Derive and validate the skill directory path.
+
+        Ensures the resolved path stays within the custom_dir root,
+        preventing path traversal via names like ``../escaped``.
+
+        :raises ValueError: if the name would escape the skills root.
+        """
+        if not name or not name.strip():
+            raise ValueError("skill name is required")
+        # Reject obvious traversal components.
+        if ".." in name or name.startswith("/") or name.startswith("\\"):
+            raise ValueError(f"invalid skill name (path traversal detected): {name!r}")
+        skill_dir = os.path.realpath(os.path.join(self.manager.custom_dir, name))
+        root = os.path.realpath(self.manager.custom_dir)
+        if not skill_dir.startswith(root + os.sep) and skill_dir != root:
+            raise ValueError(
+                f"skill name {name!r} resolves outside the skills directory"
+            )
+        return skill_dir
+
     # ------------------------------------------------------------------
     # query
     # ------------------------------------------------------------------
@@ -107,7 +128,7 @@ class SkillService:
         if not files:
             raise ValueError("skill files list is empty")
 
-        skill_dir = os.path.join(self.manager.custom_dir, name)
+        skill_dir = self._safe_skill_dir(name)
 
         tmp_dir = skill_dir + ".tmp"
         if os.path.exists(tmp_dir):
@@ -146,7 +167,7 @@ class SkillService:
             raise ValueError("package url is required")
 
         url = files[0]["url"]
-        skill_dir = os.path.join(self.manager.custom_dir, name)
+        skill_dir = self._safe_skill_dir(name)
 
         with tempfile.TemporaryDirectory() as tmp_dir:
             zip_path = os.path.join(tmp_dir, "package.zip")
@@ -217,7 +238,7 @@ class SkillService:
         if not name:
             raise ValueError("skill name is required")
 
-        skill_dir = os.path.join(self.manager.custom_dir, name)
+        skill_dir = self._safe_skill_dir(name)
         if os.path.exists(skill_dir):
             shutil.rmtree(skill_dir)
             logger.info(f"[SkillService] delete: removed directory {skill_dir}")
diff --git a/agent/tools/vision/vision.py b/agent/tools/vision/vision.py
index 19878f6d..170d2175 100644
--- a/agent/tools/vision/vision.py
+++ b/agent/tools/vision/vision.py
@@ -17,11 +17,14 @@ Provider resolution:
 """
 
 import base64
+import ipaddress
 import os
+import socket
 import subprocess
 import tempfile
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional
+from urllib.parse import urlparse
 
 import requests
 
@@ -654,6 +657,40 @@ class Vision(BaseTool):
             return api_base
         return api_base.rstrip("/") + "/v1"
 
+    @staticmethod
+    def _validate_url_safe(url: str) -> None:
+        """Reject URLs that target private/loopback/link-local addresses (SSRF guard).
+
+        Resolves the hostname to its IP address(es) and blocks any that fall
+        into non-public ranges.  Also rejects URLs with no host, non-HTTP(S)
+        schemes, or hosts that fail DNS resolution.
+
+        Raises:
+            ValueError: if the URL targets a disallowed address.
+        """
+        parsed = urlparse(url)
+        if parsed.scheme not in ("http", "https"):
+            raise ValueError(f"Unsupported URL scheme: {parsed.scheme}")
+
+        hostname = parsed.hostname
+        if not hostname:
+            raise ValueError("URL has no hostname")
+
+        try:
+            # Resolve all addresses for the hostname.
+            addr_infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
+        except socket.gaierror:
+            raise ValueError(f"Cannot resolve hostname: {hostname}")
+
+        for family, _, _, _, sockaddr in addr_infos:
+            ip_str = sockaddr[0]
+            ip = ipaddress.ip_address(ip_str)
+            if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
+                raise ValueError(
+                    f"URL resolves to a non-public address ({ip_str}), "
+                    f"request blocked for security"
+                )
+
     def _build_image_content(self, image: str) -> dict:
         """
         Build the image_url content block.
@@ -661,6 +698,7 @@ class Vision(BaseTool):
         so every bot backend can consume them without extra downloads.
         """
         if image.startswith(("http://", "https://")):
+            self._validate_url_safe(image)
             return self._download_to_data_url(image)
 
         if not os.path.isfile(image):
diff --git a/tests/test_security_ssrf_path_traversal.py b/tests/test_security_ssrf_path_traversal.py
new file mode 100644
index 00000000..6c2dbb09
--- /dev/null
+++ b/tests/test_security_ssrf_path_traversal.py
@@ -0,0 +1,194 @@
+# encoding:utf-8
+"""
+Unit tests for security fixes:
+  1. Vision tool SSRF protection (issue #2878, #2872)
+  2. Skill service path traversal protection (issue #2873)
+"""
+import os
+import sys
+import tempfile
+import types
+import unittest
+from unittest.mock import patch, MagicMock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+# Stub 'requests' if not installed so vision.py can be imported for testing.
+if "requests" not in sys.modules:
+    _requests_stub = types.ModuleType("requests")
+    _requests_stub.get = lambda *a, **k: None
+    sys.modules["requests"] = _requests_stub
+
+
+# =============================================================================
+# Vision SSRF tests
+# =============================================================================
+
+class TestVisionSSRFValidation(unittest.TestCase):
+    """Test that _validate_url_safe blocks internal/private URLs."""
+
+    def setUp(self):
+        from agent.tools.vision.vision import Vision
+        self.validate = Vision._validate_url_safe
+
+    def test_loopback_ipv4_blocked(self):
+        """127.0.0.1 must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.validate("http://127.0.0.1/canary.png")
+        self.assertIn("non-public", str(ctx.exception))
+
+    def test_loopback_localhost_blocked(self):
+        """localhost must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.validate("http://localhost/canary.png")
+        self.assertIn("non-public", str(ctx.exception))
+
+    def test_private_10_network_blocked(self):
+        """10.x.x.x RFC1918 must be rejected."""
+        with patch("socket.getaddrinfo") as mock_gai:
+            mock_gai.return_value = [
+                (2, 1, 6, "", ("10.0.0.1", 0)),
+            ]
+            with self.assertRaises(ValueError) as ctx:
+                self.validate("http://internal.corp/image.png")
+            self.assertIn("non-public", str(ctx.exception))
+
+    def test_private_172_network_blocked(self):
+        """172.16.x.x RFC1918 must be rejected."""
+        with patch("socket.getaddrinfo") as mock_gai:
+            mock_gai.return_value = [
+                (2, 1, 6, "", ("172.16.0.1", 0)),
+            ]
+            with self.assertRaises(ValueError) as ctx:
+                self.validate("http://internal.corp/image.png")
+            self.assertIn("non-public", str(ctx.exception))
+
+    def test_private_192_168_blocked(self):
+        """192.168.x.x RFC1918 must be rejected."""
+        with patch("socket.getaddrinfo") as mock_gai:
+            mock_gai.return_value = [
+                (2, 1, 6, "", ("192.168.1.1", 0)),
+            ]
+            with self.assertRaises(ValueError) as ctx:
+                self.validate("http://router.local/image.png")
+            self.assertIn("non-public", str(ctx.exception))
+
+    def test_link_local_blocked(self):
+        """169.254.x.x (link-local / cloud metadata) must be rejected."""
+        with patch("socket.getaddrinfo") as mock_gai:
+            mock_gai.return_value = [
+                (2, 1, 6, "", ("169.254.169.254", 0)),
+            ]
+            with self.assertRaises(ValueError) as ctx:
+                self.validate("http://metadata.google.internal/image.png")
+            self.assertIn("non-public", str(ctx.exception))
+
+    def test_ipv6_loopback_blocked(self):
+        """::1 (IPv6 loopback) must be rejected."""
+        with patch("socket.getaddrinfo") as mock_gai:
+            mock_gai.return_value = [
+                (10, 1, 6, "", ("::1", 0, 0, 0)),
+            ]
+            with self.assertRaises(ValueError) as ctx:
+                self.validate("http://[::1]/image.png")
+            self.assertIn("non-public", str(ctx.exception))
+
+    def test_public_url_allowed(self):
+        """A URL resolving to a public IP should pass validation."""
+        with patch("socket.getaddrinfo") as mock_gai:
+            mock_gai.return_value = [
+                (2, 1, 6, "", ("151.101.1.140", 0)),
+            ]
+            # Should not raise
+            self.validate("https://cdn.example.com/image.png")
+
+    def test_no_hostname_rejected(self):
+        """A URL with no host must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.validate("http:///path/to/image.png")
+        self.assertIn("no hostname", str(ctx.exception))
+
+    def test_non_http_scheme_rejected(self):
+        """file:// and ftp:// schemes must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.validate("file:///etc/passwd")
+        self.assertIn("scheme", str(ctx.exception))
+
+    def test_dns_failure_rejected(self):
+        """Unresolvable hostname must be rejected."""
+        import socket as sock_mod
+        with patch("socket.getaddrinfo", side_effect=sock_mod.gaierror("Name does not resolve")):
+            with self.assertRaises(ValueError) as ctx:
+                self.validate("http://nonexistent.invalid/img.png")
+            self.assertIn("Cannot resolve", str(ctx.exception))
+
+
+# =============================================================================
+# Skill service path traversal tests
+# =============================================================================
+
+class TestSkillServicePathTraversal(unittest.TestCase):
+    """Test that _safe_skill_dir blocks path traversal attempts."""
+
+    def setUp(self):
+        self.tmp_root = tempfile.mkdtemp()
+        # Create a minimal SkillManager mock with custom_dir set.
+        from agent.skills.service import SkillService
+        mock_manager = MagicMock()
+        mock_manager.custom_dir = self.tmp_root
+        self.svc = SkillService(mock_manager)
+
+    def tearDown(self):
+        import shutil
+        shutil.rmtree(self.tmp_root, ignore_errors=True)
+
+    def test_normal_name_allowed(self):
+        """A simple name like 'my-skill' should produce a valid path."""
+        result = self.svc._safe_skill_dir("my-skill")
+        expected = os.path.realpath(os.path.join(self.tmp_root, "my-skill"))
+        self.assertEqual(result, expected)
+
+    def test_dotdot_traversal_blocked(self):
+        """'../escaped' must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.svc._safe_skill_dir("../escaped")
+        self.assertIn("path traversal", str(ctx.exception))
+
+    def test_nested_dotdot_blocked(self):
+        """'foo/../../escaped' must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.svc._safe_skill_dir("foo/../../escaped")
+        self.assertIn("path traversal", str(ctx.exception))
+
+    def test_absolute_path_blocked(self):
+        """'/tmp/evil' must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.svc._safe_skill_dir("/tmp/evil")
+        self.assertIn("path traversal", str(ctx.exception))
+
+    def test_backslash_path_blocked(self):
+        r"""'\\server\share' must be rejected."""
+        with self.assertRaises(ValueError) as ctx:
+            self.svc._safe_skill_dir("\\server\\share")
+        self.assertIn("path traversal", str(ctx.exception))
+
+    def test_empty_name_blocked(self):
+        """Empty name must be rejected."""
+        with self.assertRaises(ValueError):
+            self.svc._safe_skill_dir("")
+
+    def test_whitespace_only_blocked(self):
+        """Whitespace-only name must be rejected."""
+        with self.assertRaises(ValueError):
+            self.svc._safe_skill_dir("   ")
+
+    def test_subdir_name_allowed(self):
+        """A name with a forward slash but no traversal is allowed if it stays in root."""
+        # e.g. "category/skill-name" is a valid nested skill directory
+        result = self.svc._safe_skill_dir("category/skill-name")
+        expected = os.path.realpath(os.path.join(self.tmp_root, "category/skill-name"))
+        self.assertEqual(result, expected)
+
+
+if __name__ == "__main__":
+    unittest.main()