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
50 changes: 39 additions & 11 deletions src/xhs_archive/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from pathlib import Path
from difflib import SequenceMatcher
from typing import Annotated
from urllib.parse import urlsplit

import typer
from PIL import Image, ImageStat
Expand All @@ -33,6 +34,7 @@
from .utils import write_json

app = typer.Typer(no_args_is_help=True)
_ALLOWED_BOARD_HOSTS = frozenset({"xiaohongshu.com", "www.xiaohongshu.com"})


def _json(data: object) -> None:
Expand All @@ -59,29 +61,55 @@ def _module_importable(name: str) -> bool:
return importlib.util.find_spec(name) is not None


def _validated_board_url(value: str) -> str:
candidate = value.strip()
try:
parsed = urlsplit(candidate)
hostname = (parsed.hostname or "").rstrip(".").lower()
port = parsed.port
except ValueError as exc:
raise typer.BadParameter("收藏专辑 URL 无效。") from exc
if (
parsed.scheme != "https"
or hostname not in _ALLOWED_BOARD_HOSTS
or parsed.username is not None
or parsed.password is not None
or port not in {None, 443}
or not parsed.path.startswith("/board/")
):
raise typer.BadParameter("收藏专辑 URL 必须是 https://www.xiaohongshu.com/board/...。")
return candidate


def _resolve_board_url(board_url: str | None) -> str:
if board_url:
return board_url
return _validated_board_url(board_url)
env_url = os.environ.get("XHS_BOARD_URL")
if env_url:
return env_url
return _validated_board_url(env_url)
paste = _run_command(["/usr/bin/pbpaste"])
if paste and "xiaohongshu.com" in paste:
return paste.strip()
if paste:
try:
return _validated_board_url(paste)
except typer.BadParameter:
pass
raise typer.BadParameter("缺少收藏专辑 URL:请设置 XHS_BOARD_URL 或传入 --board-url。")


def _path_is_within(candidate: str, root: str) -> bool:
normalized_candidate = os.path.normcase(os.path.normpath(candidate))
normalized_root = os.path.normcase(os.path.normpath(root))
try:
return os.path.commonpath([normalized_candidate, normalized_root]) == normalized_root
except ValueError:
return False


@app.command()
def doctor(json_output: Annotated[bool, typer.Option("--json", help="Output JSON.")] = False) -> None:
cfg = load_config()
conda_prefix = os.environ.get("CONDA_PREFIX")
sys_prefix = Path(sys.prefix).resolve()
conda_active = False
if conda_prefix:
try:
conda_active = sys_prefix.is_relative_to(Path(conda_prefix).resolve())
except Exception:
conda_active = False
conda_active = bool(conda_prefix and _path_is_within(sys.prefix, conda_prefix))
paddle_importable = _module_importable("paddle")
paddle_device = None
if paddle_importable:
Expand Down
37 changes: 37 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
import pytest
import typer

from xhs_archive.cli import _path_is_within, _validated_board_url


@pytest.mark.parametrize(
"url",
[
"https://www.xiaohongshu.com/board/demo",
"https://xiaohongshu.com/board/b/demo?tab=notes",
"https://www.xiaohongshu.com.:443/board/demo",
],
)
def test_validated_board_url_accepts_expected_hosts(url: str) -> None:
assert _validated_board_url(f" {url} ") == url


@pytest.mark.parametrize(
"url",
[
"http://www.xiaohongshu.com/board/demo",
"https://www.xiaohongshu.com.evil.example/board/demo",
"https://www.xiaohongshu.com@evil.example/board/demo",
"https://www.xiaohongshu.com/explore/demo",
"not-a-url-containing-xiaohongshu.com",
],
)
def test_validated_board_url_rejects_untrusted_urls(url: str) -> None:
with pytest.raises(typer.BadParameter):
_validated_board_url(url)


def test_path_is_within_uses_path_boundaries() -> None:
assert _path_is_within("/opt/conda/envs/archive", "/opt/conda")
assert _path_is_within("/opt/conda", "/opt/conda")
assert not _path_is_within("/opt/conda-evil/envs/archive", "/opt/conda")