Merge ce445263e6 into 7ac999bf30

* chore: update workflow templates to v0.8.9

* Update requirements.txt

comfy/cli_args.py

@@ -47,6 +47,9 @@ parser.add_argument("--extra-model-paths-config", type=str, default=None, metava
 parser.add_argument("--output-directory", type=str, default=None, help="Set the ComfyUI output directory. Overrides --base-directory.")
 parser.add_argument("--temp-directory", type=str, default=None, help="Set the ComfyUI temp directory (default is in the ComfyUI directory). Overrides --base-directory.")
 parser.add_argument("--input-directory", type=str, default=None, help="Set the ComfyUI input directory. Overrides --base-directory.")
+parser.add_argument("--enable-landlock", action="store_true", help="Use the Linux Landlock LSM to restrict filesystem writes to known ComfyUI and cache directories.")
+parser.add_argument("--landlock-allow-writable", action="append", default=[], metavar="PATH", help="Extra directories that remain writable when --enable-landlock is set. Can be provided multiple times.")
+parser.add_argument("--landlock-allow-readable", action="append", default=[], metavar="PATH", help="Extra directories to allow read access when --enable-landlock is set. Can be provided multiple times.")
 parser.add_argument("--auto-launch", action="store_true", help="Automatically launch ComfyUI in the default browser.")
 parser.add_argument("--disable-auto-launch", action="store_true", help="Disable auto launching the browser.")
 parser.add_argument("--cuda-device", type=int, default=None, metavar="DEVICE_ID", help="Set the id of the cuda device this instance will use. All other devices will not be visible.")

comfy/clip_vision.py

@@ -66,6 +66,7 @@ class ClipVisionModel():
         outputs = Output()
         outputs["last_hidden_state"] = out[0].to(comfy.model_management.intermediate_device())
         outputs["image_embeds"] = out[2].to(comfy.model_management.intermediate_device())
+        outputs["image_sizes"] = [pixel_values.shape[1:]] * pixel_values.shape[0]
         if self.return_all_hidden_states:
             all_hs = out[1].to(comfy.model_management.intermediate_device())
             outputs["penultimate_hidden_states"] = all_hs[:, -2]

comfy/supported_models.py

@@ -763,7 +763,7 @@ class Flux2(Flux):
 
     def __init__(self, unet_config):
         super().__init__(unet_config)
-        self.memory_usage_factor = self.memory_usage_factor * (2.0 * 2.0) * 2.36
+        self.memory_usage_factor = self.memory_usage_factor * (2.0 * 2.0) * (unet_config['hidden_size'] / 2604)
 
     def get_model(self, state_dict, prefix="", device=None):
         out = model_base.Flux2(self, device=device)

comfy_api_nodes/apis/bytedance_api.py

@@ -65,11 +65,13 @@ class TaskImageContent(BaseModel):
 class Text2VideoTaskCreationRequest(BaseModel):
     model: str = Field(...)
     content: list[TaskTextContent] = Field(..., min_length=1)
+    generate_audio: bool | None = Field(...)
 
 
 class Image2VideoTaskCreationRequest(BaseModel):
     model: str = Field(...)
     content: list[TaskTextContent | TaskImageContent] = Field(..., min_length=2)
+    generate_audio: bool | None = Field(...)
 
 
 class TaskCreationResponse(BaseModel):
@@ -141,4 +143,9 @@ VIDEO_TASKS_EXECUTION_TIME = {
         "720p": 65,
         "1080p": 100,
     },
+    "seedance-1-5-pro-251215": {
+        "480p": 80,
+        "720p": 100,
+        "1080p": 150,
+    },
 }

comfy_api_nodes/nodes_bytedance.py

