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:
co-authored by
Claude Fable 5
parent
f6ce112ade
commit
ea44cf41fe
@@ -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
|
||||||
|
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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():
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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
@@ -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"]
|
||||||
|
|||||||
Reference in New Issue
Block a user