security: validate client paths as strings before resolve()/exists()

On Windows, resolving (or exists()-checking) a UNC path (\host\share)
makes the server connect to that host with the account's credentials
before any containment check runs. Client-supplied paths were resolved
first and checked against their base folder afterwards.

Add is_safe_rel/resolve_untrusted (py/config.py): string-only processing
(normpath + unresolved relative_to) that rejects UNC, drive-relative and
embedded-NUL values before the first resolve(); a UNC path is allowed
only under an allowed root that is itself UNC (NAS gallery roots keep
working). Applied to:
- gmic result (routes/gmic_routes.py): also moved the raw os.path.exists
  after the check
- gallery service: all client path/folder entries now go through
  _resolve_untrusted before _check_path_allowed; .gltf external resource
  URIs must be relative (is_safe_rel) before joining
- semantic index/similar, tagger write_meta_to_file
- video media refs ({filename, subfolder}) and settings export/import
  (project media refs, backup-ZIP entry names): is_safe_rel before join
- wildcard _safe_path: also reject drive-relative segments (C:name)

The existing resolve()+relative_to containment checks are kept as defense
in depth. Same issue as ComfyUI-VRM-Pose-Studio 2be32ea.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
ketle-man
2026-11-14 13:25:49 +09:00
co-authored by Claude Fable 5
parent f6ce112ade
commit ea44cf41fe
9 changed files with 152 additions and 39 deletions
+50
View File
@@ -1,9 +1,59 @@
import os import os
import logging import logging
import re
from pathlib import Path from pathlib import Path
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# ─── パス検査(クライアント由来のパスを resolve() する前に文字列のまま検査する)───
# Windows では UNC パス(\\host\share)は resolve()・exists() を呼んだ時点で相手ホストへ
# Windows アカウントの資格情報付きで接続してしまうため、チェックはパス操作より先に行う。
_DRIVE_RE = re.compile(r"^[A-Za-z]:")
def is_unc(value) -> bool:
"""ネットワーク共有/デバイスパス(\\\\host\\share, //host/share, \\\\?\\ 等)か。"""
return str(value or "").replace("/", "\\").startswith("\\\\")
def is_safe_rel(rel) -> bool:
"""相対パスとして安全か。UNC・ドライブ指定・絶対パス・「..」・空セグメントを
結合・resolve() より前に文字列のまま拒否する。"""
s = str(rel or "")
if not s or "\x00" in s:
return False
if s.startswith(("/", "\\")) or _DRIVE_RE.match(s):
return False
return not any(part in ("", "..") for part in re.split(r"[/\\]", s))
def resolve_untrusted(value, roots=()) -> Path:
"""クライアント由来の絶対パス文字列を検査してから resolve() して返す。
ファイルシステムに触れない文字列処理(normpath)だけで「..」を正規化・拒否し、
UNC は roots のうち UNC のルート配下に文字列上一致する場合のみ許可する
(ギャラリーのルートをネットワーク共有に置いている構成だけを通すため)。
拒否時は ValueError。"""
s = str(value or "")
if not s or "\x00" in s:
raise ValueError("Access denied: invalid path")
p = Path(os.path.normpath(s))
if ".." in p.parts:
raise ValueError("Access denied: path traversal")
if is_unc(s):
for root in roots:
if root is None or not is_unc(str(root)):
continue
try:
p.relative_to(Path(root)) # 文字列上の照合(解決しない)
return p.resolve()
except (ValueError, OSError):
continue
raise ValueError("Access denied: network path")
return p.resolve()
# Plugin root directory # Plugin root directory
PLUGIN_DIR = Path(__file__).resolve().parent.parent PLUGIN_DIR = Path(__file__).resolve().parent.parent
+9 -6
View File
@@ -14,7 +14,7 @@ from pathlib import Path
from aiohttp import web from aiohttp import web
from ..services.settings_service import SettingsService from ..services.settings_service import SettingsService
from ..config import DATA_DIR from ..config import DATA_DIR, resolve_untrusted
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -153,17 +153,20 @@ async def handle_result(request: web.Request) -> web.Response:
try: try:
body = await request.json() body = await request.json()
result_path = body.get("result_path", "") result_path = body.get("result_path", "")
if not result_path or not os.path.exists(result_path):
return web.json_response({"error": f"Result file not found: {result_path}"}, status=404)
# Path traversal guard: only allow files inside gmic_temp # Path traversal guard: only allow files inside gmic_temp.
# UNC(\\host\share)は exists()/resolve() を呼んだ時点で相手ホストへ資格情報付きで
# 接続してしまうため、パス操作より先に文字列のまま検査する(resolve_untrusted)。
allowed_dir = (Path(DATA_DIR) / "gmic_temp").resolve() allowed_dir = (Path(DATA_DIR) / "gmic_temp").resolve()
try: try:
Path(result_path).resolve().relative_to(allowed_dir) resolved = resolve_untrusted(result_path)
resolved.relative_to(allowed_dir)
except ValueError: except ValueError:
return web.json_response({"error": "Access denied: path outside gmic_temp"}, status=403) return web.json_response({"error": "Access denied: path outside gmic_temp"}, status=403)
if not resolved.is_file():
return web.json_response({"error": f"Result file not found: {result_path}"}, status=404)
with open(result_path, "rb") as f: with open(resolved, "rb") as f:
img_bytes = f.read() img_bytes = f.read()
ext = os.path.splitext(result_path)[1].lower() ext = os.path.splitext(result_path)[1].lower()
+15 -2
View File
@@ -12,7 +12,7 @@ from pathlib import Path
from aiohttp import web from aiohttp import web
from ..services.settings_service import SettingsService from ..services.settings_service import SettingsService
from ..config import DEFAULT_WORKFLOWS_DIR, DATA_DIR, VIDEO_EDIT_PROJECT_DIR from ..config import DEFAULT_WORKFLOWS_DIR, DATA_DIR, VIDEO_EDIT_PROJECT_DIR, is_safe_rel
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -397,6 +397,10 @@ def _add_video_edit_media_to_zip(zf: zipfile.ZipFile) -> None:
input_dir = _comfyui_input_dir().resolve() input_dir = _comfyui_input_dir().resolve()
seen = set() seen = set()
for subfolder, filename in _iter_video_edit_media_refs(): for subfolder, filename in _iter_video_edit_media_refs():
# プロジェクトファイルは細工できるため、UNC・絶対パス・「..」は結合・resolve() の前に
# 文字列のまま拒否する(Windows では UNC を resolve() するだけで相手ホストへ接続する)
if not is_safe_rel(filename) or (subfolder and not is_safe_rel(subfolder)):
continue
src = (input_dir / subfolder / filename).resolve() src = (input_dir / subfolder / filename).resolve()
try: try:
rel = src.relative_to(input_dir) # a crafted project file must not reach outside input/ rel = src.relative_to(input_dir) # a crafted project file must not reach outside input/
@@ -466,9 +470,15 @@ def _apply_full_backup_zip(zip_source) -> dict:
summary["skipped"].append(info.filename) summary["skipped"].append(info.filename)
continue continue
if info.filename.startswith(_VIDEO_MEDIA_PREFIX): if info.filename.startswith(_VIDEO_MEDIA_PREFIX):
media_rel = info.filename[len(_VIDEO_MEDIA_PREFIX):]
# ZIP エントリ名は細工できるため、UNC・絶対パス・「..」は結合・resolve() の前に
# 文字列のまま拒否する(Windows では UNC を resolve() するだけで相手ホストへ接続する)
if not is_safe_rel(media_rel):
summary["skipped"].append(info.filename)
continue
if input_base is None: if input_base is None:
input_base = _comfyui_input_dir().resolve() input_base = _comfyui_input_dir().resolve()
target = (input_base / info.filename[len(_VIDEO_MEDIA_PREFIX):]).resolve() target = (input_base / media_rel).resolve()
try: try:
target.relative_to(input_base) target.relative_to(input_base)
except ValueError: except ValueError:
@@ -482,6 +492,9 @@ def _apply_full_backup_zip(zip_source) -> dict:
shutil.copyfileobj(src, dst) shutil.copyfileobj(src, dst)
summary["media_restored"].append(info.filename) summary["media_restored"].append(info.filename)
continue continue
if not is_safe_rel(info.filename):
summary["skipped"].append(info.filename)
continue
target = (base / info.filename).resolve() target = (base / info.filename).resolve()
try: try:
target.relative_to(base) target.relative_to(base)
+52 -25
View File
@@ -16,6 +16,8 @@ from pathlib import Path
from typing import Any from typing import Any
from urllib.parse import unquote from urllib.parse import unquote
from ..config import is_safe_rel, resolve_untrusted
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 動画拡張子(PyAV(av)がコンテナ形式を問わず共通で扱えるため、寸法/再生時間/埋め込み # 動画拡張子(PyAV(av)がコンテナ形式を問わず共通で扱えるため、寸法/再生時間/埋め込み
@@ -270,7 +272,7 @@ class GalleryService:
"""バックグラウンドスレッドで未キャッシュ画像のprompt_cacheを構築する。 """バックグラウンドスレッドで未キャッシュ画像のprompt_cacheを構築する。
10枚処理するたびに50msスリープしてメインスレッドへの影響を最小化。""" 10枚処理するたびに50msスリープしてメインスレッドへの影響を最小化。"""
try: try:
folder = Path(folder_path).resolve() folder = self._resolve_untrusted(folder_path)
if not folder.is_dir() or not self._check_path_allowed(folder): if not folder.is_dir() or not self._check_path_allowed(folder):
return return
entries = self._scan_folder(folder) entries = self._scan_folder(folder)
@@ -333,13 +335,19 @@ class GalleryService:
pass pass
return False return False
def _resolve_untrusted(self, value) -> Path:
"""クライアント由来のパスを resolve() の前に文字列のまま検査して返す(拒否時 ValueError)。
Windows では UNC パスは resolve() した時点で相手ホストへ資格情報付きで接続してしまうため、
UNC は許可ルート自身が UNC の場合のみ、その配下を許す。"""
return resolve_untrusted(value, (self._allowed_root, self._comfy_output_root))
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
# フォルダツリー # フォルダツリー
# ────────────────────────────────────────────────────────────── # ──────────────────────────────────────────────────────────────
def list_folder_tree(self, root_path: str) -> dict: def list_folder_tree(self, root_path: str) -> dict:
"""outputフォルダのフォルダツリーを返す""" """outputフォルダのフォルダツリーを返す"""
root = Path(root_path).resolve() root = self._resolve_untrusted(root_path)
if not root.is_dir(): if not root.is_dir():
return {"error": f"Directory not found: {root_path}"} return {"error": f"Directory not found: {root_path}"}
if self._allowed_root is None: if self._allowed_root is None:
@@ -484,7 +492,7 @@ class GalleryService:
recursive: bool = False, recursive: bool = False,
) -> list[dict]: ) -> list[dict]:
"""指定フォルダ内の画像一覧を返す(recursive=Trueならサブフォルダも含める)""" """指定フォルダ内の画像一覧を返す(recursive=Trueならサブフォルダも含める)"""
folder = Path(folder_path).resolve() folder = self._resolve_untrusted(folder_path)
if not folder.is_dir(): if not folder.is_dir():
return [] return []
if not self._check_path_allowed(folder): if not self._check_path_allowed(folder):
@@ -592,7 +600,7 @@ class GalleryService:
def save_image_to_folder(self, folder_path: str, filename: str, image_data: str) -> dict: def save_image_to_folder(self, folder_path: str, filename: str, image_data: str) -> dict:
"""任意フォルダへ画像を保存する(data URL想定、同名なら上書き)。 """任意フォルダへ画像を保存する(data URL想定、同名なら上書き)。
Style Catalogのカタログ作成(スタイル名でのファイル保存)で使用。""" Style Catalogのカタログ作成(スタイル名でのファイル保存)で使用。"""
folder = Path(folder_path).resolve() folder = self._resolve_untrusted(folder_path)
if not folder.is_dir(): if not folder.is_dir():
return {"ok": False, "error": "Destination folder not found"} return {"ok": False, "error": "Destination folder not found"}
if not self._check_path_allowed(folder): if not self._check_path_allowed(folder):
@@ -740,7 +748,7 @@ class GalleryService:
def save_image_prompt(self, image_path: str, text: str) -> bool: def save_image_prompt(self, image_path: str, text: str) -> bool:
"""画像と同名の.txtサイドカーにプロンプトテキストを保存する""" """画像と同名の.txtサイドカーにプロンプトテキストを保存する"""
p = Path(image_path).resolve() p = self._resolve_untrusted(image_path)
if not p.is_file(): if not p.is_file():
return False return False
if not self._check_path_allowed(p): if not self._check_path_allowed(p):
@@ -758,7 +766,7 @@ class GalleryService:
def get_image_metadata(self, image_path: str) -> dict: def get_image_metadata(self, image_path: str) -> dict:
"""PNG/JPEGからメタデータを抽出し、保存済みメタと合わせて返す""" """PNG/JPEGからメタデータを抽出し、保存済みメタと合わせて返す"""
path = Path(image_path).resolve() path = self._resolve_untrusted(image_path)
if not path.is_file(): if not path.is_file():
return {"error": "File not found"} return {"error": "File not found"}
if not self._check_path_allowed(path): if not self._check_path_allowed(path):
@@ -1022,7 +1030,7 @@ class GalleryService:
extractPromptsLiteGraphがトップレベル系統しか拾えないことがあるため、 extractPromptsLiteGraphがトップレベル系統しか拾えないことがあるため、
プロンプト抽出専用にAPI形式を優先する(Metadataタブのプロンプト抽出と同じ優先順位)。 プロンプト抽出専用にAPI形式を優先する(Metadataタブのプロンプト抽出と同じ優先順位)。
""" """
path = Path(image_path).resolve() path = self._resolve_untrusted(image_path)
if not self._check_path_allowed(path): if not self._check_path_allowed(path):
return None return None
@@ -1083,7 +1091,7 @@ class GalleryService:
""" """
from .convert_service import OUTPUT_FORMATS, convert_audio, supports_tags from .convert_service import OUTPUT_FORMATS, convert_audio, supports_tags
p = Path(image_path).resolve() p = self._resolve_untrusted(image_path)
if not p.is_file(): if not p.is_file():
return {"ok": False, "error": "File not found"} return {"ok": False, "error": "File not found"}
if p.suffix.lower() not in AUDIO_EXTENSIONS | VIDEO_EXTENSIONS: if p.suffix.lower() not in AUDIO_EXTENSIONS | VIDEO_EXTENSIONS:
@@ -1152,7 +1160,7 @@ class GalleryService:
} }
def save_image_meta(self, image_path: str, data: dict) -> bool: def save_image_meta(self, image_path: str, data: dict) -> bool:
resolved = Path(image_path).resolve() resolved = self._resolve_untrusted(image_path)
if not self._check_path_allowed(resolved): if not self._check_path_allowed(resolved):
logger.warning("save_image_meta: path not allowed: %s", resolved) logger.warning("save_image_meta: path not allowed: %s", resolved)
return False return False
@@ -1162,7 +1170,7 @@ class GalleryService:
return self.metadata_store.save(str(resolved), data) return self.metadata_store.save(str(resolved), data)
def toggle_favorite(self, image_path: str) -> bool: def toggle_favorite(self, image_path: str) -> bool:
resolved = Path(image_path).resolve() resolved = self._resolve_untrusted(image_path)
if not self._check_path_allowed(resolved): if not self._check_path_allowed(resolved):
return False return False
# get_image_metadata()/list_images() 側はresolve()済みパスをキーにして取得する # get_image_metadata()/list_images() 側はresolve()済みパスをキーにして取得する
@@ -1178,7 +1186,11 @@ class GalleryService:
"""複数画像のお気に入りを一括設定する""" """複数画像のお気に入りを一括設定する"""
ok = fail = 0 ok = fail = 0
for p in paths: for p in paths:
resolved = Path(p).resolve() try:
resolved = self._resolve_untrusted(p)
except ValueError:
fail += 1
continue
if not self._check_path_allowed(resolved): if not self._check_path_allowed(resolved):
fail += 1 fail += 1
continue continue
@@ -1216,7 +1228,7 @@ class GalleryService:
return self.metadata_store.delete_group(name) return self.metadata_store.delete_group(name)
def add_to_group(self, image_path: str, group_name: str) -> bool: def add_to_group(self, image_path: str, group_name: str) -> bool:
resolved = Path(image_path).resolve() resolved = self._resolve_untrusted(image_path)
if not self._check_path_allowed(resolved): if not self._check_path_allowed(resolved):
return False return False
# get_image_metadata()/list_images() 側はresolve()済みパスをキーにして取得する # get_image_metadata()/list_images() 側はresolve()済みパスをキーにして取得する
@@ -1231,7 +1243,7 @@ class GalleryService:
return True return True
def remove_from_group(self, image_path: str, group_name: str) -> bool: def remove_from_group(self, image_path: str, group_name: str) -> bool:
resolved = Path(image_path).resolve() resolved = self._resolve_untrusted(image_path)
if not self._check_path_allowed(resolved): if not self._check_path_allowed(resolved):
return False return False
resolved_str = str(resolved) resolved_str = str(resolved)
@@ -1260,7 +1272,7 @@ class GalleryService:
def create_folder(self, parent_path: str, name: str) -> dict: def create_folder(self, parent_path: str, name: str) -> dict:
"""選択フォルダ内に新しいサブフォルダを作成する""" """選択フォルダ内に新しいサブフォルダを作成する"""
parent = Path(parent_path).resolve() parent = self._resolve_untrusted(parent_path)
if not parent.is_dir(): if not parent.is_dir():
return {"ok": False, "error": "Parent directory not found"} return {"ok": False, "error": "Parent directory not found"}
if not self._check_path_allowed(parent): if not self._check_path_allowed(parent):
@@ -1282,7 +1294,7 @@ class GalleryService:
def delete_folder(self, folder_path: str) -> dict: def delete_folder(self, folder_path: str) -> dict:
"""フォルダを再帰的に削除する""" """フォルダを再帰的に削除する"""
import shutil import shutil
folder = Path(folder_path).resolve() folder = self._resolve_untrusted(folder_path)
if not folder.is_dir(): if not folder.is_dir():
return {"ok": False, "error": "Directory not found"} return {"ok": False, "error": "Directory not found"}
if not self._check_path_allowed(folder): if not self._check_path_allowed(folder):
@@ -1302,7 +1314,11 @@ class GalleryService:
deleted = [] deleted = []
errors = [] errors = []
for img_path in paths: for img_path in paths:
p = Path(img_path).resolve() try:
p = self._resolve_untrusted(img_path)
except ValueError:
errors.append(f"{Path(img_path).name}: access denied")
continue
if not p.is_file(): if not p.is_file():
errors.append(f"{Path(img_path).name}: not found") errors.append(f"{Path(img_path).name}: not found")
continue continue
@@ -1323,7 +1339,7 @@ class GalleryService:
def move_images(self, paths: list, dest_folder: str) -> dict: def move_images(self, paths: list, dest_folder: str) -> dict:
"""画像ファイルを別フォルダへ移動する(複数対応)""" """画像ファイルを別フォルダへ移動する(複数対応)"""
dest = Path(dest_folder).resolve() dest = self._resolve_untrusted(dest_folder)
if not dest.is_dir(): if not dest.is_dir():
return {"ok": False, "error": "Destination folder not found", "moved": [], "errors": []} return {"ok": False, "error": "Destination folder not found", "moved": [], "errors": []}
if not self._check_path_allowed(dest): if not self._check_path_allowed(dest):
@@ -1331,7 +1347,11 @@ class GalleryService:
moved = [] moved = []
errors = [] errors = []
for img_path in paths: for img_path in paths:
p = Path(img_path).resolve() try:
p = self._resolve_untrusted(img_path)
except ValueError:
errors.append(f"{Path(img_path).name}: access denied")
continue
if not p.is_file(): if not p.is_file():
errors.append(f"{Path(img_path).name}: not found") errors.append(f"{Path(img_path).name}: not found")
continue continue
@@ -1375,7 +1395,10 @@ class GalleryService:
layers_data = [] layers_data = []
for img_path in paths: for img_path in paths:
p = Path(img_path).resolve() try:
p = self._resolve_untrusted(img_path)
except ValueError:
continue
if not p.is_file() or p.suffix.lower() not in IMAGE_EXTENSIONS: if not p.is_file() or p.suffix.lower() not in IMAGE_EXTENSIONS:
continue continue
if p.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS | MODEL3D_EXTENSIONS: if p.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS | MODEL3D_EXTENSIONS:
@@ -1543,7 +1566,7 @@ class GalleryService:
"""サーバー上に既にあるPSDファイル(Gallery上のパス)をレイヤー配列に分解する。 """サーバー上に既にあるPSDファイル(Gallery上のパス)をレイヤー配列に分解する。
Galleryから直接Image Editへ送る際、アップロードの往復無しで済ませるために使う。 Galleryから直接Image Editへ送る際、アップロードの往復無しで済ませるために使う。
""" """
p = Path(image_path).resolve() p = self._resolve_untrusted(image_path)
if not p.is_file() or p.suffix.lower() != ".psd": if not p.is_file() or p.suffix.lower() != ".psd":
raise ValueError("Not a PSD file") raise ValueError("Not a PSD file")
if not self._check_path_allowed(p): if not self._check_path_allowed(p):
@@ -1567,7 +1590,7 @@ class GalleryService:
def serve_image(self, image_path: str): def serve_image(self, image_path: str):
"""画像のPathオブジェクトを返す(ルートで使用)。 """画像のPathオブジェクトを返す(ルートで使用)。
PSDはブラウザが直接レンダリングできないため、合成済みPNGにキャッシュ変換して返す。""" PSDはブラウザが直接レンダリングできないため、合成済みPNGにキャッシュ変換して返す。"""
p = Path(image_path).resolve() p = self._resolve_untrusted(image_path)
if not p.is_file(): if not p.is_file():
return None return None
if p.suffix.lower() not in IMAGE_EXTENSIONS: if p.suffix.lower() not in IMAGE_EXTENSIONS:
@@ -1583,10 +1606,14 @@ class GalleryService:
""".gltfが相対URIで参照する外部ファイル(.bin/テクスチャ)のPathを返す。 """.gltfが相対URIで参照する外部ファイル(.bin/テクスチャ)のPathを返す。
uriはglTF仕様どおりパーセントエンコードされている。""" uriはglTF仕様どおりパーセントエンコードされている。"""
try: try:
p = Path(model_path).resolve() p = self._resolve_untrusted(model_path)
if p.suffix.lower() != ".gltf" or not p.is_file() or not self._check_path_allowed(p): if p.suffix.lower() != ".gltf" or not p.is_file() or not self._check_path_allowed(p):
return None return None
target = (p.parent / unquote(uri)).resolve() # .gltf の外部参照 URI は相対パスのみ(UNC・絶対パス・「..」は結合前に拒否)
rel_uri = unquote(uri)
if not is_safe_rel(rel_uri):
return None
target = (p.parent / rel_uri).resolve()
if target.suffix.lower() not in MODEL3D_ASSET_EXTENSIONS or not target.is_file(): if target.suffix.lower() not in MODEL3D_ASSET_EXTENSIONS or not target.is_file():
return None return None
except (OSError, ValueError): except (OSError, ValueError):
@@ -1608,7 +1635,7 @@ class GalleryService:
from io import BytesIO from io import BytesIO
from PIL import Image from PIL import Image
p = Path(model_path).resolve() p = self._resolve_untrusted(model_path)
if not p.is_file() or p.suffix.lower() not in MODEL3D_EXTENSIONS: if not p.is_file() or p.suffix.lower() not in MODEL3D_EXTENSIONS:
return {"ok": False, "error": "Not a 3D model"} return {"ok": False, "error": "Not a 3D model"}
if not self._check_path_allowed(p): if not self._check_path_allowed(p):
@@ -1702,7 +1729,7 @@ class GalleryService:
(元ファイルへのフォールバックはできないため、失敗時はNoneを返す)。 (元ファイルへのフォールバックはできないため、失敗時はNoneを返す)。
PSDはpsd_toolsでレイヤーを合成してからJPEGとしてキャッシュする(同様にフォールバック不可)。 PSDはpsd_toolsでレイヤーを合成してからJPEGとしてキャッシュする(同様にフォールバック不可)。
""" """
p = Path(image_path).resolve() p = self._resolve_untrusted(image_path)
if not p.is_file(): if not p.is_file():
return None return None
if p.suffix.lower() not in IMAGE_EXTENSIONS: if p.suffix.lower() not in IMAGE_EXTENSIONS:
+8 -2
View File
@@ -219,7 +219,10 @@ class SemanticService:
# ── 状態 ─────────────────────────────────────────────── # ── 状態 ───────────────────────────────────────────────
def _scope_entries(self, folder: str, recursive: bool) -> list[tuple[str, str, int, float]]: def _scope_entries(self, folder: str, recursive: bool) -> list[tuple[str, str, int, float]]:
p = Path(folder).resolve() try:
p = self.gallery._resolve_untrusted(folder) # UNC等は resolve() 前に文字列で拒否
except ValueError:
raise SemanticError("folder_not_allowed", str(folder))
if not p.is_dir() or not self.gallery._check_path_allowed(p): if not p.is_dir() or not self.gallery._check_path_allowed(p):
raise SemanticError("folder_not_allowed", str(folder)) raise SemanticError("folder_not_allowed", str(folder))
raw = self.gallery._scan_folder_recursive(p) if recursive else self.gallery._scan_folder(p) raw = self.gallery._scan_folder_recursive(p) if recursive else self.gallery._scan_folder(p)
@@ -539,7 +542,10 @@ class SemanticService:
def search_similar(self, path: str, folder: str, recursive: bool, limit: int, min_score: float = 0.0) -> list[dict]: def search_similar(self, path: str, folder: str, recursive: bool, limit: int, min_score: float = 0.0) -> list[dict]:
self._ensure_loaded() self._ensure_loaded()
p = Path(path).resolve() try:
p = self.gallery._resolve_untrusted(path) # UNC等は resolve() 前に文字列で拒否
except ValueError:
raise SemanticError("folder_not_allowed", path)
if not p.is_file() or not self.gallery._check_path_allowed(p): if not p.is_file() or not self.gallery._check_path_allowed(p):
raise SemanticError("folder_not_allowed", path) raise SemanticError("folder_not_allowed", path)
ap = p.as_posix() ap = p.as_posix()
+8 -1
View File
@@ -12,6 +12,8 @@ import threading
import urllib.error import urllib.error
import urllib.request import urllib.request
from pathlib import Path from pathlib import Path
from ..config import resolve_untrusted
from typing import Optional from typing import Optional
from .url_guard import HINT as _URL_HINT, is_allowed_backend_url, urlopen as _guarded_urlopen from .url_guard import HINT as _URL_HINT, is_allowed_backend_url, urlopen as _guarded_urlopen
@@ -456,7 +458,12 @@ class TaggerService:
_ALLOWED_WRITE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".gif"} _ALLOWED_WRITE_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".bmp", ".gif"}
def write_meta_to_file(self, image_path: str, tags: str) -> dict: def write_meta_to_file(self, image_path: str, tags: str) -> dict:
path = Path(image_path).resolve() try:
# UNC(\\host\share)は resolve() した時点で相手ホストへ資格情報付きで
# 接続してしまうため、文字列のまま先に拒否する
path = resolve_untrusted(image_path)
except ValueError:
return {"error": "Access denied"}
if path.suffix.lower() not in self._ALLOWED_WRITE_EXTS: if path.suffix.lower() not in self._ALLOWED_WRITE_EXTS:
return {"error": "Unsupported file type"} return {"error": "Unsupported file type"}
if not path.is_file(): if not path.is_file():
+6
View File
@@ -12,6 +12,8 @@ import os
import sys import sys
from pathlib import Path from pathlib import Path
from ..config import is_safe_rel
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
try: try:
@@ -209,6 +211,10 @@ class VideoService:
base_dir = folder_paths.get_directory_by_type(type_) base_dir = folder_paths.get_directory_by_type(type_)
if not base_dir: if not base_dir:
raise ValueError(f"Invalid type: {type_}") raise ValueError(f"Invalid type: {type_}")
# UNC・ドライブ指定・絶対パス・「..」は、結合・resolve() の前に文字列のまま拒否する
# (Windows では UNC を resolve() するだけで相手ホストへ資格情報付きで接続するため)
if not is_safe_rel(filename) or (subfolder and not is_safe_rel(subfolder)):
raise ValueError("Access denied: path outside allowed directory")
base = Path(base_dir).resolve() base = Path(base_dir).resolve()
target = (base / (subfolder or "") / filename).resolve() target = (base / (subfolder or "") / filename).resolve()
try: try:
+3 -2
View File
@@ -83,10 +83,11 @@ class WildcardService:
return None return None
# Normalize to forward slashes and strip leading/trailing slashes # Normalize to forward slashes and strip leading/trailing slashes
filename = filename.replace("\\", "/").strip("/") filename = filename.replace("\\", "/").strip("/")
# Validate each path component — block traversal and empty parts # Validate each path component — block traversal, empty parts and drive-relative
# names ("C:name"; on Python <=3.11 joinpath would replace the base path with them)
parts = filename.split("/") parts = filename.split("/")
for p in parts: for p in parts:
if not p or p == ".." or p == ".": if not p or p == ".." or p == "." or ":" in p:
return None return None
ext = Path(parts[-1]).suffix.lstrip(".").lower() ext = Path(parts[-1]).suffix.lstrip(".").lower()
if ext not in ALLOWED_EXTS: if ext not in ALLOWED_EXTS:
+1 -1
View File
@@ -1,7 +1,7 @@
[project] [project]
name = "comfyui-workflow-studio" name = "comfyui-workflow-studio"
description = "A comprehensive workflow, asset management, and generation UI plugin for ComfyUI. Manage workflows, models, images, and prompts; batch generation and Image Feeder in the GenerateUI tab; library side panel with canvas drag & drop and AI tools." description = "A comprehensive workflow, asset management, and generation UI plugin for ComfyUI. Manage workflows, models, images, and prompts; batch generation and Image Feeder in the GenerateUI tab; library side panel with canvas drag & drop and AI tools."
version = "0.9.6" version = "0.9.7"
license = { file = "LICENSE" } license = { file = "LICENSE" }
requires-python = ">= 3.10" requires-python = ">= 3.10"
dependencies = ["jinja2"] dependencies = ["jinja2"]