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
co-authored by
Claude Fable 5
Claude Fable 5
parent
f6ce112ade
commit
ea44cf41fe
@@ -1,9 +1,59 @@
|
||||
import os
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
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_DIR = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ from pathlib import Path
|
||||
from aiohttp import web
|
||||
|
||||
from ..services.settings_service import SettingsService
|
||||
from ..config import DATA_DIR
|
||||
from ..config import DATA_DIR, resolve_untrusted
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -153,17 +153,20 @@ async def handle_result(request: web.Request) -> web.Response:
|
||||
try:
|
||||
body = await request.json()
|
||||
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()
|
||||
try:
|
||||
Path(result_path).resolve().relative_to(allowed_dir)
|
||||
resolved = resolve_untrusted(result_path)
|
||||
resolved.relative_to(allowed_dir)
|
||||
except ValueError:
|
||||
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()
|
||||
|
||||
ext = os.path.splitext(result_path)[1].lower()
|
||||
|
||||
@@ -12,7 +12,7 @@ from pathlib import Path
|
||||
from aiohttp import web
|
||||
|
||||
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__)
|
||||
|
||||
@@ -397,6 +397,10 @@ def _add_video_edit_media_to_zip(zf: zipfile.ZipFile) -> None:
|
||||
input_dir = _comfyui_input_dir().resolve()
|
||||
seen = set()
|
||||
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()
|
||||
try:
|
||||
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)
|
||||
continue
|
||||
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:
|
||||
input_base = _comfyui_input_dir().resolve()
|
||||
target = (input_base / info.filename[len(_VIDEO_MEDIA_PREFIX):]).resolve()
|
||||
target = (input_base / media_rel).resolve()
|
||||
try:
|
||||
target.relative_to(input_base)
|
||||
except ValueError:
|
||||
@@ -482,6 +492,9 @@ def _apply_full_backup_zip(zip_source) -> dict:
|
||||
shutil.copyfileobj(src, dst)
|
||||
summary["media_restored"].append(info.filename)
|
||||
continue
|
||||
if not is_safe_rel(info.filename):
|
||||
summary["skipped"].append(info.filename)
|
||||
continue
|
||||
target = (base / info.filename).resolve()
|
||||
try:
|
||||
target.relative_to(base)
|
||||
|
||||
@@ -16,6 +16,8 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import unquote
|
||||
|
||||
from ..config import is_safe_rel, resolve_untrusted
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 動画拡張子（PyAV(av)がコンテナ形式を問わず共通で扱えるため、寸法/再生時間/埋め込み
|
||||
@@ -270,7 +272,7 @@ class GalleryService:
|
||||
"""バックグラウンドスレッドで未キャッシュ画像のprompt_cacheを構築する。
|
||||
10枚処理するたびに50msスリープしてメインスレッドへの影響を最小化。"""
|
||||
try:
|
||||
folder = Path(folder_path).resolve()
|
||||
folder = self._resolve_untrusted(folder_path)
|
||||
if not folder.is_dir() or not self._check_path_allowed(folder):
|
||||
return
|
||||
entries = self._scan_folder(folder)
|
||||
@@ -333,13 +335,19 @@ class GalleryService:
|
||||
pass
|
||||
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:
|
||||
"""outputフォルダのフォルダツリーを返す"""
|
||||
root = Path(root_path).resolve()
|
||||
root = self._resolve_untrusted(root_path)
|
||||
if not root.is_dir():
|
||||
return {"error": f"Directory not found: {root_path}"}
|
||||
if self._allowed_root is None:
|
||||
@@ -484,7 +492,7 @@ class GalleryService:
|
||||
recursive: bool = False,
|
||||
) -> list[dict]:
|
||||
"""指定フォルダ内の画像一覧を返す（recursive=Trueならサブフォルダも含める）"""
|
||||
folder = Path(folder_path).resolve()
|
||||
folder = self._resolve_untrusted(folder_path)
|
||||
if not folder.is_dir():
|
||||
return []
|
||||
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:
|
||||
"""任意フォルダへ画像を保存する（data URL想定、同名なら上書き）。
|
||||
Style Catalogのカタログ作成（スタイル名でのファイル保存）で使用。"""
|
||||
folder = Path(folder_path).resolve()
|
||||
folder = self._resolve_untrusted(folder_path)
|
||||
if not folder.is_dir():
|
||||
return {"ok": False, "error": "Destination folder not found"}
|
||||
if not self._check_path_allowed(folder):
|
||||
@@ -740,7 +748,7 @@ class GalleryService:
|
||||
|
||||
def save_image_prompt(self, image_path: str, text: str) -> bool:
|
||||
"""画像と同名の.txtサイドカーにプロンプトテキストを保存する"""
|
||||
p = Path(image_path).resolve()
|
||||
p = self._resolve_untrusted(image_path)
|
||||
if not p.is_file():
|
||||
return False
|
||||
if not self._check_path_allowed(p):
|
||||
@@ -758,7 +766,7 @@ class GalleryService:
|
||||
|
||||
def get_image_metadata(self, image_path: str) -> dict:
|
||||
"""PNG/JPEGからメタデータを抽出し、保存済みメタと合わせて返す"""
|
||||
path = Path(image_path).resolve()
|
||||
path = self._resolve_untrusted(image_path)
|
||||
if not path.is_file():
|
||||
return {"error": "File not found"}
|
||||
if not self._check_path_allowed(path):
|
||||
@@ -1022,7 +1030,7 @@ class GalleryService:
|
||||
extractPromptsLiteGraphがトップレベル系統しか拾えないことがあるため、
|
||||
プロンプト抽出専用にAPI形式を優先する（Metadataタブのプロンプト抽出と同じ優先順位）。
|
||||
"""
|
||||
path = Path(image_path).resolve()
|
||||
path = self._resolve_untrusted(image_path)
|
||||
if not self._check_path_allowed(path):
|
||||
return None
|
||||
|
||||
@@ -1083,7 +1091,7 @@ class GalleryService:
|
||||
"""
|
||||
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():
|
||||
return {"ok": False, "error": "File not found"}
|
||||
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:
|
||||
resolved = Path(image_path).resolve()
|
||||
resolved = self._resolve_untrusted(image_path)
|
||||
if not self._check_path_allowed(resolved):
|
||||
logger.warning("save_image_meta: path not allowed: %s", resolved)
|
||||
return False
|
||||
@@ -1162,7 +1170,7 @@ class GalleryService:
|
||||
return self.metadata_store.save(str(resolved), data)
|
||||
|
||||
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):
|
||||
return False
|
||||
# get_image_metadata()/list_images() 側はresolve()済みパスをキーにして取得する
|
||||
@@ -1178,7 +1186,11 @@ class GalleryService:
|
||||
"""複数画像のお気に入りを一括設定する"""
|
||||
ok = fail = 0
|
||||
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):
|
||||
fail += 1
|
||||
continue
|
||||
@@ -1216,7 +1228,7 @@ class GalleryService:
|
||||
return self.metadata_store.delete_group(name)
|
||||
|
||||
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):
|
||||
return False
|
||||
# get_image_metadata()/list_images() 側はresolve()済みパスをキーにして取得する
|
||||
@@ -1231,7 +1243,7 @@ class GalleryService:
|
||||
return True
|
||||
|
||||
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):
|
||||
return False
|
||||
resolved_str = str(resolved)
|
||||
@@ -1260,7 +1272,7 @@ class GalleryService:
|
||||
|
||||
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():
|
||||
return {"ok": False, "error": "Parent directory not found"}
|
||||
if not self._check_path_allowed(parent):
|
||||
@@ -1282,7 +1294,7 @@ class GalleryService:
|
||||
def delete_folder(self, folder_path: str) -> dict:
|
||||
"""フォルダを再帰的に削除する"""
|
||||
import shutil
|
||||
folder = Path(folder_path).resolve()
|
||||
folder = self._resolve_untrusted(folder_path)
|
||||
if not folder.is_dir():
|
||||
return {"ok": False, "error": "Directory not found"}
|
||||
if not self._check_path_allowed(folder):
|
||||
@@ -1302,7 +1314,11 @@ class GalleryService:
|
||||
deleted = []
|
||||
errors = []
|
||||
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():
|
||||
errors.append(f"{Path(img_path).name}: not found")
|
||||
continue
|
||||
@@ -1323,7 +1339,7 @@ class GalleryService:
|
||||
|
||||
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():
|
||||
return {"ok": False, "error": "Destination folder not found", "moved": [], "errors": []}
|
||||
if not self._check_path_allowed(dest):
|
||||
@@ -1331,7 +1347,11 @@ class GalleryService:
|
||||
moved = []
|
||||
errors = []
|
||||
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():
|
||||
errors.append(f"{Path(img_path).name}: not found")
|
||||
continue
|
||||
@@ -1375,7 +1395,10 @@ class GalleryService:
|
||||
|
||||
layers_data = []
|
||||
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:
|
||||
continue
|
||||
if p.suffix.lower() in VIDEO_EXTENSIONS | AUDIO_EXTENSIONS | MODEL3D_EXTENSIONS:
|
||||
@@ -1543,7 +1566,7 @@ class GalleryService:
|
||||
"""サーバー上に既にあるPSDファイル(Gallery上のパス)をレイヤー配列に分解する。
|
||||
Galleryから直接Image Editへ送る際、アップロードの往復無しで済ませるために使う。
|
||||
"""
|
||||
p = Path(image_path).resolve()
|
||||
p = self._resolve_untrusted(image_path)
|
||||
if not p.is_file() or p.suffix.lower() != ".psd":
|
||||
raise ValueError("Not a PSD file")
|
||||
if not self._check_path_allowed(p):
|
||||
@@ -1567,7 +1590,7 @@ class GalleryService:
|
||||
def serve_image(self, image_path: str):
|
||||
"""画像のPathオブジェクトを返す（ルートで使用）。
|
||||
PSDはブラウザが直接レンダリングできないため、合成済みPNGにキャッシュ変換して返す。"""
|
||||
p = Path(image_path).resolve()
|
||||
p = self._resolve_untrusted(image_path)
|
||||
if not p.is_file():
|
||||
return None
|
||||
if p.suffix.lower() not in IMAGE_EXTENSIONS:
|
||||
@@ -1583,10 +1606,14 @@ class GalleryService:
|
||||
""".gltfが相対URIで参照する外部ファイル(.bin/テクスチャ)のPathを返す。
|
||||
uriはglTF仕様どおりパーセントエンコードされている。"""
|
||||
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):
|
||||
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():
|
||||
return None
|
||||
except (OSError, ValueError):
|
||||
@@ -1608,7 +1635,7 @@ class GalleryService:
|
||||
from io import BytesIO
|
||||
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:
|
||||
return {"ok": False, "error": "Not a 3D model"}
|
||||
if not self._check_path_allowed(p):
|
||||
@@ -1702,7 +1729,7 @@ class GalleryService:
|
||||
（元ファイルへのフォールバックはできないため、失敗時はNoneを返す）。
|
||||
PSDはpsd_toolsでレイヤーを合成してからJPEGとしてキャッシュする（同様にフォールバック不可）。
|
||||
"""
|
||||
p = Path(image_path).resolve()
|
||||
p = self._resolve_untrusted(image_path)
|
||||
if not p.is_file():
|
||||
return None
|
||||
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]]:
|
||||
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):
|
||||
raise SemanticError("folder_not_allowed", str(folder))
|
||||
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]:
|
||||
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):
|
||||
raise SemanticError("folder_not_allowed", path)
|
||||
ap = p.as_posix()
|
||||
|
||||
@@ -12,6 +12,8 @@ import threading
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
from ..config import resolve_untrusted
|
||||
from typing import Optional
|
||||
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"}
|
||||
|
||||
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:
|
||||
return {"error": "Unsupported file type"}
|
||||
if not path.is_file():
|
||||
|
||||
@@ -12,6 +12,8 @@ import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from ..config import is_safe_rel
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
try:
|
||||
@@ -209,6 +211,10 @@ class VideoService:
|
||||
base_dir = folder_paths.get_directory_by_type(type_)
|
||||
if not base_dir:
|
||||
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()
|
||||
target = (base / (subfolder or "") / filename).resolve()
|
||||
try:
|
||||
|
||||
@@ -83,10 +83,11 @@ class WildcardService:
|
||||
return None
|
||||
# Normalize to forward slashes and strip leading/trailing slashes
|
||||
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("/")
|
||||
for p in parts:
|
||||
if not p or p == ".." or p == ".":
|
||||
if not p or p == ".." or p == "." or ":" in p:
|
||||
return None
|
||||
ext = Path(parts[-1]).suffix.lstrip(".").lower()
|
||||
if ext not in ALLOWED_EXTS:
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
[project]
|
||||
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."
|
||||
version = "0.9.6"
|
||||
version = "0.9.7"
|
||||
license = { file = "LICENSE" }
|
||||
requires-python = ">= 3.10"
|
||||
dependencies = ["jinja2"]
|
||||
|
||||
Reference in New Issue
Block a user