Compare commits

..

3 Commits

8 changed files with 385 additions and 2 deletions

File diff suppressed because one or more lines are too long

View File

@@ -17,6 +17,7 @@ try: # pragma: no cover - import fallback for pytest collection
from .py.nodes.lora_cycler import LoraCyclerLM
from .py.nodes.lora_info import LoraInfoLM
from .py.nodes.lora_syntax_to_path import LoraSyntaxToPath
from .py.nodes.create_hook_lora import CreateHookLoraLM
from .py.metadata_collector import init as init_metadata_collector
except (
ImportError
@@ -62,6 +63,9 @@ except (
LoraSyntaxToPath = importlib.import_module(
"py.nodes.lora_syntax_to_path"
).LoraSyntaxToPath
CreateHookLoraLM = importlib.import_module(
"py.nodes.create_hook_lora"
).CreateHookLoraLM
init_metadata_collector = importlib.import_module("py.metadata_collector").init
NODE_CLASS_MAPPINGS = {
@@ -83,6 +87,7 @@ NODE_CLASS_MAPPINGS = {
LoraCyclerLM.NAME: LoraCyclerLM,
LoraInfoLM.NAME: LoraInfoLM,
LoraSyntaxToPath.NAME: LoraSyntaxToPath,
CreateHookLoraLM.NAME: CreateHookLoraLM,
}
WEB_DIRECTORY = "./web/comfyui"

View File

@@ -0,0 +1,116 @@
"""Create Hook LoRA (LoraManager) — multi-LoRA hook node compatible with ComfyUI's built-in hook pipeline.
Produces ``("HOOKS",)`` output that chains seamlessly with downstream hook consumers
(ConditioningSetProperties, SetHookKeyframes, CombineHooks, SetClipHooks, etc.).
"""
from __future__ import annotations
import logging
import os
import comfy.hooks # type: ignore
import comfy.utils # type: ignore
from ..utils.utils import get_lora_info_absolute
from .utils import (
FlexibleOptionalInputType,
any_type,
apply_lora_syntax_format,
get_loras_list,
)
logger = logging.getLogger(__name__)
class CreateHookLoraLM:
NAME = "Create Hook LoRA (LoraManager)"
CATEGORY = "Lora Manager/hooks"
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"text": (
"AUTOCOMPLETE_TEXT_LORAS",
{
"placeholder": "Search LoRAs to add...",
"tooltip": (
"Search and select LoRAs. Each LoRA gets its own "
"model/clip strength. Hooks chain with prev_hooks."
),
},
),
},
"optional": FlexibleOptionalInputType(any_type),
}
RETURN_TYPES = ("HOOKS", "STRING", "STRING")
RETURN_NAMES = ("HOOKS", "trigger_words", "active_loras")
FUNCTION = "create_hook"
def create_hook(self, text: str, **kwargs):
"""Create a HookGroup from the selected LoRAs, chained with prev_hooks.
Each active LoRA from the widget is loaded and wrapped in a WeightHook
via :func:`comfy.hooks.create_hook_lora`. All hooks are combined into a
single group and returned alongside trigger words and a human-readable
summary of the active LoRAs.
"""
del text # used by the frontend widget only
prev_hooks: comfy.hooks.HookGroup | None = kwargs.get("prev_hooks")
hook_group = prev_hooks.clone() if prev_hooks is not None else comfy.hooks.HookGroup()
all_trigger_words: list[str] = []
active_loras: list[tuple[str, float, float]] = []
for lora in get_loras_list(kwargs):
if not lora.get("active", False):
continue
lora_name = apply_lora_syntax_format(lora["name"])
model_strength = float(lora["strength"])
clip_strength = float(lora.get("clipStrength", model_strength))
# Skip useless no-op entries (both strengths are zero)
if model_strength == 0.0 and clip_strength == 0.0:
continue
lora_path, trigger_words = get_lora_info_absolute(lora_name)
if not lora_path or not os.path.isfile(lora_path):
logger.warning("LoRA '%s' not found — skipping", lora_name)
continue
try:
lora_weights = comfy.utils.load_torch_file(lora_path, safe_load=True)
lora_hooks = comfy.hooks.create_hook_lora(
lora=lora_weights,
strength_model=model_strength,
strength_clip=clip_strength,
)
except Exception:
logger.exception("Failed to load LoRA '%s' — skipping", lora_name)
continue
hook_group = hook_group.clone_and_combine(lora_hooks)
active_loras.append((lora_name, model_strength, clip_strength))
all_trigger_words.extend(trigger_words)
# Format trigger words (group mode separator)
trigger_words_text = ",, ".join(all_trigger_words) if all_trigger_words else ""
# Format active LoRAs summary
formatted_loras = []
for name, model_s, clip_s in active_loras:
if abs(model_s - clip_s) > 0.001:
formatted_loras.append(
f"<lora:{name}:{model_s}:{clip_s}>"
)
else:
formatted_loras.append(f"<lora:{name}:{model_s}>")
active_loras_text = " ".join(formatted_loras)
return (hook_group, trigger_words_text, active_loras_text)

View File

@@ -1389,7 +1389,17 @@ class DownloadManager:
# Update save directory with relative path if provided
if relative_path:
base_save_dir = save_dir
save_dir = os.path.join(save_dir, relative_path)
# Security: validate path containment after joining
resolved_dir = os.path.realpath(os.path.normpath(save_dir))
base_dir = os.path.realpath(os.path.normpath(base_save_dir))
if not resolved_dir.startswith(base_dir + os.sep) and resolved_dir != base_dir:
logger.warning(
"Path traversal detected: %s escapes %s",
resolved_dir, base_dir,
)
return {"success": False, "error": "Download path is outside allowed directory"}
# Create directory if it doesn't exist
os.makedirs(save_dir, exist_ok=True)
@@ -1827,6 +1837,9 @@ class DownloadManager:
model_tags, model_type
)
if not first_tag:
first_tag = "no tags" # Default if no tags available
# Format the template with available data
formatted_path = path_template
formatted_path = formatted_path.replace("{base_model}", mapped_base_model)
@@ -1842,6 +1855,15 @@ class DownloadManager:
if model_type == "embedding":
formatted_path = formatted_path.replace(" ", "_")
# Sanitize the resolved path to prevent path traversal:
# - Strip leading slashes (prevents os.path.join from treating path as absolute)
# - Collapse double slashes from empty placeholder substitutions
# - Strip trailing slashes for cleanliness
formatted_path = formatted_path.lstrip("/")
while "//" in formatted_path:
formatted_path = formatted_path.replace("//", "/")
formatted_path = formatted_path.rstrip("/")
return formatted_path
async def _execute_download(

View File

@@ -488,6 +488,12 @@ def calculate_relative_path_for_model(
if model_type == "embedding":
formatted_path = formatted_path.replace(" ", "_")
# Sanitize the resolved path to prevent path traversal
formatted_path = formatted_path.lstrip("/")
while "//" in formatted_path:
formatted_path = formatted_path.replace("//", "/")
formatted_path = formatted_path.rstrip("/")
return formatted_path

View File

@@ -1189,6 +1189,65 @@ def test_relative_path_sanitizes_model_and_version_placeholders():
assert relative_path == "Fancy_Model/Version_One"
def test_relative_path_empty_first_tag_fallback():
"""Test that empty first_tag falls back to 'no tags'."""
manager = DownloadManager()
settings_manager = get_settings_manager()
settings_manager.settings["download_path_templates"]["lora"] = (
"{base_model}/{first_tag}"
)
version_info = {
"baseModel": "SDXL",
"model": {"name": "Test Model", "tags": []},
"creator": {"username": "Author"},
}
relative_path = manager._calculate_relative_path(version_info, "lora")
assert relative_path == "SDXL/no tags"
def test_relative_path_empty_base_model_and_first_tag():
"""Test that empty base_model + empty first_tag does NOT produce a leading slash."""
manager = DownloadManager()
settings_manager = get_settings_manager()
settings_manager.settings["download_path_templates"]["lora"] = (
"{base_model}/{first_tag}"
)
version_info = {
"baseModel": "",
"model": {"name": "Test Model", "tags": []},
"creator": {"username": "Author"},
}
relative_path = manager._calculate_relative_path(version_info, "lora")
assert not relative_path.startswith("/")
assert relative_path == "no tags"
def test_relative_path_sanitizes_double_slashes():
"""Test that empty placeholder substitutions don't produce double slashes."""
manager = DownloadManager()
settings_manager = get_settings_manager()
settings_manager.settings["download_path_templates"]["lora"] = (
"{base_model}/{first_tag}/{author}"
)
version_info = {
"baseModel": "SDXL",
"model": {"name": "Test Model", "tags": []},
"creator": {"username": "Author"},
}
relative_path = manager._calculate_relative_path(version_info, "lora")
assert "//" not in relative_path
assert relative_path == "SDXL/no tags/Author"
def test_distribute_preview_to_entries_moves_and_copies(tmp_path):
"""Test that preview distribution moves file to first entry and copies to others."""
manager = DownloadManager()

View File

@@ -114,6 +114,38 @@ def test_calculate_relative_path_sanitizes_model_and_version_names(isolated_sett
assert relative_path == "Fancy_Model/Version_One"
def test_calculate_relative_path_sanitizes_leading_slash(isolated_settings):
"""Test that empty base_model does NOT produce a leading slash in the path."""
isolated_settings["download_path_templates"]["lora"] = "{base_model}/{first_tag}"
model_data = {
"base_model": "",
"tags": [],
"civitai": {"id": 1, "creator": {"username": "Author"}},
}
relative_path = calculate_relative_path_for_model(model_data, "lora")
assert not relative_path.startswith("/")
assert relative_path == "no tags"
def test_calculate_relative_path_sanitizes_double_slashes(isolated_settings):
"""Test that empty substitutions don't produce double slashes."""
isolated_settings["download_path_templates"]["lora"] = "{base_model}/{first_tag}/{author}"
model_data = {
"base_model": "",
"tags": [],
"civitai": {"id": 1, "creator": {"username": "Author"}},
}
relative_path = calculate_relative_path_for_model(model_data, "lora")
assert "//" not in relative_path
assert relative_path == "no tags/Author"
def test_calculate_recipe_fingerprint_filters_and_sorts():
loras = [
{"hash": "ABC", "strength": 0.1234},

View File

@@ -0,0 +1,143 @@
import { app } from "../../scripts/app.js";
import {
getActiveLorasFromNode,
updateConnectedTriggerWords,
chainCallback,
mergeLoras,
getWidgetByName,
getWidgetSerializedValue,
} from "./utils.js";
import { addLorasWidget } from "./loras_widget.js";
import { applyLoraValuesToText, debounce } from "./lora_syntax_utils.js";
import { applySelectionHighlight } from "./trigger_word_highlight.js";
import { updateConnectedLoraInfoNodes } from "./lora_info.js";
app.registerExtension({
name: "LoraManager.CreateHookLora",
async beforeRegisterNodeDef(nodeType, nodeData, app) {
if (nodeType.comfyClass === "Create Hook LoRA (LoraManager)") {
chainCallback(nodeType.prototype, "onNodeCreated", function () {
// Enable widget serialization so loras widget state is persisted
this.serialize_widgets = true;
this.addInput("prev_hooks", "HOOKS", {
shape: 7,
});
// Flags to prevent callback loops between text widget ↔ loras widget
let isUpdating = false;
let isSyncingInput = false;
// Get the text input widget (AUTOCOMPLETE_TEXT_LORAS type, created by Vue widgets)
const inputWidget = getWidgetByName(this, "text");
if (!inputWidget) {
console.warn(
"LoRA Manager: text widget not found for Create Hook LoRA"
);
return;
}
this.inputWidget = inputWidget;
const scheduleInputSync = debounce((lorasValue) => {
if (isSyncingInput) {
return;
}
isSyncingInput = true;
isUpdating = true;
try {
const nextText = applyLoraValuesToText(
inputWidget.value,
lorasValue
);
if (inputWidget.value !== nextText) {
inputWidget.value = nextText;
}
} finally {
isUpdating = false;
isSyncingInput = false;
}
});
// Create the LoRA list widget
const result = addLorasWidget(
this,
"loras",
{
onSelectionChange: (selection) => {
applySelectionHighlight(this, selection);
updateConnectedLoraInfoNodes(this, selection);
},
},
(value) => {
// Prevent recursive calls
if (isUpdating) return;
isUpdating = true;
try {
// Update connected trigger word toggles with active LoRA names
const activeLoraNames = new Set();
value.forEach((lora) => {
if (lora.active) {
activeLoraNames.add(lora.name);
}
});
updateConnectedTriggerWords(this, activeLoraNames);
} finally {
isUpdating = false;
}
scheduleInputSync(value);
}
);
this.lorasWidget = result.widget;
// Set up callback for the text input widget to trigger merge logic
inputWidget.callback = (value) => {
if (isUpdating) return;
isUpdating = true;
try {
const currentLoras = this.lorasWidget?.value || [];
const mergedLoras = mergeLoras(value, currentLoras);
if (this.lorasWidget) {
this.lorasWidget.value = mergedLoras;
}
// Update connected trigger word toggles
const activeLoraNames = getActiveLorasFromNode(this);
updateConnectedTriggerWords(this, activeLoraNames);
} finally {
isUpdating = false;
}
};
});
}
},
async loadedGraphNode(node) {
if (node.comfyClass === "Create Hook LoRA (LoraManager)") {
// Restore saved loras widget values on workflow load
let existingLoras = [];
if (node.widgets_values && node.widgets_values.length > 0) {
const savedValue = getWidgetSerializedValue(node, "loras");
existingLoras = savedValue || [];
}
// Merge the loras data from text widget with saved values
const inputWidget =
node.inputWidget || getWidgetByName(node, "text");
if (!inputWidget) {
console.warn(
"LoRA Manager: text widget not found while restoring Create Hook LoRA"
);
return;
}
const mergedLoras = mergeLoras(inputWidget.value, existingLoras);
node.lorasWidget.value = mergedLoras;
}
},
});