@@ -477,7 +477,12 @@ class ByteDanceTextToVideoNode(IO.ComfyNode):
             inputs=[
                 IO.Combo.Input(
                     "model",
-                    options=["seedance-1-0-pro-250528", "seedance-1-0-lite-t2v-250428", "seedance-1-0-pro-fast-251015"],
+                    options=[
+                        "seedance-1-5-pro-251215",
+                        "seedance-1-0-pro-250528",
+                        "seedance-1-0-lite-t2v-250428",
+                        "seedance-1-0-pro-fast-251015",
+                    ],
                     default="seedance-1-0-pro-fast-251015",
                 ),
                 IO.String.Input(
@@ -528,6 +533,12 @@ class ByteDanceTextToVideoNode(IO.ComfyNode):
                     tooltip='Whether to add an "AI generated" watermark to the video.',
                     optional=True,
                 ),
+                IO.Boolean.Input(
+                    "generate_audio",
+                    default=False,
+                    tooltip="This parameter is ignored for any model except seedance-1-5-pro.",
+                    optional=True,
+                ),
             ],
             outputs=[
                 IO.Video.Output(),
@@ -552,7 +563,10 @@ class ByteDanceTextToVideoNode(IO.ComfyNode):
         seed: int,
         camera_fixed: bool,
         watermark: bool,
+        generate_audio: bool = False,
     ) -> IO.NodeOutput:
+        if model == "seedance-1-5-pro-251215" and duration < 4:
+            raise ValueError("Minimum supported duration for Seedance 1.5 Pro is 4 seconds.")
         validate_string(prompt, strip_whitespace=True, min_length=1)
         raise_if_text_params(prompt, ["resolution", "ratio", "duration", "seed", "camerafixed", "watermark"])
 
@@ -567,7 +581,11 @@ class ByteDanceTextToVideoNode(IO.ComfyNode):
         )
         return await process_video_task(
             cls,
-            payload=Text2VideoTaskCreationRequest(model=model, content=[TaskTextContent(text=prompt)]),
+            payload=Text2VideoTaskCreationRequest(
+                model=model,
+                content=[TaskTextContent(text=prompt)],
+                generate_audio=generate_audio if model == "seedance-1-5-pro-251215" else None,
+            ),
             estimated_duration=max(1, math.ceil(VIDEO_TASKS_EXECUTION_TIME[model][resolution] * (duration / 10.0))),
         )
 
@@ -584,7 +602,12 @@ class ByteDanceImageToVideoNode(IO.ComfyNode):
             inputs=[
                 IO.Combo.Input(
                     "model",
-                    options=["seedance-1-0-pro-250528", "seedance-1-0-lite-t2v-250428", "seedance-1-0-pro-fast-251015"],
+                    options=[
+                        "seedance-1-5-pro-251215",
+                        "seedance-1-0-pro-250528",
+                        "seedance-1-0-lite-i2v-250428",
+                        "seedance-1-0-pro-fast-251015",
+                    ],
                     default="seedance-1-0-pro-fast-251015",
                 ),
                 IO.String.Input(
@@ -639,6 +662,12 @@ class ByteDanceImageToVideoNode(IO.ComfyNode):
                     tooltip='Whether to add an "AI generated" watermark to the video.',
                     optional=True,
                 ),
+                IO.Boolean.Input(
+                    "generate_audio",
+                    default=False,
+                    tooltip="This parameter is ignored for any model except seedance-1-5-pro.",
+                    optional=True,
+                ),
             ],
             outputs=[
                 IO.Video.Output(),
@@ -664,7 +693,10 @@ class ByteDanceImageToVideoNode(IO.ComfyNode):
         seed: int,
         camera_fixed: bool,
         watermark: bool,
+        generate_audio: bool = False,
     ) -> IO.NodeOutput:
+        if model == "seedance-1-5-pro-251215" and duration < 4:
+            raise ValueError("Minimum supported duration for Seedance 1.5 Pro is 4 seconds.")
         validate_string(prompt, strip_whitespace=True, min_length=1)
         raise_if_text_params(prompt, ["resolution", "ratio", "duration", "seed", "camerafixed", "watermark"])
         validate_image_dimensions(image, min_width=300, min_height=300, max_width=6000, max_height=6000)
@@ -686,6 +718,7 @@ class ByteDanceImageToVideoNode(IO.ComfyNode):
             payload=Image2VideoTaskCreationRequest(
                 model=model,
                 content=[TaskTextContent(text=prompt), TaskImageContent(image_url=TaskImageContentUrl(url=image_url))],
+                generate_audio=generate_audio if model == "seedance-1-5-pro-251215" else None,
             ),
             estimated_duration=max(1, math.ceil(VIDEO_TASKS_EXECUTION_TIME[model][resolution] * (duration / 10.0))),
         )
