fix: restrict remote skill sources to GitHub

backend/package/yuxi/agents/skills/remote_install.py

@@ -8,17 +8,23 @@
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING
+from urllib.parse import urlparse
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from yuxi.agents.skills.service import import_skill_dir, is_valid_skill_slug
+from yuxi.knowledge.utils.url_fetcher import is_private_ip
 
 if TYPE_CHECKING:
     from yuxi.storage.postgres.models_business import Skill
 
 ANSI_ESCAPE_RE = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")
 CONTROL_SEQUENCE_RE = re.compile(r"\x1B\][^\x07]*(?:\x07|\x1B\\)|\x1B[\(\)][A-Za-z0-9]")
 CLI_TIMEOUT_SECONDS = 300
+GITHUB_REPO_PATTERN = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
+GITHUB_ALLOWED_HOSTS = {"github.com", "www.github.com"}
+INVALID_SOURCE_MESSAGE = "source 仅支持 GitHub owner/repo 或公共 https URL"
+PRIVATE_SOURCE_MESSAGE = "source 不能指向 localhost、私有网段或其他内网地址"
 
 
 @dataclass(slots=True)
@@ -37,7 +43,42 @@ def _normalize_source(source: str) -> str:
         raise ValueError("source 不能为空")
     if any(ch in value for ch in ("\n", "\r", "\x00")):
         raise ValueError("source 包含非法字符")
-    return value
+    if GITHUB_REPO_PATTERN.fullmatch(value):
+        return value
+
+    parsed = urlparse(value)
+    if parsed.scheme != "https" or not parsed.hostname:
+        raise ValueError(INVALID_SOURCE_MESSAGE)
+    if parsed.username or parsed.password or parsed.query or parsed.fragment:
+        raise ValueError(INVALID_SOURCE_MESSAGE)
+    if parsed.port not in (None, 443):
+        raise ValueError(PRIVATE_SOURCE_MESSAGE)
+
+    hostname = parsed.hostname.rstrip(".").lower()
+    if hostname in {"localhost"} or hostname.endswith((".local", ".localhost", ".internal")):
+        raise ValueError(PRIVATE_SOURCE_MESSAGE)
+
+    path = parsed.path.rstrip("/") or "/"
+    if hostname in GITHUB_ALLOWED_HOSTS:
+        repo_path = path.strip("/")
+        if repo_path.endswith(".git"):
+            repo_path = repo_path[:-4]
+        if GITHUB_REPO_PATTERN.fullmatch(repo_path):
+            return f"https://github.com/{repo_path}"
+
+    normalized = parsed._replace(scheme="https", netloc=hostname if parsed.port in (None, 443) else f"{hostname}:{parsed.port}", path=path)
+    return normalized.geturl().rstrip("/")
+
+
+async def _validate_source_destination(source: str) -> None:
+    if GITHUB_REPO_PATTERN.fullmatch(source):
+        return
+
+    hostname = urlparse(source).hostname
+    if not hostname:
+        raise ValueError(INVALID_SOURCE_MESSAGE)
+    if await is_private_ip(hostname):
+        raise ValueError(PRIVATE_SOURCE_MESSAGE)
 
 
 def _normalize_skill_name(skill: str) -> str:
@@ -144,6 +185,7 @@ def _create_isolated_workdir() -> tuple[str, dict[str, str], str]:
 
 async def list_remote_skills(source: str) -> list[dict[str, str]]:
     normalized_source = _normalize_source(source)
+    await _validate_source_destination(normalized_source)
 
     temp_home, env, workdir = _create_isolated_workdir()
     try:
@@ -170,6 +212,7 @@ async def install_remote_skill(
 ) -> Skill:
     normalized_source = _normalize_source(source)
     normalized_skill = _normalize_skill_name(skill)
+    await _validate_source_destination(normalized_source)
 
     temp_home, env, workdir = _create_isolated_workdir()
     try:
@@ -265,6 +308,7 @@ async def prepare_remote_skills_batch(
 ) -> RemoteSkillsBatchPreparation:
     """批量从远程仓库拉取 skill 目录，但不写数据库。"""
     normalized_source = _normalize_source(source)
+    await _validate_source_destination(normalized_source)
     if not skills:
         raise ValueError("skills 列表不能为空")
 

backend/test/unit/agents/skills/test_remote_install.py

@@ -8,6 +8,64 @@
 from yuxi.agents.skills import remote_install as svc
 
 
+def test_normalize_source_accepts_owner_repo() -> None:
+    assert svc._normalize_source("anthropics/skills") == "anthropics/skills"
+
+
+def test_normalize_source_accepts_canonical_github_url() -> None:
+    assert svc._normalize_source("https://www.github.com/anthropics/skills.git") == "https://github.com/anthropics/skills"
+
+
+def test_normalize_source_accepts_public_https_skill_url() -> None:
+    assert (
+        svc._normalize_source("https://modelscope.cn/skills/@pskoett/self-improving-agent/")
+        == "https://modelscope.cn/skills/@pskoett/self-improving-agent"
+    )
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        "http://127.0.0.1:9/repo.git",
+        "git@github.com:anthropics/skills.git",
+        "file:///tmp/skills",
+        "https://github.com/anthropics/skills?tab=readme",
+        "https://localhost/skills/demo",
+        "https://modelscope.cn:8443/skills/demo",
+    ],
+)
+def test_normalize_source_rejects_unsafe_url_shapes(source: str) -> None:
+    with pytest.raises(ValueError):
+        svc._normalize_source(source)
+
+
+@pytest.mark.asyncio
+async def test_validate_source_destination_skips_dns_for_owner_repo(monkeypatch: pytest.MonkeyPatch) -> None:
+    called = False
+
+    async def fake_is_private_ip(_hostname: str) -> bool:
+        nonlocal called
+        called = True
+        return False
+
+    monkeypatch.setattr(svc, "is_private_ip", fake_is_private_ip)
+
+    await svc._validate_source_destination("anthropics/skills")
+
+    assert called is False
+
+
+@pytest.mark.asyncio
+async def test_validate_source_destination_rejects_private_host(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def fake_is_private_ip(_hostname: str) -> bool:
+        return True
+
+    monkeypatch.setattr(svc, "is_private_ip", fake_is_private_ip)
+
+    with pytest.raises(ValueError, match="内网地址"):
+        await svc._validate_source_destination("https://modelscope.cn/skills/demo")
+
+
 def test_parse_available_skills_from_cli_output() -> None:
     output = """
     \x1b[38;5;250m███████╗\x1b[0m