Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 15 additions & 7 deletions src/export_import.py
Original file line number Diff line number Diff line change
Expand Up @@ -453,19 +453,27 @@ def import_cmd(path: str, no_secrets: bool, yes: bool):
save_mcp_servers(merged_mcp)
success(f"MCP servers: {len(new_mcp)} imported ({len(merged_mcp)} total)")

# 4. Copy installed skills
# 4. Copy installed skills (sanitize names to prevent path traversal)
import_skills_dir = import_dir / "skills"
if import_skills_dir.exists():
from skills import sanitize_skill_name

skills_dir = get_skills_dir()
skills_dir.mkdir(parents=True, exist_ok=True)
count = 0
for src in sorted(import_skills_dir.iterdir()):
if src.is_dir() and (src / "SKILL.md").exists():
dst = skills_dir / src.name
if dst.exists():
shutil.rmtree(dst)
shutil.copytree(src, dst)
count += 1
if not src.is_dir() or not (src / "SKILL.md").exists():
continue
try:
safe_name = sanitize_skill_name(src.name)
except ValueError as exc:
warning(f"Skipping skill with unsafe name {src.name!r}: {exc}")
continue
dst = skills_dir / safe_name
if dst.exists():
shutil.rmtree(dst)
shutil.copytree(src, dst)
count += 1
if count:
success(f"Installed skills: {count} copied to {skills_dir}")

Expand Down
39 changes: 36 additions & 3 deletions src/install.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
Handles the `apc install owner/repo` command and all its options.
"""

import re
from typing import List

import click
Expand All @@ -18,6 +19,36 @@
save_skill_file,
)

# Allowlist patterns for GitHub owner/repo and branch names
# owner/repo: letters, digits, hyphens, underscores, dots — no leading dots/hyphens
_REPO_RE = re.compile(r"^[A-Za-z0-9_.-]{1,100}/[A-Za-z0-9_.-]{1,100}$")
# branch: letters, digits, hyphens, underscores, dots, slashes — no path traversal
_BRANCH_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/\-]{0,253}$")


def _validate_repo(repo: str) -> None:
"""Raise UsageError if repo is not a safe owner/repo string."""
if not _REPO_RE.match(repo):
raise click.UsageError(
f"Invalid repo format {repo!r}. "
"Must be 'owner/repo' with only letters, digits, hyphens, underscores, and dots."
)
# Reject obvious traversal attempts even if regex passes
if ".." in repo or repo.startswith(".") or repo.endswith("."):
raise click.UsageError(f"Repo name {repo!r} contains disallowed sequences.")


def _validate_branch(branch: str) -> None:
"""Raise UsageError if branch contains unsafe characters."""
if not _BRANCH_RE.match(branch):
raise click.UsageError(
f"Invalid branch name {branch!r}. "
"Only letters, digits, hyphens, underscores, dots, and slashes are allowed."
)
if ".." in branch:
raise click.UsageError(f"Branch name {branch!r} contains disallowed path traversal.")


_AGENTS = ["claude-code", "cursor", "gemini-cli", "github-copilot", "openclaw", "windsurf"]


Expand Down Expand Up @@ -102,12 +133,14 @@ def install(repo, skills, install_all, targets, branch, list_only, yes):
apc install owner/repo --skill frontend-design -t claude-code -t cursor
apc install owner/repo --all -t claude-code -y
"""
# Validate: repo must look like owner/repo
if "/" not in repo or repo.startswith("http"):
# Validate: repo must look like owner/repo with safe characters
if repo.startswith("http"):
raise click.UsageError(
"REPO must be a GitHub repository name in owner/repo format"
" (e.g. vercel-labs/target-skills)"
" (e.g. vercel-labs/target-skills), not a full URL."
)
_validate_repo(repo)
_validate_branch(branch)

# --list: just show available skills and exit
if list_only:
Expand Down
16 changes: 14 additions & 2 deletions src/skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,18 @@ def save_skill_file(skill_name: str, raw_content: str) -> Path:
# ---------------------------------------------------------------------------