@@ -703,7 +736,7 @@ class ByteDanceFirstLastFrameNode(IO.ComfyNode):
             inputs=[
                 IO.Combo.Input(
                     "model",
-                    options=["seedance-1-0-pro-250528", "seedance-1-0-lite-i2v-250428"],
+                    options=["seedance-1-5-pro-251215", "seedance-1-0-pro-250528", "seedance-1-0-lite-i2v-250428"],
                     default="seedance-1-0-lite-i2v-250428",
                 ),
                 IO.String.Input(
@@ -762,6 +795,12 @@ class ByteDanceFirstLastFrameNode(IO.ComfyNode):
                     tooltip='Whether to add an "AI generated" watermark to the video.',
                     optional=True,
                 ),
+                IO.Boolean.Input(
+                    "generate_audio",
+                    default=False,
+                    tooltip="This parameter is ignored for any model except seedance-1-5-pro.",
+                    optional=True,
+                ),
             ],
             outputs=[
                 IO.Video.Output(),
@@ -788,7 +827,10 @@ class ByteDanceFirstLastFrameNode(IO.ComfyNode):
         seed: int,
         camera_fixed: bool,
         watermark: bool,
+        generate_audio: bool = False,
     ) -> IO.NodeOutput:
+        if model == "seedance-1-5-pro-251215" and duration < 4:
+            raise ValueError("Minimum supported duration for Seedance 1.5 Pro is 4 seconds.")
         validate_string(prompt, strip_whitespace=True, min_length=1)
         raise_if_text_params(prompt, ["resolution", "ratio", "duration", "seed", "camerafixed", "watermark"])
         for i in (first_frame, last_frame):
@@ -821,6 +863,7 @@ class ByteDanceFirstLastFrameNode(IO.ComfyNode):
                     TaskImageContent(image_url=TaskImageContentUrl(url=str(download_urls[0])), role="first_frame"),
                     TaskImageContent(image_url=TaskImageContentUrl(url=str(download_urls[1])), role="last_frame"),
                 ],
+                generate_audio=generate_audio if model == "seedance-1-5-pro-251215" else None,
             ),
             estimated_duration=max(1, math.ceil(VIDEO_TASKS_EXECUTION_TIME[model][resolution] * (duration / 10.0))),
         )
@@ -896,7 +939,41 @@ class ByteDanceImageReferenceNode(IO.ComfyNode):
                 IO.Hidden.unique_id,
             ],
             is_api_node=True,
-            price_badge=PRICE_BADGE_VIDEO,
+            price_badge=IO.PriceBadge(
+                depends_on=IO.PriceBadgeDepends(widgets=["model", "duration", "resolution"]),
+                expr="""
+                (
+                  $priceByModel := {
+                    "seedance-1-0-pro": {
+                      "480p":[0.23,0.24],
+                      "720p":[0.51,0.56]
+                    },
+                    "seedance-1-0-lite": {
+                      "480p":[0.17,0.18],
+                      "720p":[0.37,0.41]
+                    }
+                  };
+                  $model := widgets.model;
+                  $modelKey :=
+                    $contains($model, "seedance-1-0-pro")  ? "seedance-1-0-pro" :
+                    "seedance-1-0-lite";
+                  $resolution := widgets.resolution;
+                  $resKey :=
+                    $contains($resolution, "720") ? "720p" :
+                    "480p";
+                  $modelPrices := $lookup($priceByModel, $modelKey);
+                  $baseRange := $lookup($modelPrices, $resKey);
+                  $min10s := $baseRange[0];
+                  $max10s := $baseRange[1];
+                  $scale := widgets.duration / 10;
+                  $minCost := $min10s * $scale;
+                  $maxCost := $max10s * $scale;
+                  ($minCost = $maxCost)
+                    ? {"type":"usd","usd": $minCost}
+                    : {"type":"range_usd","min_usd": $minCost, "max_usd": $maxCost}
+                )
+                """,
+            ),
         )
 
     @classmethod
