55 lines
1.9 KiB
Python
55 lines
1.9 KiB
Python
"""Filesystem containment and atomic JSON helpers."""
|
|
|
|
import json
|
|
import os
|
|
import tempfile
|
|
|
|
import folder_paths
|
|
|
|
|
|
def resolve_within(base_dir, *parts):
|
|
"""Resolve a path and require it to stay inside base_dir, including through symlinks."""
|
|
base_real = os.path.realpath(base_dir)
|
|
candidate = os.path.realpath(os.path.join(base_real, *parts))
|
|
try:
|
|
if os.path.commonpath([base_real, candidate]) != base_real:
|
|
raise ValueError("Path escapes the configured directory")
|
|
except ValueError:
|
|
raise ValueError("Path escapes the configured directory")
|
|
return candidate
|
|
|
|
|
|
def get_folder_root(folder_type, path_idx=0):
|
|
paths = folder_paths.get_folder_paths(folder_type)
|
|
if not paths or not isinstance(path_idx, int) or path_idx < 0 or path_idx >= len(paths):
|
|
raise ValueError("Invalid folder path")
|
|
return os.path.realpath(paths[path_idx])
|
|
|
|
|
|
def resolve_folder_subdir(folder_type, path_idx=0, subfolder='/'):
|
|
base_dir = get_folder_root(folder_type, path_idx)
|
|
relative = "" if subfolder in (None, "", "/") else str(subfolder).strip("/\\")
|
|
return base_dir, resolve_within(base_dir, relative)
|
|
|
|
|
|
def require_filename(filename):
|
|
if not isinstance(filename, str) or not filename or os.path.basename(filename) != filename:
|
|
raise ValueError("Invalid filename")
|
|
return filename
|
|
|
|
|
|
def atomic_write_json(path, value, max_bytes=12 * 1024 * 1024):
|
|
encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
if len(encoded) > max_bytes:
|
|
raise ValueError("JSON record is too large")
|
|
fd, temporary = tempfile.mkstemp(prefix=".write-", suffix=".tmp", dir=os.path.dirname(path))
|
|
try:
|
|
with os.fdopen(fd, "wb") as output:
|
|
output.write(encoded)
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
os.replace(temporary, path)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.remove(temporary)
|