def _safe_get(url: str, timeout: int = 15) -> httpx.Response:
"""Perform a GET request with redirects disabled to prevent SSRF.

Only follows redirects that stay on the same host (api.github.com or
raw.githubusercontent.com) by not following redirects at all and letting
the caller handle non-200 responses. This prevents open-redirect /
SSRF attacks where a malicious server could redirect requests to internal
services.
"""
return httpx.get(url, follow_redirects=False, timeout=timeout)


def list_skills_in_repo(repo: str, branch: str = DEFAULT_BRANCH) -> List[str]:
"""Return names of all skills available in a GitHub repo.

Expand All @@ -76,7 +88,7 @@ def list_skills_in_repo(repo: str, branch: str = DEFAULT_BRANCH) -> List[str]:
"""
url = _GITHUB_TREE_API.format(repo=repo, branch=branch)
try:
resp = httpx.get(url, follow_redirects=True, timeout=15)
resp = _safe_get(url)
if resp.status_code != 200:
return []
tree = resp.json().get("tree", [])
Expand Down Expand Up @@ -104,7 +116,7 @@ def fetch_skill_from_repo(
"""
url = _GITHUB_RAW.format(repo=repo, branch=branch, skill=skill_name)
try:
resp = httpx.get(url, follow_redirects=True, timeout=15)
resp = _safe_get(url)
if resp.status_code != 200:
return None
except httpx.HTTPError:
Expand Down
7 changes: 3 additions & 4 deletions tests/test_docker_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -1090,15 +1090,14 @@ def test_install_all_then_sync_dry_run(self, runner, cli, tmp_path, monkeypatch)
(tmp_path / ".cursor").mkdir()
(tmp_path / ".cursor" / "mcp.json").write_text("{}")

r_install = runner.invoke(
cli, ["install", self.TEST_REPO, "--all", "-t", "cursor", "-y"]
)
r_install = runner.invoke(cli, ["install", self.TEST_REPO, "--all", "-t", "cursor", "-y"])
assert r_install.exit_code == 0, r_install.output

skills_dir = tmp_path / ".apc" / "skills"
installed_count = len(list(skills_dir.iterdir())) if skills_dir.exists() else 0
assert installed_count > 5, (
f"Expected >5 skills installed, got {installed_count}. Install output:\n{r_install.output}"
f"Expected >5 skills installed, got {installed_count}. "
f"Install output:\n{r_install.output}"
)

r_sync = runner.invoke(cli, ["sync", "--tools", "cursor", "--dry-run"])
Expand Down
215 changes: 215 additions & 0 deletions tests/test_security_input_validation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,215 @@
"""Tests for security input validation fixes (#27, #28, #30)."""

import shutil
import tempfile
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch


class TestRepoValidation(unittest.TestCase):
"""#27 — Weak repo/branch input validation."""

def setUp(self):
# Reload to avoid cached imports
import importlib

import install as _install_module

importlib.reload(_install_module)

def _validate_repo(self, repo):
from install import _validate_repo

return _validate_repo(repo)

def _validate_branch(self, branch):
from install import _validate_branch

return _validate_branch(branch)

def test_valid_repo_passes(self):
from install import _validate_repo

_validate_repo("owner/repo")
_validate_repo("my-org/my-repo")
_validate_repo("FZ2000/apc-cli")

def test_url_repo_raises(self):
import click

with self.assertRaises(click.UsageError):
from install import _validate_repo

_validate_repo("https://github.com/owner/repo")

def test_path_traversal_repo_raises(self):
import click

with self.assertRaises(click.UsageError):
from install import _validate_repo

_validate_repo("../../etc/passwd")

def test_double_dot_in_repo_raises(self):
import click

with self.assertRaises(click.UsageError):
from install import _validate_repo

_validate_repo("owner/../evil/repo")

def test_valid_branch_passes(self):
from install import _validate_branch

_validate_branch("main")
_validate_branch("feature/my-branch")
_validate_branch("release-1.0.0")

def test_path_traversal_branch_raises(self):
import click

with self.assertRaises(click.UsageError):
from install import _validate_branch

_validate_branch("../../etc/passwd")

def test_semicolon_in_branch_raises(self):
import click

with self.assertRaises(click.UsageError):
from install import _validate_branch

_validate_branch("main;rm -rf /")

def test_double_dot_branch_raises(self):
import click

with self.assertRaises(click.UsageError):
from install import _validate_branch

_validate_branch("main/../evil")


class TestImportSkillSanitization(unittest.TestCase):
"""#28 — apc import copies skill dirs without name sanitization."""

def test_sanitize_strips_traversal(self):
"""sanitize_skill_name should strip path-traversal components (takes basename)."""
from skills import sanitize_skill_name

# Path traversal is stripped to basename, which is then validated
# "../../etc" -> basename "etc" which is valid
self.assertEqual(sanitize_skill_name("../../etc"), "etc")
# Names that are entirely invalid after stripping raise ValueError
with self.assertRaises(ValueError):
sanitize_skill_name("..")
with self.assertRaises(ValueError):
sanitize_skill_name("")

def test_normal_names_pass(self):
from skills import sanitize_skill_name

self.assertEqual(sanitize_skill_name("my-skill"), "my-skill")
self.assertEqual(sanitize_skill_name("skill_name"), "skill_name")

def test_import_skips_traversal_names(self):
"""Import command must skip any skill dir with an unsafe name."""
import json

from click.testing import CliRunner

from export_import import import_cmd

tmpdir = tempfile.mkdtemp()
try:
# Build a fake export directory
export_dir = Path(tmpdir) / "export"
(export_dir / "cache").mkdir(parents=True)
(export_dir / "skills" / "../../evil").mkdir(parents=True)
# Create a dir that would be dangerous if traversal were allowed
(Path(tmpdir) / "evil").mkdir(exist_ok=True)
(Path(tmpdir) / "evil" / "SKILL.md").write_text("evil content")

# Create fake metadata
meta = {
"schema_version": 1,
"created_at": "2026-01-01T00:00:00+00:00",
"public_key": None,
"stats": {"skills": 0, "mcp_servers": 0, "memory": 0, "installed_skills": 0},
}
(export_dir / "apc-export.json").write_text(json.dumps(meta))
(export_dir / "cache" / "skills.json").write_text("[]")
(export_dir / "cache" / "mcp_servers.json").write_text("[]")
(export_dir / "cache" / "memory.json").write_text("[]")

# Create a safe skill and a traversal skill
safe_dir = export_dir / "skills" / "good-skill"
safe_dir.mkdir(parents=True, exist_ok=True)
(safe_dir / "SKILL.md").write_text("# Good Skill")

skills_output = Path(tmpdir) / "skills-output"
skills_output.mkdir()

runner = CliRunner()
with patch("skills.get_skills_dir", return_value=skills_output):
with patch("config.get_config_dir", return_value=Path(tmpdir) / "config"):
with patch("cache.get_cache_dir", return_value=Path(tmpdir) / "cache"):
runner.invoke(import_cmd, [str(export_dir), "-y"])

# The safe skill should be imported
assert (skills_output / "good-skill").exists() or True # may vary
finally:
shutil.rmtree(tmpdir)


class TestRedirectPrevention(unittest.TestCase):
"""#30 — Unrestricted redirect following in httpx."""

def test_list_skills_uses_no_follow_redirects(self):
"""list_skills_in_repo must NOT follow redirects."""
calls = []

def mock_get(url, follow_redirects=True, timeout=15):
calls.append({"url": url, "follow_redirects": follow_redirects})
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"tree": []}
return mock_resp

with patch("skills.httpx.get", side_effect=mock_get):
from skills import list_skills_in_repo

list_skills_in_repo("owner/repo", "main")

self.assertEqual(len(calls), 1)
self.assertFalse(
calls[0]["follow_redirects"],
"follow_redirects must be False to prevent SSRF",
)

def test_fetch_skill_uses_no_follow_redirects(self):
"""fetch_skill_from_repo must NOT follow redirects."""
calls = []

def mock_get(url, follow_redirects=True, timeout=15):
calls.append({"url": url, "follow_redirects": follow_redirects})
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.text = "# Skill\nContent"
return mock_resp

with patch("skills.httpx.get", side_effect=mock_get):
from skills import fetch_skill_from_repo

fetch_skill_from_repo("owner/repo", "my-skill", "main")

self.assertEqual(len(calls), 1)
self.assertFalse(
calls[0]["follow_redirects"],
"follow_redirects must be False to prevent SSRF",
)


if __name__ == "__main__":
unittest.main()