@@ -967,10 +1044,15 @@ def raise_if_text_params(prompt: str, text_params: list[str]) -> None:
 
 
 PRICE_BADGE_VIDEO = IO.PriceBadge(
-    depends_on=IO.PriceBadgeDepends(widgets=["model", "duration", "resolution"]),
+    depends_on=IO.PriceBadgeDepends(widgets=["model", "duration", "resolution", "generate_audio"]),
     expr="""
     (
       $priceByModel := {
+        "seedance-1-5-pro": {
+          "480p":[0.12,0.12],
+          "720p":[0.26,0.26],
+          "1080p":[0.58,0.59]
+        },
         "seedance-1-0-pro": {
           "480p":[0.23,0.24],
           "720p":[0.51,0.56],
@@ -989,6 +1071,7 @@ PRICE_BADGE_VIDEO = IO.PriceBadge(
       };
       $model := widgets.model;
       $modelKey :=
+        $contains($model, "seedance-1-5-pro")      ? "seedance-1-5-pro" :
         $contains($model, "seedance-1-0-pro-fast") ? "seedance-1-0-pro-fast" :
         $contains($model, "seedance-1-0-pro")      ? "seedance-1-0-pro" :
         "seedance-1-0-lite";
@@ -1002,11 +1085,12 @@ PRICE_BADGE_VIDEO = IO.PriceBadge(
       $min10s := $baseRange[0];
       $max10s := $baseRange[1];
       $scale := widgets.duration / 10;
-      $minCost := $min10s * $scale;
-      $maxCost := $max10s * $scale;
+      $audioMultiplier := ($modelKey = "seedance-1-5-pro" and widgets.generate_audio) ? 2 : 1;
+      $minCost := $min10s * $scale * $audioMultiplier;
+      $maxCost := $max10s * $scale * $audioMultiplier;
       ($minCost = $maxCost)
-        ? {"type":"usd","usd": $minCost}
-        : {"type":"range_usd","min_usd": $minCost, "max_usd": $maxCost}
+        ? {"type":"usd","usd": $minCost, "format": { "approximate": true }}
+        : {"type":"range_usd","min_usd": $minCost, "max_usd": $maxCost, "format": { "approximate": true }}
     )
     """,
 )

main.py

@@ -10,6 +10,7 @@ from app.logger import setup_logger
 from app.assets.scanner import seed_assets
 import itertools
 import utils.extra_config
+import utils.landlock
 import logging
 import sys
 from comfy_execution.progress import get_progress_state
@@ -342,6 +343,13 @@ def start_comfyui(asyncio_loop=None):
         folder_paths.set_temp_directory(temp_dir)
     cleanup_temp()
 
+    if args.enable_landlock:
+        logging.info("Enabling Landlock")
+        landlock_ok = utils.landlock.enable_landlock(args, logging.getLogger("landlock"))
+        if not landlock_ok:
+            logging.critical("Requested Landlock sandbox but it could not be enabled. Exiting.")
+            sys.exit(1)
+
     if args.windows_standalone_build:
         try:
             import new_updater

requirements.txt

@@ -1,5 +1,5 @@
 comfyui-frontend-package==1.36.14
-comfyui-workflow-templates==0.8.4
+comfyui-workflow-templates==0.8.11
 comfyui-embedded-docs==0.4.0
 torch
 torchsde

server.py

@@ -686,7 +686,10 @@ class PromptServer():
 
         @routes.get("/object_info")
         async def get_object_info(request):
-            seed_assets(["models"])
+            try:
+                seed_assets(["models"])
+            except Exception as e:
+                logging.error(f"Failed to seed assets: {e}")
             with folder_paths.cache_helper:
                 out = {}
                 for x in nodes.NODE_CLASS_MAPPINGS:

utils/landlock.py

@@ -0,0 +1,332 @@
+import ctypes
+import errno
+import logging
+import os
+import sys
+import tempfile
+from dataclasses import dataclass
+
+# Landlock constants copied from linux/landlock.h
+PR_SET_NO_NEW_PRIVS = 38
+
+LANDLOCK_RULE_PATH_BENEATH = 1
+LANDLOCK_CREATE_RULESET_VERSION = 1
+
+LANDLOCK_ACCESS_FS_EXECUTE = 1 << 0
+LANDLOCK_ACCESS_FS_WRITE_FILE = 1 << 1
+LANDLOCK_ACCESS_FS_READ_FILE = 1 << 2
+LANDLOCK_ACCESS_FS_READ_DIR = 1 << 3
+LANDLOCK_ACCESS_FS_REMOVE_DIR = 1 << 4
+LANDLOCK_ACCESS_FS_REMOVE_FILE = 1 << 5
+LANDLOCK_ACCESS_FS_MAKE_CHAR = 1 << 6
+LANDLOCK_ACCESS_FS_MAKE_DIR = 1 << 7
+LANDLOCK_ACCESS_FS_MAKE_REG = 1 << 8
+LANDLOCK_ACCESS_FS_MAKE_SOCK = 1 << 9
+LANDLOCK_ACCESS_FS_MAKE_FIFO = 1 << 10
+LANDLOCK_ACCESS_FS_MAKE_BLOCK = 1 << 11
+LANDLOCK_ACCESS_FS_MAKE_SYM = 1 << 12
+LANDLOCK_ACCESS_FS_REFER = 1 << 13
+LANDLOCK_ACCESS_FS_TRUNCATE = 1 << 14
+LANDLOCK_ACCESS_FS_IOCTL_DEV = 1 << 15  # ABI v5+
+
+# Pre-computed access masks
+FS_READ_ACCESS = (
+    LANDLOCK_ACCESS_FS_READ_FILE
+    | LANDLOCK_ACCESS_FS_READ_DIR
+    | LANDLOCK_ACCESS_FS_EXECUTE
+)
+FS_WRITE_ACCESS = (
+    FS_READ_ACCESS
+    | LANDLOCK_ACCESS_FS_WRITE_FILE
+    | LANDLOCK_ACCESS_FS_MAKE_DIR
+    | LANDLOCK_ACCESS_FS_MAKE_REG
+    | LANDLOCK_ACCESS_FS_MAKE_SOCK
+    | LANDLOCK_ACCESS_FS_MAKE_FIFO
+    | LANDLOCK_ACCESS_FS_MAKE_BLOCK
+    | LANDLOCK_ACCESS_FS_MAKE_CHAR
+    | LANDLOCK_ACCESS_FS_MAKE_SYM
+    | LANDLOCK_ACCESS_FS_REMOVE_DIR
+    | LANDLOCK_ACCESS_FS_REMOVE_FILE
+)
+
+# Syscall numbers are ABI-stable across all 64-bit Linux architectures
+SYS_LANDLOCK_CREATE_RULESET = 444
+SYS_LANDLOCK_ADD_RULE = 445
+SYS_LANDLOCK_RESTRICT_SELF = 446
+
+
+class _RulesetAttr(ctypes.Structure):
+    _fields_ = [("handled_access_fs", ctypes.c_uint64)]
+
+
+class _PathBeneathAttr(ctypes.Structure):
+    _fields_ = [
+        ("allowed_access", ctypes.c_uint64),
+        ("parent_fd", ctypes.c_int32),
+        ("reserved", ctypes.c_uint32),
+    ]
+
+
+@dataclass(frozen=True)
+class LandlockRules:
+    read_paths: set[str]
+    write_paths: set[str]
+    ioctl_paths: set[str]
+
+
+def _normalize_paths(paths: set[str]) -> set[str]:
+    normalized = set()
+    for path in paths:
+        if not path:
+            continue
+        normalized.add(os.path.realpath(path))
+    return normalized
+
+
+class LandlockEnforcer:
+    def __init__(self, logger: logging.Logger | None = None):
+        self.log = logger or logging.getLogger(__name__)
+        self.libc = ctypes.CDLL(None, use_errno=True)
+        self.libc.syscall.restype = ctypes.c_long
+        self.libc.prctl.restype = ctypes.c_int
+
+    def _syscall(self, syscall_nr, *args) -> tuple[int | None, int]:
+        ctypes.set_errno(0)
+        res = self.libc.syscall(ctypes.c_long(syscall_nr), *args)
+        if res == -1:
+            return None, ctypes.get_errno()
+        return res, 0
+
+    def _abi_version(self) -> int:
+        res, err = self._syscall(
+            SYS_LANDLOCK_CREATE_RULESET,
+            ctypes.c_void_p(0),
+            ctypes.c_size_t(0),
+            ctypes.c_uint(LANDLOCK_CREATE_RULESET_VERSION),
+        )
+        if res is None:
+            if err in (errno.ENOSYS, errno.EOPNOTSUPP):
+                return 0
+            return -err
+        return res
+
+    def _create_ruleset(self, handled_access: int) -> tuple[int | None, int]:
+        ruleset = _RulesetAttr(ctypes.c_uint64(handled_access))
+        return self._syscall(
+            SYS_LANDLOCK_CREATE_RULESET,
+            ctypes.byref(ruleset),
+            ctypes.c_size_t(ctypes.sizeof(ruleset)),
+            ctypes.c_uint(0),
+        )
+
+    def _add_rule(self, ruleset_fd: int, path: str, access_mask: int, allow_ioctl: bool) -> bool:
+        if allow_ioctl:
+            access_mask |= LANDLOCK_ACCESS_FS_IOCTL_DEV
+
+        try:
+            dir_fd = os.open(path, os.O_PATH | os.O_CLOEXEC)
+        except OSError as exc:
+            self.log.warning("Landlock: skipping %s (%s)", path, exc)
+            return False
+
+        try:
+            rule = _PathBeneathAttr(
+                ctypes.c_uint64(access_mask), ctypes.c_int32(dir_fd), ctypes.c_uint32(0)
+            )
+            res, err = self._syscall(
+                SYS_LANDLOCK_ADD_RULE,
+                ctypes.c_int(ruleset_fd),
+                ctypes.c_int(LANDLOCK_RULE_PATH_BENEATH),
+                ctypes.byref(rule),
+                ctypes.c_uint(0),
+            )
+            if res is None:
+                self.log.warning("Landlock: failed to add %s (errno=%s)", path, err)
+                return False
+            return True
+        finally:
+            os.close(dir_fd)
+
+    def _restrict_self(self, ruleset_fd: int) -> bool:
+        if self.libc.prctl(PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0) != 0:
+            self.log.warning(
+                "Landlock: prctl(PR_SET_NO_NEW_PRIVS) failed (errno=%s)",
+                ctypes.get_errno(),
+            )
+            return False
+
+        res, err = self._syscall(SYS_LANDLOCK_RESTRICT_SELF, ctypes.c_int(ruleset_fd), ctypes.c_uint(0))
+        if res is None:
+            self.log.warning("Landlock: restrict_self failed (errno=%s)", err)
+            return False
+        return True
+
+    def apply(self, rules: LandlockRules) -> bool:
+        if not sys.platform.startswith("linux"):
+            self.log.info("Landlock: not a Linux platform, skipping.")
+            return False
+
+        abi_version = self._abi_version()
+        if abi_version <= 0:
+            self.log.info("Landlock: not available on this kernel (abi=%s).", abi_version)
+            return False
+
+        read_access = FS_READ_ACCESS
+        handled_write_access = FS_WRITE_ACCESS
+        allowed_write_access = (
+            read_access
+            | LANDLOCK_ACCESS_FS_WRITE_FILE
+            | LANDLOCK_ACCESS_FS_MAKE_DIR
+            | LANDLOCK_ACCESS_FS_MAKE_REG
+            | LANDLOCK_ACCESS_FS_REMOVE_DIR
+            | LANDLOCK_ACCESS_FS_REMOVE_FILE
+        )  # leave other handled rights (symlinks, device nodes, sockets) denied
+        if abi_version >= 2:
+            handled_write_access |= LANDLOCK_ACCESS_FS_TRUNCATE
+            allowed_write_access |= LANDLOCK_ACCESS_FS_TRUNCATE
+        if abi_version >= 3:
+            handled_write_access |= LANDLOCK_ACCESS_FS_REFER
+            allowed_write_access |= LANDLOCK_ACCESS_FS_REFER
+
+        handled_access = handled_write_access | read_access
+        ioctl_supported = abi_version >= 5
+        if ioctl_supported:
+            handled_access |= LANDLOCK_ACCESS_FS_IOCTL_DEV
+
+        write_paths = _normalize_paths(rules.write_paths)
+        read_paths = _normalize_paths(rules.read_paths) - write_paths
+        # In theory these could require write or read access. Though in practice it's just /dev.
+        ioctl_paths = _normalize_paths(rules.ioctl_paths)
+
+        if ioctl_paths and not ioctl_supported:
+            self.log.info(
+                "Landlock: ioctl access requested but ABI %s has no support; continuing without ioctl.",
+                abi_version,
+            )
+
+        ruleset_fd = None
+        ruleset_fd, err = self._create_ruleset(handled_access)
+        if ruleset_fd is None:
+            self.log.warning("Landlock: failed to create ruleset (errno=%s)", err)
+            return False
+
+        try:
+            for path in write_paths:
+                if path != os.path.sep:
+                    try:
+                        os.makedirs(path, exist_ok=True)
+                    except Exception as exc:
+                        self.log.warning("Landlock: unable to prepare %s (%s)", path, exc)
+                        return False
+                if not self._add_rule(
+                    ruleset_fd, path, allowed_write_access, ioctl_supported and path in ioctl_paths
+                ):
+                    return False
+
+            for path in read_paths:
+                self._add_rule(ruleset_fd, path, read_access, ioctl_supported and path in ioctl_paths)
+
+            if not self._restrict_self(ruleset_fd):
+                return False
+        finally:
+            if ruleset_fd is not None:
+                os.close(ruleset_fd)
+
+        if write_paths:
+            self.log.info(
+                "Landlock enabled (ABI %s). Writable roots: %s",
+                abi_version,
+                ", ".join(sorted(write_paths)),
+            )
+        else:
+            self.log.info("Landlock enabled (ABI %s). No writable roots configured.", abi_version)
+        return True
+
+
+_landlock_applied = False
+
+
+def build_default_rules(args) -> LandlockRules:
+    import folder_paths
+    from urllib.parse import urlparse
+
+    write_paths: set[str] = {
+        folder_paths.get_output_directory(),
+        folder_paths.get_input_directory(),
+        folder_paths.get_temp_directory(),
+        folder_paths.get_user_directory(),
+    }
+
+    ioctl_paths: set[str] = set()
+
+    # Torch and some backends use system temp and /dev/shm
+    write_paths.add(tempfile.gettempdir())
+
+    if args.temp_directory:
+        write_paths.add(os.path.join(os.path.abspath(args.temp_directory), "temp"))
+
+    db_url = getattr(args, "database_url", None)
+    if db_url and db_url.startswith("sqlite"):
+        parsed = urlparse(db_url)
+        if parsed.scheme == "sqlite" and parsed.path:
+            write_paths.add(os.path.abspath(os.path.dirname(parsed.path)))
+
+    for path in args.landlock_allow_writable or []:
+        if path:
+            write_paths.add(path)
+
+    # Build read paths - only what's actually needed
+    read_paths: set[str] = set()
+
+    # ComfyUI codebase
+    read_paths.add(folder_paths.base_path)
+
+    # All configured model directories (includes extra_model_paths.yaml)
+    for folder_name in folder_paths.folder_names_and_paths:
+        for path in folder_paths.folder_names_and_paths[folder_name][0]:
+            read_paths.add(path)
+
+    # Python installation and site-packages
+    read_paths.add(sys.prefix)
+    if sys.base_prefix != sys.prefix:
+        read_paths.add(sys.base_prefix)
+    for path in sys.path:
+        if path and os.path.isdir(path):
+            read_paths.add(path)
+
+    # System libraries (required for shared libs, CUDA, etc.)
+    for system_path in ["/usr", "/lib", "/lib64", "/opt", "/etc", "/proc", "/sys"]:
+        if os.path.exists(system_path):
+            read_paths.add(system_path)
+
+    # NixOS: /nix/store contains the entire system
+    if os.path.exists("/nix"):
+        read_paths.add("/nix")
+
+    # /dev needs write + ioctl for CUDA/GPU access
+    write_paths.add("/dev")
+    ioctl_paths.add("/dev")
+
+    # User-specified additional read paths
+    for path in getattr(args, "landlock_allow_readable", None) or []:
+        if path:
+            read_paths.add(path)
+
+    return LandlockRules(read_paths=_normalize_paths(read_paths), write_paths=_normalize_paths(write_paths), ioctl_paths=_normalize_paths(ioctl_paths))
+
+
+def enable_landlock(args, logger: logging.Logger | None = None) -> bool:
+    global _landlock_applied
+
+    if _landlock_applied:
+        return True
+    if not getattr(args, "enable_landlock", False):
+        return False
+
+    enforcer = LandlockEnforcer(logger)
+    try:
+        _landlock_applied = enforcer.apply(build_default_rules(args))
+    except Exception:
+        enforcer.log.exception("Landlock: unexpected failure while applying ruleset.")
+        _landlock_applied = False
+    return _landlock_applied