mirror of
https://github.com/Comfy-Org/ComfyUI-Manager.git
synced 2026-07-26 10:07:38 +08:00
feat(security): dedicated install flags decouple git_url/pip from security_level
Install via git URL and pip install are no longer gated by
security_level. Each surface gets a dedicated config.ini flag —
allow_git_url_install / allow_pip_install (both default false, secure
by default) — that fully REPLACES the security-level term for these two
features. The network-position invariant is retained: a non-local
listener stays denied regardless of the flags unless
network_mode = personal_cloud.
- New pure predicate is_dedicated_install_allowed() in
common/manager_security (no config access; callers resolve config)
- Legacy endpoints /v2/customnode/install/git_url and .../pip switch
from is_allowed_security_level('high+') to the flag gate; batch
installs of unknown git URLs likewise (middle+ entry gate unchanged,
unknown-pip 'block' stays unconditional; response shapes preserved)
- Config readers/writers (glob + legacy) parse and persist the flags;
denial logs and frontend 403 messages name the responsible flag and
note the non-local-listener requirement (network_mode=personal_cloud)
- No auto-seed from security_level — users previously on weak/normal-
must opt in explicitly (see CHANGELOG migration notes; README
documents the new contract)
- Update the pre-existing permissive E2E harness
(start_comfyui_permissive.sh + test_e2e_legacy_real_ops.py) to the
new contract: it now also sets allow_git_url_install /
allow_pip_install = true, since security_level = normal- alone no
longer opens the git_url/pip endpoints
Tests: predicate truth table proving security_level independence in
both directions, dual-reader config contract, security-level-matrix
freeze guards, legacy gate regression guards (121 unit), plus 22
real-server E2E tests incl. URL-form pip install with self-clean.
This commit is contained in:
Dr.Lt.Data
@@ -0,0 +1,177 @@
|
||||
"""Shared test-support for the goal265 dedicated-install-flag test modules.
|
||||
|
||||
NOT a test module (no ``test_`` prefix — pytest does not collect it).
|
||||
|
||||
Provides minimal runtime stubs so ``comfyui_manager`` package modules can be
|
||||
imported in a plain dev venv (outside a real ComfyUI runtime):
|
||||
|
||||
- ``comfy.cli_args.args`` — consumed by ``comfyui_manager/__init__.py`` at
|
||||
package import time (``from comfy.cli_args import args``). The stub pins
|
||||
``listen='127.0.0.1'`` (loopback) which matches the default precondition of
|
||||
every scenario row (goal265-scenarios.md §0 preamble).
|
||||
- ``folder_paths`` — consumed by ``comfyui_manager/common/context.py`` at
|
||||
import time to derive the manager user directory. The stub redirects it to
|
||||
a throwaway temp dir so importing the package NEVER creates directories
|
||||
inside the repository checkout.
|
||||
|
||||
SUITE-ORDER INDEPENDENCE (goal265 FU, task #293): every public ``import_*``
|
||||
entry point is self-sufficient — it repairs whatever ``sys.modules`` state
|
||||
earlier test modules left behind, instead of assuming a pristine interpreter:
|
||||
|
||||
- Other test modules legitimately install FAKE ``comfyui_manager`` lineage
|
||||
entries at module level (e.g. tests/test_unified_dep_resolver.py:63-73
|
||||
registers a plain ``types.ModuleType("comfyui_manager")`` — NOT a package,
|
||||
no ``__path__`` — to host a file-loaded module under a dotted name). Those
|
||||
modules are imported at pytest COLLECTION time, so under full-suite
|
||||
ordering the fake is already in ``sys.modules`` before any fixture here
|
||||
runs, and ``import comfyui_manager.glob.manager_core`` dies with
|
||||
``ModuleNotFoundError: 'comfyui_manager' is not a package``.
|
||||
- ``_purge_fake_comfyui_manager()`` therefore validates the cached lineage
|
||||
against the REAL package directory on disk and drops fake/broken entries
|
||||
before importing. Dropping ``sys.modules`` entries is safe for the other
|
||||
test modules: they hold direct object references to what they loaded;
|
||||
they do not re-resolve through ``sys.modules``.
|
||||
- ``ensure_comfy_stubs()`` likewise repairs half-formed ``comfy`` /
|
||||
``folder_paths`` entries (present but missing the attributes we need)
|
||||
rather than only handling the absent case.
|
||||
|
||||
Used by:
|
||||
- tests/test_install_flags_config.py (dual-reader config tests)
|
||||
- tests/test_install_flags_guards.py (out-of-scope guard tests)
|
||||
|
||||
Spec SoT: ~/.claude/pair-cowork/scratch/gm-team/goal265-spec.md §4
|
||||
(test surface contract — binding for Step-4 test authoring).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
|
||||
#: repo root (parent of tests/)
|
||||
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
#: the real on-disk package directory the cached lineage must resolve to
|
||||
_REAL_PKG_DIR = os.path.join(REPO_ROOT, "comfyui_manager")
|
||||
|
||||
|
||||
def ensure_comfy_stubs() -> None:
|
||||
"""Install or REPAIR minimal ``comfy``/``folder_paths`` stubs (idempotent).
|
||||
|
||||
Handles three prior states per module:
|
||||
- absent -> install a fresh stub
|
||||
- present but inadequate -> patch the missing attributes in place
|
||||
(another test's stub, or a partially-initialized real module)
|
||||
- present and adequate -> leave untouched (e.g. real ComfyUI runtime)
|
||||
"""
|
||||
# --- comfy.cli_args.args ------------------------------------------------
|
||||
comfy = sys.modules.get("comfy")
|
||||
if comfy is None:
|
||||
comfy = types.ModuleType("comfy")
|
||||
sys.modules["comfy"] = comfy
|
||||
|
||||
cli_args = sys.modules.get("comfy.cli_args")
|
||||
if cli_args is None:
|
||||
cli_args = getattr(comfy, "cli_args", None)
|
||||
if not isinstance(cli_args, types.ModuleType):
|
||||
cli_args = types.ModuleType("comfy.cli_args")
|
||||
sys.modules["comfy.cli_args"] = cli_args
|
||||
|
||||
args = getattr(cli_args, "args", None)
|
||||
if args is None or not hasattr(args, "listen"):
|
||||
setattr(cli_args, "args", types.SimpleNamespace(
|
||||
listen="127.0.0.1",
|
||||
enable_manager=True,
|
||||
enable_manager_legacy_ui=False,
|
||||
))
|
||||
# re-link parent attr idempotently (a bare ``comfy`` stub from another
|
||||
# test may lack the submodule attribute even when both entries exist)
|
||||
setattr(comfy, "cli_args", cli_args)
|
||||
|
||||
# --- folder_paths -------------------------------------------------------
|
||||
fp = sys.modules.get("folder_paths")
|
||||
if fp is None:
|
||||
fp = types.ModuleType("folder_paths")
|
||||
sys.modules["folder_paths"] = fp
|
||||
if not hasattr(fp, "get_system_user_directory"):
|
||||
base = tempfile.mkdtemp(prefix="cm-flags-test-user-")
|
||||
setattr(fp, "get_system_user_directory",
|
||||
lambda name: os.path.join(base, name))
|
||||
|
||||
|
||||
def _module_is_real(mod: types.ModuleType) -> bool:
|
||||
"""True when a cached ``comfyui_manager*`` module resolves to the real
|
||||
on-disk package tree (by ``__path__`` for packages, ``__file__`` for
|
||||
leaf modules)."""
|
||||
real_root = os.path.abspath(_REAL_PKG_DIR)
|
||||
paths = getattr(mod, "__path__", None)
|
||||
if paths is not None:
|
||||
return any(os.path.abspath(p).startswith(real_root) for p in paths)
|
||||
file = getattr(mod, "__file__", None)
|
||||
if file is not None:
|
||||
return os.path.abspath(file).startswith(real_root)
|
||||
# neither __path__ nor __file__: a bare ModuleType fake
|
||||
return False
|
||||
|
||||
|
||||
def _purge_fake_comfyui_manager() -> None:
|
||||
"""Drop fake/broken ``comfyui_manager`` lineage entries from
|
||||
``sys.modules`` so a subsequent real-package import succeeds.
|
||||
|
||||
Uniform rule — an entry is dropped iff it does NOT resolve to the real
|
||||
on-disk package tree:
|
||||
|
||||
- A fake top-level entry (bare ``ModuleType`` without ``__path__`` —
|
||||
'not a package') is dropped so the real package can import.
|
||||
- Fake SUBentries injected under dotted names (e.g.
|
||||
``comfyui_manager.common.manager_util`` replaced by a stub) are
|
||||
dropped; the import machinery re-imports the real module on next
|
||||
access.
|
||||
- REAL-file modules registered under dotted names by other test
|
||||
modules (e.g. ``comfyui_manager.common.unified_dep_resolver`` loaded
|
||||
via spec_from_file_location by tests/test_unified_dep_resolver.py)
|
||||
are KEPT: those tests later resolve the SAME cached entry through
|
||||
``mock.patch("comfyui_manager.common.unified_dep_resolver.X")``;
|
||||
evicting it would make mock.patch import a fresh second instance and
|
||||
silently patch the wrong module object.
|
||||
"""
|
||||
lineage = [n for n in list(sys.modules)
|
||||
if n == "comfyui_manager" or n.startswith("comfyui_manager.")]
|
||||
for name in lineage:
|
||||
mod = sys.modules.get(name)
|
||||
if mod is None or not _module_is_real(mod):
|
||||
del sys.modules[name]
|
||||
|
||||
|
||||
def _import_real(dotted: str):
|
||||
"""ensure stubs -> purge fakes -> import; shared by all entry points."""
|
||||
ensure_comfy_stubs()
|
||||
_purge_fake_comfyui_manager()
|
||||
return importlib.import_module(dotted)
|
||||
|
||||
|
||||
def import_reader(reader: str):
|
||||
"""Import and return ``comfyui_manager.<reader>.manager_core``.
|
||||
|
||||
``reader`` is ``"glob"`` or ``"legacy"`` — the two independent config
|
||||
reader implementations sharing one config.ini (dual-reader rule,
|
||||
goal265-mm.md §1.1 / goal265-spec.md §3).
|
||||
"""
|
||||
assert reader in ("glob", "legacy"), reader
|
||||
return _import_real(f"comfyui_manager.{reader}.manager_core")
|
||||
|
||||
|
||||
def import_context():
|
||||
"""Import and return ``comfyui_manager.common.context`` (holds
|
||||
``manager_config_path``, consumed by both readers at call time)."""
|
||||
return _import_real("comfyui_manager.common.context")
|
||||
|
||||
|
||||
def import_glob_security_utils():
|
||||
"""Import and return ``comfyui_manager.glob.utils.security_utils``
|
||||
(the glob copy of the security_level gate matrix — guard rows
|
||||
SC-25C / SC-28)."""
|
||||
return _import_real("comfyui_manager.glob.utils.security_utils")
|
||||
@@ -0,0 +1,267 @@
|
||||
"""[goal265 step4 — RED] Predicate truth table for ``is_dedicated_install_allowed``.
|
||||
|
||||
TARGET CONTRACT (NOT YET IMPLEMENTED — goal265-spec.md §1.2, LOCKED):
|
||||
|
||||
comfyui_manager/common/manager_security.py::is_dedicated_install_allowed(
|
||||
flag_value: bool, listen_address: str, network_mode: str) -> bool
|
||||
|
||||
P-direct: allowed iff bool(flag_value)
|
||||
AND (is_loopback(listen_address)
|
||||
OR network_mode.lower() == 'personal_cloud')
|
||||
|
||||
These tests are EXPECTED TO FAIL/ERROR against current code (the predicate
|
||||
does not exist yet) — RED confirmation is goal265 Step 5. Do NOT weaken them
|
||||
to pass against today's code.
|
||||
|
||||
SC rows covered (goal265-scenarios.md — predicate-level arm):
|
||||
SC-01, SC-02, SC-03, SC-05, SC-06, SC-07 ([D1] allow_git_url_install)
|
||||
SC-11, SC-12, SC-13, SC-14, SC-15 ([D2] allow_pip_install)
|
||||
SC-16, SC-17 (flag independence, [D1][D2])
|
||||
|
||||
security_level is ABSENT from the predicate signature — its irrelevance is
|
||||
proven BY CONSTRUCTION (spec §4 row 1): rows whose preconditions differ only
|
||||
in security_level (SC-01 vs SC-02; SC-06/SC-14 parametrizations) map onto the
|
||||
IDENTICAL predicate call, so no security_level value can change the outcome.
|
||||
|
||||
Fixtures: none — pure function (spec §4 binding patching constraint).
|
||||
The module under test is loaded directly by file path so the test does not
|
||||
import the ``comfyui_manager`` package (whose __init__ needs the ComfyUI
|
||||
runtime); ``manager_security.py`` itself is dependency-light by design
|
||||
(spec §1.2: MUST stay config-import-free).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import inspect
|
||||
import pathlib
|
||||
|
||||
import pytest
|
||||
|
||||
_MANAGER_SECURITY_PATH = (
|
||||
pathlib.Path(__file__).resolve().parents[2]
|
||||
/ "comfyui_manager" / "common" / "manager_security.py"
|
||||
)
|
||||
|
||||
# Address / network-mode vocabulary (scenarios §0 preamble):
|
||||
LOOPBACK = "127.0.0.1"
|
||||
LOOPBACK_V6 = "::1"
|
||||
PUBLIC_ADDR = "203.0.113.5" # TEST-NET-3 — non-loopback ("listen=public")
|
||||
NM_PUBLIC = "public"
|
||||
NM_PERSONAL_CLOUD = "personal_cloud"
|
||||
|
||||
# security_level context values for irrelevance-by-construction params.
|
||||
# The predicate takes NO security_level argument; these ids document which
|
||||
# scenario-row context each identical call stands for.
|
||||
SECURITY_LEVELS = ("strong", "normal", "normal-", "weak")
|
||||
|
||||
|
||||
def _load_manager_security():
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"_manager_security_under_test", _MANAGER_SECURITY_PATH
|
||||
)
|
||||
assert spec is not None and spec.loader is not None, _MANAGER_SECURITY_PATH
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def predicate():
|
||||
"""The predicate under test. FAILS (RED) while the function is absent."""
|
||||
mod = _load_manager_security()
|
||||
assert hasattr(mod, "is_dedicated_install_allowed"), (
|
||||
"RED: comfyui_manager/common/manager_security.py does not define "
|
||||
"is_dedicated_install_allowed yet (goal265-spec.md §1.2). "
|
||||
"This is the expected state before Step 6 (Develop)."
|
||||
)
|
||||
return mod.is_dedicated_install_allowed
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Structural: security_level is not even an input (irrelevance by construction)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_signature_has_no_security_level_parameter(predicate):
|
||||
"""[SC-01..SC-17 foundation] Spec §1.2 locks the signature to exactly
|
||||
(flag_value, listen_address, network_mode) — security_level CANNOT
|
||||
influence the decision because it is not an input."""
|
||||
params = list(inspect.signature(predicate).parameters)
|
||||
assert params == ["flag_value", "listen_address", "network_mode"], (
|
||||
f"Locked signature drifted: {params!r} "
|
||||
"(goal265-spec.md §1.2 — security_level must NOT appear)"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# [D1] allow_git_url_install — happy paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc01_flag_true_loopback_allows_under_strong_security_level(predicate):
|
||||
"""SC-01: git=true, sl=strong, listen=loopback, nm=public -> allowed.
|
||||
sl=strong is context only — the call has no security_level input."""
|
||||
assert predicate(True, LOOPBACK, NM_PUBLIC) is True
|
||||
|
||||
|
||||
def test_sc02_flag_true_loopback_allows_under_default_security_level(predicate):
|
||||
"""SC-02: git=true, sl=normal (default), listen=loopback, nm=public ->
|
||||
allowed. Identical call to SC-01: proves sl irrelevance by construction."""
|
||||
assert predicate(True, LOOPBACK, NM_PUBLIC) is True
|
||||
|
||||
|
||||
def test_sc03_flag_true_public_listener_personal_cloud_allows(predicate):
|
||||
"""SC-03: git=true, sl=normal, listen=public, nm=personal_cloud ->
|
||||
allowed (personal_cloud arm of the network-position invariant)."""
|
||||
assert predicate(True, PUBLIC_ADDR, NM_PERSONAL_CLOUD) is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# [D1] allow_git_url_install — failure paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc05_flag_false_denies_even_at_weak_security_level(predicate):
|
||||
"""SC-05: git=false, sl=weak, listen=loopback -> denied. Deny-direction
|
||||
proof of security_level irrelevance (today sl=weak would ALLOW)."""
|
||||
assert predicate(False, LOOPBACK, NM_PUBLIC) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("security_level_context", SECURITY_LEVELS[:3],
|
||||
ids=[f"sl={s}" for s in SECURITY_LEVELS[:3]])
|
||||
def test_sc06_flag_false_denies_for_every_security_level(
|
||||
predicate, security_level_context):
|
||||
"""SC-06: git=false, sl in {strong, normal, normal-}, listen=loopback ->
|
||||
denied for EVERY sl value. The parametrize axis documents that all three
|
||||
scenario contexts collapse onto the same flag-only call — the flag is
|
||||
the sole decider."""
|
||||
assert predicate(False, LOOPBACK, NM_PUBLIC) is False
|
||||
|
||||
|
||||
def test_sc07_flag_true_cannot_open_public_listener(predicate):
|
||||
"""SC-07: git=true, sl=weak, listen=public, nm=public -> denied.
|
||||
Network-position invariant retained (spec §1.1 invariant 2): the flag
|
||||
must never widen exposure beyond what is possible today."""
|
||||
assert predicate(True, PUBLIC_ADDR, NM_PUBLIC) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# [D2] allow_pip_install — happy paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc11_pip_flag_true_loopback_allows_under_strong(predicate):
|
||||
"""SC-11: pip=true, sl=strong, listen=loopback, nm=public -> allowed."""
|
||||
assert predicate(True, LOOPBACK, NM_PUBLIC) is True
|
||||
|
||||
|
||||
def test_sc12_pip_flag_true_public_listener_personal_cloud_allows(predicate):
|
||||
"""SC-12: pip=true, sl=normal, listen=public, nm=personal_cloud ->
|
||||
allowed."""
|
||||
assert predicate(True, PUBLIC_ADDR, NM_PERSONAL_CLOUD) is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# [D2] allow_pip_install — failure paths
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc13_pip_flag_false_denies_even_at_weak(predicate):
|
||||
"""SC-13: pip=false, sl=weak, listen=loopback -> denied."""
|
||||
assert predicate(False, LOOPBACK, NM_PUBLIC) is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("security_level_context", SECURITY_LEVELS[:3],
|
||||
ids=[f"sl={s}" for s in SECURITY_LEVELS[:3]])
|
||||
def test_sc14_pip_flag_false_denies_for_every_security_level(
|
||||
predicate, security_level_context):
|
||||
"""SC-14: pip=false, sl in {strong, normal, normal-}, listen=loopback ->
|
||||
denied each. Same irrelevance-by-construction pattern as SC-06."""
|
||||
assert predicate(False, LOOPBACK, NM_PUBLIC) is False
|
||||
|
||||
|
||||
def test_sc15_pip_flag_true_cannot_open_public_listener(predicate):
|
||||
"""SC-15: pip=true, sl=weak, listen=public, nm=public -> denied
|
||||
(network invariant)."""
|
||||
assert predicate(True, PUBLIC_ADDR, NM_PUBLIC) is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Flag independence (cross-link [D1]+[D2])
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc16_git_false_pip_true_independent_outcomes(predicate):
|
||||
"""SC-16: git=false, pip=true, sl=normal, listen=loopback -> git denied,
|
||||
pip allowed. At predicate level each surface passes ONLY its own flag —
|
||||
same network inputs, opposite outcomes."""
|
||||
git_allowed = predicate(False, LOOPBACK, NM_PUBLIC)
|
||||
pip_allowed = predicate(True, LOOPBACK, NM_PUBLIC)
|
||||
assert git_allowed is False
|
||||
assert pip_allowed is True
|
||||
|
||||
|
||||
def test_sc17_git_true_pip_false_independent_outcomes(predicate):
|
||||
"""SC-17: git=true, pip=false, sl=normal, listen=loopback -> git allowed,
|
||||
pip denied (mirror of SC-16)."""
|
||||
git_allowed = predicate(True, LOOPBACK, NM_PUBLIC)
|
||||
pip_allowed = predicate(False, LOOPBACK, NM_PUBLIC)
|
||||
assert git_allowed is True
|
||||
assert pip_allowed is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Supplementary: exhaustive flag x loopback x personal_cloud truth table
|
||||
# (spec §4 row 1 — "flag x loopback x personal_cloud" full table)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("flag", "listen", "network_mode", "expected"),
|
||||
[
|
||||
(True, LOOPBACK, NM_PUBLIC, True), # SC-01/02/11
|
||||
(True, LOOPBACK, NM_PERSONAL_CLOUD, True), # both arms true
|
||||
(True, PUBLIC_ADDR, NM_PERSONAL_CLOUD, True), # SC-03/12
|
||||
(True, PUBLIC_ADDR, NM_PUBLIC, False), # SC-07/15
|
||||
(False, LOOPBACK, NM_PUBLIC, False), # SC-05/06/13/14
|
||||
(False, LOOPBACK, NM_PERSONAL_CLOUD, False),
|
||||
(False, PUBLIC_ADDR, NM_PERSONAL_CLOUD, False),
|
||||
(False, PUBLIC_ADDR, NM_PUBLIC, False),
|
||||
],
|
||||
ids=[
|
||||
"flag+loopback+nm_public",
|
||||
"flag+loopback+personal_cloud",
|
||||
"flag+public+personal_cloud",
|
||||
"flag+public+nm_public",
|
||||
"noflag+loopback+nm_public",
|
||||
"noflag+loopback+personal_cloud",
|
||||
"noflag+public+personal_cloud",
|
||||
"noflag+public+nm_public",
|
||||
],
|
||||
)
|
||||
def test_truth_table_flag_x_loopback_x_personal_cloud(
|
||||
predicate, flag, listen, network_mode, expected):
|
||||
"""Full 2x2x2 truth table for P-direct (spec §1.1): allowed iff flag AND
|
||||
(loopback OR personal_cloud). Consolidates SC-01/02/03/05/06/07 ([D1])
|
||||
and SC-11/12/13/14/15 ([D2]) plus the two combinations no single row
|
||||
pins (flag+loopback+personal_cloud, noflag+public+personal_cloud)."""
|
||||
assert predicate(flag, listen, network_mode) is expected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Supplementary edge handling locked by the spec text
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_ipv6_loopback_counts_as_loopback(predicate):
|
||||
"""SC-01/SC-02 edge (loopback arm): `is_loopback` is ipaddress-based
|
||||
(manager_security.py) — ::1 must be treated as loopback too."""
|
||||
assert predicate(True, LOOPBACK_V6, NM_PUBLIC) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("network_mode", ["Personal_Cloud", "PERSONAL_CLOUD"],
|
||||
ids=["mixed-case", "upper-case"])
|
||||
def test_network_mode_personal_cloud_is_case_insensitive(predicate, network_mode):
|
||||
"""SC-03/SC-12 edge (personal_cloud arm): spec §1.2 predicate body —
|
||||
network_mode.lower() == 'personal_cloud' (case-insensitive)."""
|
||||
assert predicate(True, PUBLIC_ADDR, network_mode) is True
|
||||
|
||||
|
||||
def test_unparseable_listen_address_is_not_loopback(predicate):
|
||||
"""SC-07/SC-15 edge (network invariant): is_loopback returns False for
|
||||
unparseable addresses (ValueError arm) — the predicate must fail CLOSED
|
||||
on a malformed listen address."""
|
||||
assert predicate(True, "not-an-ip-address", NM_PUBLIC) is False
|
||||
@@ -10,15 +10,18 @@
|
||||
# script does NOT restore on its own, so fixture teardown MUST cleanup.
|
||||
#
|
||||
# Why permissive mode is needed:
|
||||
# Three endpoints check is_allowed_security_level('high+')
|
||||
# (security_utils.py:20-26): at is_local_mode=True (127.0.0.1 listen)
|
||||
# the gate requires security_level ∈ {weak, normal-}. Default
|
||||
# `security_level = normal` fails, so the POST returns 403.
|
||||
# - wi-014 POST /v2/comfyui_manager/comfyui_switch_version
|
||||
# - wi-037 POST /v2/customnode/install/git_url
|
||||
# - wi-038 POST /v2/customnode/install/pip
|
||||
# Setting security_level = normal- allows real E2E execution of these
|
||||
# endpoints with fixed, trusted inputs (never test-input-derived URLs).
|
||||
# - wi-014 POST /v2/comfyui_manager/comfyui_switch_version checks
|
||||
# is_allowed_security_level('high+'): at is_local_mode=True
|
||||
# (127.0.0.1 listen) the gate requires security_level ∈ {weak,
|
||||
# normal-}. Default `security_level = normal` fails (403).
|
||||
# - wi-037 POST /v2/customnode/install/git_url and
|
||||
# wi-038 POST /v2/customnode/install/pip are gated by the dedicated
|
||||
# config.ini flags `allow_git_url_install` / `allow_pip_install`
|
||||
# (goal265 / #2962) — independent of security_level, default false
|
||||
# (403).
|
||||
# Patching security_level = normal- plus both flags = true allows real
|
||||
# E2E execution of these endpoints with fixed, trusted inputs (never
|
||||
# test-input-derived URLs).
|
||||
#
|
||||
# SECURITY NOTE:
|
||||
# The endpoints are gated at high+ because they execute arbitrary remote
|
||||
@@ -39,7 +42,8 @@
|
||||
# Exit: 0=ready, 1=timeout/failure
|
||||
#
|
||||
# Side effect: $E2E_ROOT/comfyui/user/__manager/config.ini gets
|
||||
# `security_level = normal-`. The original value is preserved at
|
||||
# `security_level = normal-`, `allow_git_url_install = true` and
|
||||
# `allow_pip_install = true`. The original config is preserved at
|
||||
# config.ini.before-permissive for the fixture to restore on teardown.
|
||||
|
||||
set -euo pipefail
|
||||
@@ -68,4 +72,15 @@ else
|
||||
fi
|
||||
echo "[start_comfyui_permissive] Patched security_level = normal- in $CONFIG"
|
||||
|
||||
# Patch the dedicated install flags (goal265 / #2962): install/git_url and
|
||||
# install/pip are gated by these flags instead of security_level (idempotent).
|
||||
for flag in allow_git_url_install allow_pip_install; do
|
||||
if grep -qE "^${flag}\s*=" "$CONFIG"; then
|
||||
sed -i -E "s/^${flag}\s*=.*/${flag} = true/" "$CONFIG"
|
||||
else
|
||||
sed -i -E "/^\[default\]/a ${flag} = true" "$CONFIG"
|
||||
fi
|
||||
done
|
||||
echo "[start_comfyui_permissive] Patched allow_git_url_install/allow_pip_install = true in $CONFIG"
|
||||
|
||||
exec env ENABLE_LEGACY_UI=1 bash "$SCRIPT_DIR/start_comfyui.sh" "$@"
|
||||
|
||||
@@ -5,10 +5,13 @@ operation via HTTP:
|
||||
Default-security fixture (middle+ / no-gate endpoints):
|
||||
- wi-020 POST /v2/manager/queue/install_model (tiny TAEF1 model, <5MB)
|
||||
- wi-024 POST /v2/manager/queue/update_comfyui (safe via env var)
|
||||
Permissive-security fixture (high+ endpoints — normal- harness):
|
||||
- wi-014 POST /v2/comfyui_manager/comfyui_switch_version (no-op self-switch)
|
||||
- wi-037 POST /v2/customnode/install/git_url (nodepack-test1-do-not-install)
|
||||
- wi-038 POST /v2/customnode/install/pip (text-unidecode)
|
||||
Permissive-security fixture (normal- + dedicated install flags):
|
||||
- wi-014 POST /v2/comfyui_manager/comfyui_switch_version (no-op self-switch;
|
||||
high+ gate, opened by security_level = normal-)
|
||||
- wi-037 POST /v2/customnode/install/git_url (nodepack-test1-do-not-install;
|
||||
gated by allow_git_url_install — goal265 / #2962)
|
||||
- wi-038 POST /v2/customnode/install/pip (text-unidecode;
|
||||
gated by allow_pip_install — goal265 / #2962)
|
||||
Pre-seeded broken-pack fixture (no-gate endpoint, needs scan-time state):
|
||||
- wi-015 POST /v2/customnode/import_fail_info (pre-seeded broken pack)
|
||||
|
||||
@@ -21,13 +24,17 @@ the test venv.
|
||||
|
||||
Permissive harness security rationale:
|
||||
wi-014/037/038 execute arbitrary remote code (version switch, git
|
||||
clone, pip install) and are gated at `high+` precisely to prevent
|
||||
such operations at default security. The permissive harness
|
||||
clone, pip install) and are gated precisely to prevent such
|
||||
operations at default config — wi-014 at `high+` (security_level),
|
||||
wi-037/038 by the dedicated `allow_git_url_install` /
|
||||
`allow_pip_install` flags (goal265 / #2962, independent of
|
||||
security_level, default false). The permissive harness
|
||||
(start_comfyui_permissive.sh) reflects the production use case
|
||||
these endpoints exist to serve — operators in a trusted environment
|
||||
lower security_level to normal-/weak to enable these features. The
|
||||
200 path IS a supported feature, and testing it requires exactly
|
||||
this configuration. Permissive harness uses HARDCODED trusted inputs:
|
||||
set security_level = normal- and opt in via the flags to enable
|
||||
these features. The 200 path IS a supported feature, and testing it
|
||||
requires exactly this configuration. Permissive harness uses
|
||||
HARDCODED trusted inputs:
|
||||
- wi-014: the CURRENT ComfyUI version (self-switch no-op)
|
||||
- wi-037: https://github.com/ltdrdata/nodepack-test1-do-not-install
|
||||
(project's test-fixture repo, also used by tests/cli/test_uv_compile.py)
|
||||
@@ -162,7 +169,8 @@ def comfyui_legacy():
|
||||
|
||||
def _start_comfyui_permissive() -> int:
|
||||
"""Launch via start_comfyui_permissive.sh — patches config.ini to
|
||||
`security_level = normal-` (backup at config.ini.before-permissive)
|
||||
`security_level = normal-` plus `allow_git_url_install = true` /
|
||||
`allow_pip_install = true` (backup at config.ini.before-permissive)
|
||||
then delegates to start_comfyui.sh with ENABLE_LEGACY_UI=1.
|
||||
The permissive fixture MUST restore config on teardown.
|
||||
"""
|
||||
@@ -198,9 +206,10 @@ def _restore_permissive_config():
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def comfyui_permissive():
|
||||
"""Module-scoped fixture: start server with security_level=normal-,
|
||||
tear down with config restore. Use for wi-014/037/038 which require
|
||||
`high+` (security_utils.py:20-26 allows weak/normal- at is_local_mode).
|
||||
"""Module-scoped fixture: start server with security_level=normal- and
|
||||
the dedicated install flags enabled, tear down with config restore.
|
||||
Use for wi-014 (high+ gate: weak/normal- at is_local_mode) and
|
||||
wi-037/038 (allow_git_url_install / allow_pip_install — goal265 / #2962).
|
||||
"""
|
||||
pid = _start_comfyui_permissive()
|
||||
try:
|
||||
@@ -489,7 +498,7 @@ class TestInstallViaGitUrlRealClone:
|
||||
)
|
||||
assert resp.status_code == 200, (
|
||||
f"install/git_url with trusted URL {TRUSTED_GIT_URL!r} "
|
||||
f"should return 200 at security_level=normal-, got "
|
||||
f"should return 200 with allow_git_url_install=true, got "
|
||||
f"{resp.status_code}: {resp.text[:300]}"
|
||||
)
|
||||
assert os.path.isdir(target), (
|
||||
@@ -555,7 +564,7 @@ class TestInstallPipRealExecute:
|
||||
)
|
||||
assert resp.status_code == 200, (
|
||||
f"install/pip with trusted pkg {TRUSTED_PIP_PKG!r} should "
|
||||
f"return 200 at security_level=normal-, got "
|
||||
f"return 200 with allow_pip_install=true, got "
|
||||
f"{resp.status_code}: {resp.text[:300]}"
|
||||
)
|
||||
|
||||
|
||||
@@ -0,0 +1,408 @@
|
||||
"""[goal299 step3] E2E: URL-form pip install through the S-B endpoint.
|
||||
|
||||
CONTRACT UNDER TEST (SHIPPED — goal265; this module adds tests only):
|
||||
the ``allow_pip_install`` gate on POST /v2/customnode/install/pip is
|
||||
**argument-content-agnostic** (goal299-mm-addendum.md §1): a URL-form
|
||||
argument (``git+https://github.com/...``) exercises the SAME
|
||||
dedicated-flag gate as a bare package name — no code path inspects
|
||||
the argument before the gate decision (legacy/manager_server.py:1574).
|
||||
|
||||
FIXTURE NOTE (goal329 — WHY the owned fixture): the install arm
|
||||
originally exercised ``git+https://github.com/facebookresearch/sam2``;
|
||||
that external, unpinned, heavy dependency (torch-ecosystem build at
|
||||
restart time) made the test hostage to upstream maintainer drift
|
||||
(maintainer-fragility decision, goal329). It is replaced by the OWNED,
|
||||
purpose-built fixture ``git+https://github.com/ltdrdata/pip-test1-do-not-install``
|
||||
(PUBLIC; package ``pip-test1-do-not-install``, module
|
||||
``pip_test1_do_not_install``, ``MARKER = "pip-test1-do-not-install:ok"``,
|
||||
zero deps, pure Python, no import side effects). The gate contract is
|
||||
argument-content-agnostic, so the URL swap changes NOTHING about what
|
||||
is being proven — only the install payload shrinks to a harmless,
|
||||
owned package.
|
||||
|
||||
Two arms (goal299-spec.md §1.1, LOCKED):
|
||||
|
||||
DENY arm flag=false at security_level=weak -> 403, NO reservation
|
||||
entry appended, denial log names allow_pip_install
|
||||
(weak proves the FLAG, not the security_level, decides).
|
||||
INSTALL arm flag=true at security_level=strong, loopback -> 200 =
|
||||
gate-pass + RESERVATION (deferred-execution semantics,
|
||||
addendum §1: '#FORCE' reserves into install-scripts.txt;
|
||||
the real `uv pip install` runs at the NEXT server start)
|
||||
-> restart -> fixture module REALLY importable (MARKER
|
||||
verified) in the isolated E2E venv -> self-clean ->
|
||||
absent again.
|
||||
|
||||
Placement: NEW SIBLING of tests/e2e/test_e2e_secgate_legacy_flags.py
|
||||
(spec §1.2) — that module carries an engineered zero-install guarantee
|
||||
(its header :96-100), so the real-install arm must NOT live there.
|
||||
Helper reuse contract (spec §1.2, LOCKED): import-reuse is limited to
|
||||
``_stage_flags_config`` / ``_stop_legacy`` / ``_make_flags_server_fixture``
|
||||
/ ``_post_pip`` / ``_log_offset``; everything else needed here is a
|
||||
sibling-local definition (provenance-commented copies where applicable).
|
||||
|
||||
Self-cleaning + idempotent (spec §1.3 install-arm steps 1/6): pre-guard
|
||||
uninstall, teardown uninstall (runs even on failure), reservation-file
|
||||
hygiene on both ends. The venv is uv-managed and has NO pip module
|
||||
(addendum R4) — all (un)installs go through ``<venv>/bin/uv``.
|
||||
|
||||
Requires a pre-built E2E environment (setup_e2e_env.sh); skip-marked
|
||||
otherwise. The install arm additionally requires github.com reachability
|
||||
(addendum R5) and is skip-marked offline; the deny arm always runs.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
# Import-reuse: the LOCKED helper subset only (goal299-spec.md §1.2).
|
||||
from test_e2e_secgate_legacy_flags import (
|
||||
_log_offset,
|
||||
_make_flags_server_fixture,
|
||||
_post_pip,
|
||||
_stop_legacy,
|
||||
)
|
||||
|
||||
E2E_ROOT = os.environ.get("E2E_ROOT", "")
|
||||
COMFYUI_PATH = os.path.join(E2E_ROOT, "comfyui") if E2E_ROOT else ""
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")
|
||||
SERVER_LOG = os.path.join(E2E_ROOT, "logs", "comfyui.log") if E2E_ROOT else ""
|
||||
|
||||
# Same port as the flags module (its module constant is not in the locked
|
||||
# import set; the value is part of the shared legacy-fixture contract).
|
||||
PORT = 8199
|
||||
|
||||
# Owned fixture (goal329 — see FIXTURE NOTE in the module docstring).
|
||||
# Dist name uses hyphens, import name uses underscores; MARKER gives a
|
||||
# stronger installed-for-real check than a bare import.
|
||||
FIXTURE_URL = "git+https://github.com/ltdrdata/pip-test1-do-not-install"
|
||||
FIXTURE_DIST = "pip-test1-do-not-install"
|
||||
FIXTURE_MODULE = "pip_test1_do_not_install"
|
||||
FIXTURE_MARKER = "pip-test1-do-not-install:ok"
|
||||
VENV_PY = os.path.join(E2E_ROOT, "venv", "bin", "python") if E2E_ROOT else ""
|
||||
VENV_UV = os.path.join(E2E_ROOT, "venv", "bin", "uv") if E2E_ROOT else ""
|
||||
# Reservation file (path shape per tests/e2e/test_e2e_legacy_real_ops.py:538;
|
||||
# producer: reserve_script, legacy/manager_core.py:1836-1843; consumer:
|
||||
# comfyui_manager/prestartup_script.py:485 at next server start).
|
||||
SCRIPTS_PATH = (
|
||||
os.path.join(
|
||||
COMFYUI_PATH, "user", "__manager", "startup-scripts", "install-scripts.txt"
|
||||
)
|
||||
if E2E_ROOT
|
||||
else ""
|
||||
)
|
||||
|
||||
# Post-reservation restart budget (goal329): the owned fixture is pure
|
||||
# Python with zero deps — its `uv pip install` completes in seconds, so
|
||||
# the restart wall time is dominated by the plain server boot
|
||||
# (+prestartup resolver), observed well under 60s in this harness. 180s
|
||||
# equals the proven readiness budget the flags module's _start_legacy
|
||||
# uses for a plain legacy start (>=3x headroom over observed boot); the
|
||||
# python-side timeout is shell+60 per the helper invariant.
|
||||
RESTART_BUDGET_S = 180
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not E2E_ROOT
|
||||
or not os.path.isfile(os.path.join(E2E_ROOT, ".e2e_setup_complete")),
|
||||
reason="E2E_ROOT not set or E2E environment not ready",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sibling-local helpers (spec §1.2 — outside the locked import set)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _start_legacy_install(extra_env: dict, shell_timeout: int = RESTART_BUDGET_S) -> int:
|
||||
"""Parameterized variant of test_e2e_secgate_legacy_flags._start_legacy
|
||||
(:147-158) — provenance copy per goal299-spec.md §1.2. The flags
|
||||
module's helper hardcodes its env dict and python-side timeout, which
|
||||
can neither inject per-restart env nor take a per-call readiness
|
||||
budget. TIMEOUT is forwarded to start_comfyui_legacy.sh as the shell
|
||||
readiness budget; the python-side subprocess timeout MUST exceed it
|
||||
(+60s) so the start script, not python, owns the timeout."""
|
||||
env = {
|
||||
**os.environ,
|
||||
"E2E_ROOT": E2E_ROOT,
|
||||
"PORT": str(PORT),
|
||||
"TIMEOUT": str(shell_timeout),
|
||||
**extra_env,
|
||||
}
|
||||
r = subprocess.run(
|
||||
["bash", os.path.join(SCRIPTS_DIR, "start_comfyui_legacy.sh")],
|
||||
capture_output=True, text=True, timeout=shell_timeout + 60, env=env,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"Failed to start ComfyUI (legacy, install restart):\n{r.stderr}"
|
||||
)
|
||||
for part in r.stdout.strip().split():
|
||||
if part.startswith("COMFYUI_PID="):
|
||||
return int(part.split("=")[1])
|
||||
raise RuntimeError(f"Could not parse PID:\n{r.stdout}")
|
||||
|
||||
|
||||
def _log_slice(offset: int) -> str:
|
||||
"""Provenance copy of test_e2e_secgate_legacy_flags._log_slice
|
||||
(:241-246) — not in the locked import set (spec §1.2 fallback clause:
|
||||
fixture semantics are the lock, import path is not)."""
|
||||
if not os.path.isfile(SERVER_LOG):
|
||||
return ""
|
||||
with open(SERVER_LOG, "r", encoding="utf-8", errors="replace") as f:
|
||||
f.seek(offset)
|
||||
return f.read()
|
||||
|
||||
|
||||
def _scripts_state() -> str:
|
||||
"""Current install-scripts.txt content, or the sentinel 'ABSENT' —
|
||||
the file may legitimately not exist (spec §1.3 deny step 2: it is
|
||||
created lazily by reserve_script on first reservation)."""
|
||||
if not os.path.isfile(SCRIPTS_PATH):
|
||||
return "ABSENT"
|
||||
with open(SCRIPTS_PATH, "r", encoding="utf-8", errors="replace") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
def _venv_fixture_probe() -> subprocess.CompletedProcess:
|
||||
"""Import the fixture module in the isolated E2E venv and print its
|
||||
MARKER (subprocess — the observable for really-installed /
|
||||
really-absent). rc != 0 means absent; rc == 0 AND the MARKER value in
|
||||
stdout means installed for real (stronger than a bare import)."""
|
||||
return subprocess.run(
|
||||
[VENV_PY, "-c", f"import {FIXTURE_MODULE} as m; print(m.MARKER)"],
|
||||
capture_output=True, text=True, timeout=60,
|
||||
)
|
||||
|
||||
|
||||
def _uninstall_fixture() -> None:
|
||||
"""Uninstall the fixture distribution from the E2E venv via uv.
|
||||
|
||||
The venv has NO pip module (`python -m pip` fails — addendum R4);
|
||||
`<venv>/bin/uv pip uninstall --python <venv-python>` is the working
|
||||
path. A no-op when the dist is absent (uv warns, exits 0)."""
|
||||
subprocess.run(
|
||||
[VENV_UV, "pip", "uninstall", FIXTURE_DIST, "--python", VENV_PY],
|
||||
capture_output=True, text=True, timeout=180,
|
||||
)
|
||||
|
||||
|
||||
def _strip_fixture_reservation() -> None:
|
||||
"""Remove any fixture residual line from install-scripts.txt
|
||||
(addendum R8: a leftover reservation would silently mutate the venv
|
||||
on the NEXT restart, possibly a different test's). Matches both the
|
||||
hyphenated dist/URL form and the underscored module form."""
|
||||
if not os.path.isfile(SCRIPTS_PATH):
|
||||
return
|
||||
with open(SCRIPTS_PATH, "r", encoding="utf-8", errors="replace") as f:
|
||||
lines = f.readlines()
|
||||
kept = [
|
||||
ln for ln in lines
|
||||
if FIXTURE_DIST not in ln.lower() and FIXTURE_MODULE not in ln.lower()
|
||||
]
|
||||
if kept != lines:
|
||||
with open(SCRIPTS_PATH, "w", encoding="utf-8") as f:
|
||||
f.writelines(kept)
|
||||
|
||||
|
||||
def _github_reachable() -> bool:
|
||||
"""Runtime network probe for the install arm (addendum R5)."""
|
||||
try:
|
||||
socket.create_connection(("github.com", 443), timeout=10).close()
|
||||
return True
|
||||
except OSError:
|
||||
return False
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# DENY arm — flag=false, security_level=weak (weak proves flag-not-level
|
||||
# decides; goal299-spec.md §1.3 TestPipUrlFormDeny)
|
||||
# ===========================================================================
|
||||
|
||||
comfyui_pip_url_deny = _make_flags_server_fixture(False, False, "weak")
|
||||
|
||||
|
||||
class TestPipUrlFormDeny:
|
||||
"""URL-form POST is denied by the allow_pip_install flag gate exactly
|
||||
like a bare package name (argument-content-agnostic, addendum §1):
|
||||
403, no install-side-effect (no reservation entry), denial log names
|
||||
the flag. Shared-POST shape (spec §1.3, LOCKED): exactly ONE POST via
|
||||
a class-scoped record fixture; the three tests assert exclusively
|
||||
against the record — independent of execution order, no re-reads of
|
||||
live state."""
|
||||
|
||||
@pytest.fixture(scope="class")
|
||||
def deny_post_record(self, comfyui_pip_url_deny):
|
||||
offset = _log_offset()
|
||||
snapshot_before = _scripts_state()
|
||||
resp = _post_pip(FIXTURE_URL)
|
||||
return {
|
||||
"log_offset_before": offset,
|
||||
"scripts_snapshot_before": snapshot_before,
|
||||
"status_code": resp.status_code,
|
||||
"log_slice_after": _log_slice(offset),
|
||||
"scripts_state_after": _scripts_state(),
|
||||
}
|
||||
|
||||
def test_url_form_denied_403(self, deny_post_record):
|
||||
"""403 from the dedicated-flag gate despite security_level=weak
|
||||
(deny-direction decoupling, mm §2 row 'deny arm, after POST')."""
|
||||
assert deny_post_record["status_code"] == 403, (
|
||||
f"URL-form pip POST: expected 403 (allow_pip_install=false "
|
||||
f"overrides sl=weak), got {deny_post_record['status_code']} — "
|
||||
f"the gate must be argument-content-agnostic."
|
||||
)
|
||||
|
||||
def test_url_form_deny_no_reservation(self, deny_post_record):
|
||||
"""No reservation entry appended on deny (mm §2: install-scripts.txt
|
||||
'no entry written'; HTTP 200 would have meant reservation, so the
|
||||
durable on-disk state is the decisive side-effect observable)."""
|
||||
before = deny_post_record["scripts_snapshot_before"]
|
||||
after = deny_post_record["scripts_state_after"]
|
||||
assert after == before, (
|
||||
f"deny arm: install-scripts.txt changed across the denied POST "
|
||||
f"— a reservation leaked past the gate.\nbefore:\n{before}\n"
|
||||
f"after:\n{after}"
|
||||
)
|
||||
if after != "ABSENT":
|
||||
assert FIXTURE_DIST not in after.lower(), (
|
||||
f"deny arm: a fixture line is present in install-scripts.txt "
|
||||
f"after a denied POST:\n{after}"
|
||||
)
|
||||
|
||||
def test_url_form_deny_log_names_flag(self, deny_post_record):
|
||||
"""Denial log names allow_pip_install (SECURITY_MESSAGE_FLAG_PIP,
|
||||
legacy/manager_server.py:47) and carries no security-level framing
|
||||
for this denial (goal265 spec §1.1 invariant 6 — same assertion
|
||||
shape as the flags module's SC-23)."""
|
||||
log = deny_post_record["log_slice_after"]
|
||||
assert "allow_pip_install" in log, (
|
||||
"deny arm: denial log does not name allow_pip_install "
|
||||
f"(SECURITY_MESSAGE_FLAG_PIP). Slice:\n{log[-1500:]}"
|
||||
)
|
||||
assert "security level to 'normal-'" not in log, (
|
||||
"deny arm: denial log still carries the misleading "
|
||||
"security-level copy (SECURITY_MESSAGE_NORMAL_MINUS)."
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# INSTALL arm — flag=true, security_level=strong (strong proves allow-side
|
||||
# decoupling; goal299-spec.md §1.3 TestPipUrlFormInstall)
|
||||
# ===========================================================================
|
||||
|
||||
comfyui_pip_url_install = _make_flags_server_fixture(False, True, "strong")
|
||||
|
||||
|
||||
@pytest.mark.network
|
||||
class TestPipUrlFormInstall:
|
||||
"""Full deferred-install round trip (mm §2 observable-outcome table):
|
||||
POST 200 = gate-pass + reservation (NOT installation) -> restart
|
||||
executes the reserved `uv pip install -U git+...` at prestartup
|
||||
-> fixture module imports with the expected MARKER in the isolated
|
||||
venv -> self-clean -> absent.
|
||||
|
||||
Asserting importability right after the 200 would be a guaranteed
|
||||
false-FAIL (mm §2 note) — the restart between POST and the import
|
||||
assertion is the production execution path (addendum §4.1)."""
|
||||
|
||||
@pytest.fixture(scope="class", autouse=True)
|
||||
def _network_guard(self):
|
||||
if not _github_reachable():
|
||||
pytest.skip("offline: github.com unreachable — install arm "
|
||||
"requires network (deny arm unaffected)")
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _self_clean(self):
|
||||
"""Teardown runs even on failure (spec §1.3 step 6): uninstall the
|
||||
dist, strip any residual reservation line (addendum R8), and prove
|
||||
the venv is back to baseline (import fails again)."""
|
||||
yield
|
||||
_uninstall_fixture()
|
||||
_strip_fixture_reservation()
|
||||
assert _venv_fixture_probe().returncode != 0, (
|
||||
f"self-clean: {FIXTURE_MODULE} is still importable in the E2E "
|
||||
f"venv after `uv pip uninstall {FIXTURE_DIST}` — venv NOT "
|
||||
f"restored to baseline."
|
||||
)
|
||||
|
||||
def test_url_form_install_end_to_end(self, comfyui_pip_url_install):
|
||||
# Step 1 — pre-guard (idempotency, spec §1.3 step 1): a previous
|
||||
# aborted run must not turn this into a false-PASS.
|
||||
_uninstall_fixture()
|
||||
_strip_fixture_reservation()
|
||||
assert _venv_fixture_probe().returncode != 0, (
|
||||
f"pre-guard: {FIXTURE_MODULE} already importable before the "
|
||||
f"test — uninstall guard failed; venv not at baseline."
|
||||
)
|
||||
|
||||
# Step 2 — POST the URL-form argument; 200 = gate-pass +
|
||||
# reservation under the deferred-execution semantics (mm §1).
|
||||
resp = _post_pip(FIXTURE_URL)
|
||||
assert resp.status_code == 200, (
|
||||
f"install arm: expected 200 (allow_pip_install=true overrides "
|
||||
f"sl=strong), got {resp.status_code} — allow-side decoupling "
|
||||
f"broken for the URL-form argument."
|
||||
)
|
||||
|
||||
# Step 3 — reservation entry: a Python-list-repr line containing
|
||||
# the fixture URL substring (producer reserve_script,
|
||||
# legacy/manager_core.py:1836-1843; precedent assertion shape
|
||||
# test_e2e_legacy_real_ops.py:516-538). NOT a shell-string match.
|
||||
state = _scripts_state()
|
||||
assert state != "ABSENT", (
|
||||
"install arm: install-scripts.txt missing after 200 — "
|
||||
"no reservation was written."
|
||||
)
|
||||
fixture_lines = [ln for ln in state.splitlines() if FIXTURE_URL in ln]
|
||||
assert fixture_lines, (
|
||||
f"install arm: no reservation line contains {FIXTURE_URL!r}.\n"
|
||||
f"file content:\n{state}"
|
||||
)
|
||||
reserved = ast.literal_eval(fixture_lines[0])
|
||||
assert isinstance(reserved, list) and "#FORCE" in reserved, (
|
||||
f"install arm: reservation line is not the expected "
|
||||
f"['..', '#FORCE', <pip cmd>...] list-repr shape: {reserved!r}"
|
||||
)
|
||||
|
||||
# Step 4 — restart: the production execution path for the
|
||||
# reservation (prestartup_script.py:485 consumes the file at
|
||||
# server start). Budget rationale at RESTART_BUDGET_S — the
|
||||
# owned zero-dep fixture installs in seconds; the budget covers
|
||||
# the plain server boot, no build env vars needed (goal329
|
||||
# retired the CUDA-build determinism knob with the heavy
|
||||
# upstream package).
|
||||
_stop_legacy()
|
||||
_start_legacy_install({}, shell_timeout=RESTART_BUDGET_S)
|
||||
|
||||
# Step 5 — post-restart: reservation CONSUMED (prestartup removes
|
||||
# the processed file) and the fixture REALLY importable in the
|
||||
# venv, proven by its MARKER value (stronger than bare import:
|
||||
# the marker ties the import to the owned fixture's content).
|
||||
post_state = _scripts_state()
|
||||
assert post_state == "ABSENT" or FIXTURE_URL not in post_state, (
|
||||
f"install arm: reservation NOT consumed by the restart — "
|
||||
f"install-scripts.txt still references the fixture:\n{post_state}"
|
||||
)
|
||||
probe = _venv_fixture_probe()
|
||||
assert probe.returncode == 0, (
|
||||
f"install arm: `import {FIXTURE_MODULE}` still fails in the "
|
||||
f"E2E venv after the reservation-executing restart — install "
|
||||
f"did not happen or did not target the venv.\n"
|
||||
f"stderr:\n{probe.stderr[-500:]}"
|
||||
)
|
||||
assert FIXTURE_MARKER in probe.stdout, (
|
||||
f"install arm: fixture imported but MARKER mismatch — "
|
||||
f"expected {FIXTURE_MARKER!r} in stdout, got: {probe.stdout!r}"
|
||||
)
|
||||
# Version breadcrumb for triage (owned repo, so drift risk is
|
||||
# retired — the breadcrumb now just documents what was installed).
|
||||
show = subprocess.run(
|
||||
[VENV_UV, "pip", "show", FIXTURE_DIST, "--python", VENV_PY],
|
||||
capture_output=True, text=True, timeout=60,
|
||||
)
|
||||
print(f"\n[goal329 install-evidence] uv pip show {FIXTURE_DIST}:\n"
|
||||
f"{show.stdout}")
|
||||
@@ -0,0 +1,641 @@
|
||||
"""[goal265 step4 — RED] E2E endpoint binding for the dedicated install flags
|
||||
``allow_git_url_install`` / ``allow_pip_install`` on the LEGACY server.
|
||||
|
||||
TARGET CONTRACT (NOT YET IMPLEMENTED — goal265-spec.md §1, LOCKED):
|
||||
|
||||
S-A POST /v2/customnode/install/git_url gated by allow_git_url_install
|
||||
S-B POST /v2/customnode/install/pip gated by allow_pip_install
|
||||
S-C batch unknown-URL install: retained middle+ entry gate AND the
|
||||
allow_git_url_install full predicate replaces the high+ risky check;
|
||||
unknown-pip 'block' branch stays UNCONDITIONAL (Q1).
|
||||
The security_level term is fully REMOVED from S-A/S-B (spec §1.1 inv. 1);
|
||||
denial logs name the responsible flag, not the security level (inv. 6).
|
||||
|
||||
These tests are EXPECTED TO FAIL against current code (today S-A/S-B are
|
||||
``is_allowed_security_level('high+')``-gated) — RED confirmation is Step 5.
|
||||
Do NOT weaken them to pass against today's code.
|
||||
|
||||
This module delivers the LGU2/LPP2 legacy-fixture follow-up explicitly
|
||||
deferred by tests/e2e/test_e2e_secgate_default.py (its header, lines 31-34).
|
||||
|
||||
SC rows covered (goal265-scenarios.md), grouped by server config combination
|
||||
(one legacy-server launch per combination, config.ini staged per flags):
|
||||
|
||||
CFG-A (git=T, pip=T, sl=strong, nm=public): SC-01, SC-11, SC-08
|
||||
CFG-B (git=T, pip=F, sl=normal, nm=public): SC-02, SC-04, SC-10, SC-17,
|
||||
SC-18 (transitive-pip arm)
|
||||
CFG-C (git=F, pip=T, sl=normal, nm=public): SC-16
|
||||
CFG-D (git=F, pip=F, sl=weak, nm=public): SC-05, SC-09, SC-13,
|
||||
SC-23 (denial-log copy arm)
|
||||
CFG-E (git=T, pip=T, sl=weak, nm=public): SC-24 (Q1 unknown-pip block)
|
||||
CFG-F (git=T, pip=T, sl=normal, nm=public): SC-25A/B (guards, flags-on)
|
||||
CFG-G (git=F, pip=F, sl=normal, nm=public): SC-25A/B (guards, flags-off)
|
||||
CFG-H (git=T, pip=F, sl=normal, nm=personal_cloud): SC-30 (note below)
|
||||
|
||||
SC-30 limitation (deliberate, documented): the E2E harness listens on
|
||||
127.0.0.1, so the personal_cloud arm of the middle+ entry gate is satisfied
|
||||
via is_local_mode as well; the public-listener arm of personal_cloud is
|
||||
proven at predicate level (tests/common/test_install_flag_predicate.py
|
||||
SC-03/SC-12) — this row re-proves the endpoint binding with
|
||||
network_mode=personal_cloud staged.
|
||||
|
||||
SC-18 strategy: endpoint-level arm — git-URL install transaction completes
|
||||
(200) with allow_pip_install=false and the captured log slice contains NO
|
||||
allow_pip_install denial, proving the pip flag is never consulted inside a
|
||||
git transaction (MM §1.4). The spec's integration-level alternative
|
||||
(in-process gitclone_install with execute_install_script mocked) remains
|
||||
open to Step-7 if this arm proves under-discriminating.
|
||||
|
||||
Side-effect control: batch rows pre-seed the target node directory so the
|
||||
queue worker's gitclone_install short-circuits on "Already exists" — the
|
||||
GATE decision (the contract under test) happens synchronously in
|
||||
_install_custom_node BEFORE the worker runs, so the batch response's
|
||||
``failed`` list is a complete gate observable without real clones.
|
||||
|
||||
Requires a pre-built E2E environment (setup_e2e_env.sh); skip-marked
|
||||
otherwise (harness precedent: tests/e2e/test_e2e_secgate_default.py).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import configparser
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
E2E_ROOT = os.environ.get("E2E_ROOT", "")
|
||||
COMFYUI_PATH = os.path.join(E2E_ROOT, "comfyui") if E2E_ROOT else ""
|
||||
SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "scripts")
|
||||
|
||||
PORT = 8199
|
||||
BASE_URL = f"http://127.0.0.1:{PORT}"
|
||||
|
||||
CONFIG_PATH = os.path.join(COMFYUI_PATH, "user", "__manager", "config.ini")
|
||||
CONFIG_BACKUP = CONFIG_PATH + ".before-flags"
|
||||
SERVER_LOG = os.path.join(E2E_ROOT, "logs", "comfyui.log") if E2E_ROOT else ""
|
||||
|
||||
# Purpose-built test-fixture repo (same constant as
|
||||
# tests/e2e/test_e2e_legacy_real_ops.py TRUSTED_GIT_URL) — the DESIGNATED
|
||||
# do-not-install sample for every S-A surface test that performs a REAL
|
||||
# clone (direct /v2/customnode/install/git_url).
|
||||
UNKNOWN_GIT_URL = "https://github.com/ltdrdata/nodepack-test1-do-not-install"
|
||||
UNKNOWN_GIT_DIRNAME = "nodepack-test1-do-not-install"
|
||||
|
||||
# Batch S-C "unknown-URL" rows need a files URL genuinely NOT in the
|
||||
# custom-node DB (scenario preconditions SC-04/08/09/30: "file URL NOT in
|
||||
# node DB" -> get_risky_level returns 'high+' -> the dedicated-flag gate
|
||||
# engages). DISCOVERY (task #297 real-server run): the do-not-install
|
||||
# sample above IS registered in the Comfy-Org main-channel
|
||||
# custom-node-list (served from the manager cache), so get_risky_level
|
||||
# returns 'middle+' for it and the flag gate is bypassed — using it for
|
||||
# batch rows silently tests the wrong path (observed as an SC-09
|
||||
# false-FAIL: flag=false yet the item queued via middle+ at sl=weak).
|
||||
# The batch rows therefore use a SYNTHETIC nonexistent URL. ZERO-INSTALL
|
||||
# guarantee holds by construction: deny rows (SC-08/09) never reach the
|
||||
# worker, and allow rows (SC-04/30) pre-seed the matching placeholder dir
|
||||
# so the worker's gitclone_install short-circuits on "Already exists" —
|
||||
# the URL is never cloned (and could not be: the repo does not exist).
|
||||
BATCH_UNKNOWN_GIT_URL = "https://github.com/ltdrdata/goal265-nonexistent-unknown-node"
|
||||
BATCH_UNKNOWN_DIRNAME = "goal265-nonexistent-unknown-node"
|
||||
|
||||
# Long-standing custom-node-list.json entry — used by SC-24/SC-25B rows that
|
||||
# need a files URL the DB KNOWS (get_risky_level must not short-circuit at
|
||||
# high+ on the URL check).
|
||||
KNOWN_GIT_URL = "https://github.com/ltdrdata/ComfyUI-Impact-Pack"
|
||||
KNOWN_GIT_DIRNAME = "ComfyUI-Impact-Pack"
|
||||
|
||||
UNKNOWN_PIP_PKG = "cm-goal265-nonexistent-pip-pkg-xyz"
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not E2E_ROOT
|
||||
or not os.path.isfile(os.path.join(E2E_ROOT, ".e2e_setup_complete")),
|
||||
reason="E2E_ROOT not set or E2E environment not ready",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config staging + server lifecycle helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _stage_flags_config(git: bool, pip: bool, security_level: str,
|
||||
network_mode: str) -> None:
|
||||
"""Patch config.ini with the row preconditions (backup-once pattern,
|
||||
mirrors start_comfyui_strict.sh)."""
|
||||
if not os.path.isfile(CONFIG_BACKUP):
|
||||
shutil.copy(CONFIG_PATH, CONFIG_BACKUP)
|
||||
|
||||
cfg = configparser.ConfigParser(strict=False)
|
||||
cfg.read(CONFIG_PATH)
|
||||
if "default" not in cfg:
|
||||
cfg["default"] = {}
|
||||
cfg["default"]["allow_git_url_install"] = "true" if git else "false"
|
||||
cfg["default"]["allow_pip_install"] = "true" if pip else "false"
|
||||
cfg["default"]["security_level"] = security_level
|
||||
cfg["default"]["network_mode"] = network_mode
|
||||
with open(CONFIG_PATH, "w", encoding="utf-8") as f:
|
||||
cfg.write(f)
|
||||
|
||||
|
||||
def _restore_config() -> None:
|
||||
if os.path.isfile(CONFIG_BACKUP):
|
||||
shutil.move(CONFIG_BACKUP, CONFIG_PATH)
|
||||
|
||||
|
||||
def _start_legacy() -> int:
|
||||
env = {**os.environ, "E2E_ROOT": E2E_ROOT, "PORT": str(PORT)}
|
||||
r = subprocess.run(
|
||||
["bash", os.path.join(SCRIPTS_DIR, "start_comfyui_legacy.sh")],
|
||||
capture_output=True, text=True, timeout=180, env=env,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
raise RuntimeError(f"Failed to start ComfyUI (legacy):\n{r.stderr}")
|
||||
for part in r.stdout.strip().split():
|
||||
if part.startswith("COMFYUI_PID="):
|
||||
return int(part.split("=")[1])
|
||||
raise RuntimeError(f"Could not parse PID:\n{r.stdout}")
|
||||
|
||||
|
||||
def _stop_legacy() -> None:
|
||||
env = {**os.environ, "E2E_ROOT": E2E_ROOT, "PORT": str(PORT)}
|
||||
subprocess.run(
|
||||
["bash", os.path.join(SCRIPTS_DIR, "stop_comfyui.sh")],
|
||||
capture_output=True, text=True, timeout=30, env=env,
|
||||
)
|
||||
|
||||
|
||||
def _make_flags_server_fixture(git: bool, pip: bool, security_level: str,
|
||||
network_mode: str = "public"):
|
||||
"""Class-scoped fixture factory: stage config -> start legacy server ->
|
||||
yield -> stop server -> restore config."""
|
||||
|
||||
@pytest.fixture(scope="class")
|
||||
def _server():
|
||||
_stage_flags_config(git, pip, security_level, network_mode)
|
||||
try:
|
||||
pid = _start_legacy()
|
||||
try:
|
||||
yield pid
|
||||
finally:
|
||||
_stop_legacy()
|
||||
finally:
|
||||
_restore_config()
|
||||
|
||||
return _server
|
||||
|
||||
|
||||
comfyui_flags_a = _make_flags_server_fixture(True, True, "strong")
|
||||
comfyui_flags_b = _make_flags_server_fixture(True, False, "normal")
|
||||
comfyui_flags_c = _make_flags_server_fixture(False, True, "normal")
|
||||
comfyui_flags_d = _make_flags_server_fixture(False, False, "weak")
|
||||
comfyui_flags_e = _make_flags_server_fixture(True, True, "weak")
|
||||
comfyui_flags_f = _make_flags_server_fixture(True, True, "normal")
|
||||
comfyui_flags_g = _make_flags_server_fixture(False, False, "normal")
|
||||
comfyui_flags_h = _make_flags_server_fixture(
|
||||
True, False, "normal", network_mode="personal_cloud")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Request / observation helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _post_git_url(url: str) -> requests.Response:
|
||||
return requests.post(
|
||||
f"{BASE_URL}/v2/customnode/install/git_url", data=url, timeout=120,
|
||||
)
|
||||
|
||||
|
||||
def _post_pip(packages: str) -> requests.Response:
|
||||
return requests.post(
|
||||
f"{BASE_URL}/v2/customnode/install/pip", data=packages, timeout=60,
|
||||
)
|
||||
|
||||
|
||||
def _batch_install_item(ui_id: str, files: list[str],
|
||||
pip: list[str] | None = None) -> dict:
|
||||
"""Unknown-version batch install item shape consumed by
|
||||
legacy _install_custom_node (version='unknown' -> files URL path)."""
|
||||
return {
|
||||
"id": ui_id,
|
||||
"ui_id": ui_id,
|
||||
"version": "unknown",
|
||||
"files": files,
|
||||
"pip": pip or [],
|
||||
"channel": "default",
|
||||
"mode": "cache",
|
||||
}
|
||||
|
||||
|
||||
def _post_batch(payload: dict) -> requests.Response:
|
||||
return requests.post(
|
||||
f"{BASE_URL}/v2/manager/queue/batch", json=payload, timeout=120,
|
||||
)
|
||||
|
||||
|
||||
def _log_offset() -> int:
|
||||
return os.path.getsize(SERVER_LOG) if os.path.isfile(SERVER_LOG) else 0
|
||||
|
||||
|
||||
def _log_slice(offset: int) -> str:
|
||||
if not os.path.isfile(SERVER_LOG):
|
||||
return ""
|
||||
with open(SERVER_LOG, "r", encoding="utf-8", errors="replace") as f:
|
||||
f.seek(offset)
|
||||
return f.read()
|
||||
|
||||
|
||||
def _custom_nodes_dir() -> str:
|
||||
return os.path.join(COMFYUI_PATH, "custom_nodes")
|
||||
|
||||
|
||||
def _preseed_node_dir(dirname: str) -> str:
|
||||
"""Create a placeholder node dir so the queue worker's gitclone_install
|
||||
short-circuits on 'Already exists' (no real clone; the gate decision
|
||||
under test already happened synchronously)."""
|
||||
target = os.path.join(_custom_nodes_dir(), dirname)
|
||||
os.makedirs(target, exist_ok=True)
|
||||
marker = os.path.join(target, ".goal265-preseed")
|
||||
with open(marker, "w", encoding="utf-8") as f:
|
||||
f.write("goal265 step4 gate-test placeholder\n")
|
||||
return target
|
||||
|
||||
|
||||
def _remove_node_dir(dirname: str) -> None:
|
||||
for candidate in (
|
||||
os.path.join(_custom_nodes_dir(), dirname),
|
||||
os.path.join(_custom_nodes_dir(), ".disabled", dirname),
|
||||
os.path.join(_custom_nodes_dir(), dirname + ".disabled"),
|
||||
):
|
||||
if os.path.isdir(candidate):
|
||||
shutil.rmtree(candidate, ignore_errors=True)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-A: git=T, pip=T, sl=strong, nm=public
|
||||
# ===========================================================================
|
||||
|
||||
class TestFlagsOnStrongLevel:
|
||||
"""Allow-direction decoupling: flags=true must allow S-A/S-B even at
|
||||
security_level=strong (today: 403). Batch entry gate stays
|
||||
security_level-coupled (middle+ fails at strong)."""
|
||||
|
||||
def test_sc01_git_url_allowed_at_strong_when_flag_true(
|
||||
self, comfyui_flags_a):
|
||||
"""SC-01: git=true, sl=strong, loopback -> POST git_url returns 200
|
||||
(today 403 — proves security_level irrelevance, allow direction).
|
||||
Pre-removes the fixture node so a real (tiny) clone proceeds; a
|
||||
200 via res.action=='skip' is equally a gate-pass observable."""
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
try:
|
||||
resp = _post_git_url(UNKNOWN_GIT_URL)
|
||||
assert resp.status_code == 200, (
|
||||
f"SC-01: expected 200 (flag=true overrides sl=strong), got "
|
||||
f"{resp.status_code} — security_level still gates S-A. "
|
||||
f"Body: {resp.text[:200]}"
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
|
||||
def test_sc11_pip_allowed_at_strong_when_flag_true(self, comfyui_flags_a):
|
||||
"""SC-11: pip=true, sl=strong, loopback -> POST pip returns 200
|
||||
(today 403). text-unidecode: pure-Python, tiny, idempotent — same
|
||||
trusted constant as test_e2e_legacy_real_ops.py."""
|
||||
resp = _post_pip("text-unidecode")
|
||||
assert resp.status_code == 200, (
|
||||
f"SC-11: expected 200 (flag=true overrides sl=strong), got "
|
||||
f"{resp.status_code} — security_level still gates S-B."
|
||||
)
|
||||
|
||||
def test_sc08_batch_unknown_url_denied_at_strong(self, comfyui_flags_a):
|
||||
"""SC-08: git=true, sl=strong -> batch unknown-URL install DENIED —
|
||||
the retained middle+ entry gate fails first (composite gate:
|
||||
middle+ AND flag; spec §1.2 S-C row, ':1427' gate UNCHANGED)."""
|
||||
item = _batch_install_item("sc08-unknown", [BATCH_UNKNOWN_GIT_URL])
|
||||
resp = _post_batch({"install": [item]})
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
assert "sc08-unknown" in failed, (
|
||||
"SC-08: batch unknown-URL install must stay denied at "
|
||||
"sl=strong (middle+ entry gate) even with "
|
||||
f"allow_git_url_install=true — failed={failed!r}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-B: git=T, pip=F, sl=normal, nm=public
|
||||
# ===========================================================================
|
||||
|
||||
class TestGitFlagOnDefaultLevel:
|
||||
"""Default security_level: git flag alone opens S-A; pip stays closed."""
|
||||
|
||||
def test_sc02_git_url_allowed_at_default_level(self, comfyui_flags_b):
|
||||
"""SC-02: git=true, sl=normal (default), loopback -> 200
|
||||
(today 403). Self-cleaning (pre+post remove): a leftover clone
|
||||
would turn the NEXT install of the same repo into an
|
||||
'Already exists' 400 (gitclone_install unknown-repo path has no
|
||||
skip arm — observed as an SC-17 false-FAIL in the first #297 run)."""
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
try:
|
||||
resp = _post_git_url(UNKNOWN_GIT_URL)
|
||||
assert resp.status_code == 200, (
|
||||
f"SC-02: expected 200 at sl=normal with flag=true, got "
|
||||
f"{resp.status_code}. Body: {resp.text[:200]}"
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
|
||||
def test_sc10_invalid_url_reaches_validation_400(self, comfyui_flags_b):
|
||||
"""SC-10: git=true, sl=normal -> POST an INVALID url -> 400 from
|
||||
installer validation, NOT 403 from the gate. Distinguishes gate-403
|
||||
from install-400 (today the gate 403s before URL validation)."""
|
||||
resp = _post_git_url("not-a-url")
|
||||
assert resp.status_code == 400, (
|
||||
f"SC-10: expected 400 (installer validation), got "
|
||||
f"{resp.status_code} — a 403 means the flag gate did not open."
|
||||
)
|
||||
|
||||
def test_sc17_git_open_pip_closed_independent(self, comfyui_flags_b):
|
||||
"""SC-17: git=true, pip=false, sl=normal -> git_url 200, pip 403
|
||||
(flags fully independent, [D1][D2]). Self-cleaning like SC-02 —
|
||||
a stale clone from an earlier S-A test would 400 the git arm."""
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
try:
|
||||
git_resp = _post_git_url(UNKNOWN_GIT_URL)
|
||||
pip_resp = _post_pip("text-unidecode")
|
||||
assert git_resp.status_code == 200, (
|
||||
f"SC-17 git arm: expected 200, got {git_resp.status_code}"
|
||||
)
|
||||
assert pip_resp.status_code == 403, (
|
||||
f"SC-17 pip arm: expected 403, got {pip_resp.status_code}"
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
|
||||
def test_sc18_git_transaction_never_consults_pip_flag(
|
||||
self, comfyui_flags_b):
|
||||
"""SC-18: git=true, pip=false -> a git-URL install transaction
|
||||
(clone + execute_install_script dependency step) COMPLETES, and the
|
||||
captured log slice contains no allow_pip_install denial — the pip
|
||||
flag governs ONLY the standalone S-B surface (MM §1.4 transaction
|
||||
scope; spec §1.1 invariant 4).
|
||||
|
||||
Fresh install forced (pre-remove) so execute_install_script actually
|
||||
runs inside the transaction window."""
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
offset = _log_offset()
|
||||
try:
|
||||
resp = _post_git_url(UNKNOWN_GIT_URL)
|
||||
log = _log_slice(offset)
|
||||
assert resp.status_code == 200, (
|
||||
f"SC-18: git transaction failed ({resp.status_code}) with "
|
||||
"pip=false — transitive scope broken? Body: "
|
||||
f"{resp.text[:200]}"
|
||||
)
|
||||
assert "allow_pip_install" not in log, (
|
||||
"SC-18: the git-URL install transaction consulted/logged "
|
||||
"allow_pip_install — per-surface scope violated (MM §1.4):\n"
|
||||
+ log[-2000:]
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(UNKNOWN_GIT_DIRNAME)
|
||||
|
||||
def test_sc04_batch_unknown_url_queued_at_default_level(
|
||||
self, comfyui_flags_b):
|
||||
"""SC-04: git=true, sl=normal -> batch unknown-URL install is
|
||||
QUEUED (failed list empty): middle+ entry passes at normal local
|
||||
AND the flag replaces the high+ risky check (today 403: risky
|
||||
high+ fails at normal). Target dir pre-seeded — see module
|
||||
docstring 'Side-effect control'."""
|
||||
_preseed_node_dir(BATCH_UNKNOWN_DIRNAME)
|
||||
try:
|
||||
item = _batch_install_item("sc04-unknown", [BATCH_UNKNOWN_GIT_URL])
|
||||
resp = _post_batch({"install": [item]})
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
assert "sc04-unknown" not in failed, (
|
||||
"SC-04: batch unknown-URL install still denied at "
|
||||
"sl=normal with allow_git_url_install=true — the risky "
|
||||
f"high+ check was not replaced by the flag. failed={failed!r}"
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(BATCH_UNKNOWN_DIRNAME)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-C: git=F, pip=T, sl=normal, nm=public
|
||||
# ===========================================================================
|
||||
|
||||
class TestPipFlagOnDefaultLevel:
|
||||
def test_sc16_git_closed_pip_open_independent(self, comfyui_flags_c):
|
||||
"""SC-16: git=false, pip=true, sl=normal -> git_url 403, pip 200
|
||||
(flags fully independent, [D1][D2])."""
|
||||
git_resp = _post_git_url(UNKNOWN_GIT_URL)
|
||||
pip_resp = _post_pip("text-unidecode")
|
||||
assert git_resp.status_code == 403, (
|
||||
f"SC-16 git arm: expected 403, got {git_resp.status_code}"
|
||||
)
|
||||
assert pip_resp.status_code == 200, (
|
||||
f"SC-16 pip arm: expected 200, got {pip_resp.status_code}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-D: git=F, pip=F, sl=weak, nm=public
|
||||
# ===========================================================================
|
||||
|
||||
class TestFlagsOffWeakLevel:
|
||||
"""Deny-direction decoupling: flags=false must deny S-A/S-B even at
|
||||
security_level=weak (today: 200) — and the denial log must name the
|
||||
responsible flag (SC-23 log arm, spec §1.3)."""
|
||||
|
||||
def test_sc05_sc23_git_url_denied_at_weak_log_names_flag(
|
||||
self, comfyui_flags_d):
|
||||
"""SC-05: git=false, sl=weak, loopback -> 403 (today 200 — deny
|
||||
direction of security_level irrelevance).
|
||||
SC-23 (log arm): the denial log names allow_git_url_install and
|
||||
drops the security-level framing (SECURITY_MESSAGE_NORMAL_MINUS)."""
|
||||
offset = _log_offset()
|
||||
resp = _post_git_url(UNKNOWN_GIT_URL)
|
||||
log = _log_slice(offset)
|
||||
|
||||
assert resp.status_code == 403, (
|
||||
f"SC-05: expected 403 (flag=false overrides sl=weak), got "
|
||||
f"{resp.status_code} — security_level still opens S-A."
|
||||
)
|
||||
assert "allow_git_url_install" in log, (
|
||||
"SC-23: denial log does not name allow_git_url_install "
|
||||
f"(spec §1.3 SECURITY_MESSAGE_FLAG_GIT_URL). Slice:\n{log[-1500:]}"
|
||||
)
|
||||
assert "security level to 'normal-'" not in log, (
|
||||
"SC-23: denial log still carries the misleading "
|
||||
"security-level copy (SECURITY_MESSAGE_NORMAL_MINUS)."
|
||||
)
|
||||
|
||||
def test_sc13_sc23_pip_denied_at_weak_log_names_flag(
|
||||
self, comfyui_flags_d):
|
||||
"""SC-13: pip=false, sl=weak, loopback -> 403 (today 200).
|
||||
SC-23 (log arm): denial log names allow_pip_install
|
||||
(spec §1.3 SECURITY_MESSAGE_FLAG_PIP)."""
|
||||
offset = _log_offset()
|
||||
resp = _post_pip("text-unidecode")
|
||||
log = _log_slice(offset)
|
||||
|
||||
assert resp.status_code == 403, (
|
||||
f"SC-13: expected 403 (flag=false overrides sl=weak), got "
|
||||
f"{resp.status_code} — security_level still opens S-B."
|
||||
)
|
||||
assert "allow_pip_install" in log, (
|
||||
"SC-23: denial log does not name allow_pip_install "
|
||||
f"(spec §1.3 SECURITY_MESSAGE_FLAG_PIP). Slice:\n{log[-1500:]}"
|
||||
)
|
||||
|
||||
def test_sc09_batch_unknown_url_denied_when_flag_false(
|
||||
self, comfyui_flags_d):
|
||||
"""SC-09: git=false, sl=weak -> batch unknown-URL install denied —
|
||||
the flag gate fails despite weak (today: allowed). Composite gate's
|
||||
flag term is the decider here (middle+ passes at weak local)."""
|
||||
item = _batch_install_item("sc09-unknown", [BATCH_UNKNOWN_GIT_URL])
|
||||
resp = _post_batch({"install": [item]})
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
assert "sc09-unknown" in failed, (
|
||||
"SC-09: batch unknown-URL install must be denied with "
|
||||
f"allow_git_url_install=false even at sl=weak. failed={failed!r}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-E: git=T, pip=T, sl=weak, nm=public — Q1 guard
|
||||
# ===========================================================================
|
||||
|
||||
class TestUnknownPipStaysBlocked:
|
||||
def test_sc24_batch_unknown_pip_block_is_unconditional(
|
||||
self, comfyui_flags_e):
|
||||
"""SC-24 (Q1 guard): batch item whose files are ALL DB-known (no
|
||||
high+ short-circuit on the URL check) but whose pip list names an
|
||||
UNKNOWN package -> still denied. get_risky_level's 'block' branch
|
||||
stays UNCONDITIONAL — the flags do NOT open it (spec §5 freeze
|
||||
item 7), even with both flags true at sl=weak."""
|
||||
item = _batch_install_item(
|
||||
"sc24-known-url-unknown-pip",
|
||||
[KNOWN_GIT_URL],
|
||||
pip=[UNKNOWN_PIP_PKG],
|
||||
)
|
||||
resp = _post_batch({"install": [item]})
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
assert "sc24-known-url-unknown-pip" in failed, (
|
||||
"SC-24: unknown-pip 'block' branch opened — it must remain "
|
||||
f"unconditional regardless of the new flags. failed={failed!r}"
|
||||
)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-F / CFG-G: out-of-scope legacy guards (SC-25A/B) — flags have NO effect
|
||||
# ===========================================================================
|
||||
|
||||
def _batch_op_item(ui_id: str) -> dict:
|
||||
"""Item for middle-gated batch ops (fix/uninstall/update). The GATE
|
||||
check precedes node resolution, so a placeholder spec is sufficient:
|
||||
the queue worker's later failure on the nonexistent node is irrelevant
|
||||
to the synchronous gate observable (the response's failed list)."""
|
||||
return {
|
||||
"id": ui_id,
|
||||
"ui_id": ui_id,
|
||||
"version": "unknown",
|
||||
"files": [f"https://example.com/{ui_id}"],
|
||||
}
|
||||
|
||||
|
||||
class _LegacyGuardRows:
|
||||
"""Shared assertions for SC-25A/B — executed under BOTH flag combos
|
||||
((t,t) via CFG-F and (f,f) via CFG-G); outcomes must be IDENTICAL,
|
||||
proving the flags have zero effect on middle/middle+ surfaces."""
|
||||
|
||||
flags_label = ""
|
||||
|
||||
def _assert_sc25a(self):
|
||||
"""SC-25A: legacy middle-gated ops (fix/uninstall/update) stay
|
||||
allowed at sl=normal local for this flag combination."""
|
||||
payload = {
|
||||
"fix": [_batch_op_item(f"sc25a-fix-{self.flags_label}")],
|
||||
"uninstall": [_batch_op_item(f"sc25a-unin-{self.flags_label}")],
|
||||
"update": [_batch_op_item(f"sc25a-upd-{self.flags_label}")],
|
||||
}
|
||||
resp = _post_batch(payload)
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
gate_denied = [x for x in failed if x.startswith("sc25a-")]
|
||||
assert gate_denied == [], (
|
||||
f"SC-25A({self.flags_label}): middle-gated batch ops were "
|
||||
f"gate-denied at sl=normal — flags must have ZERO effect on "
|
||||
f"out-of-scope surfaces. failed={failed!r}"
|
||||
)
|
||||
|
||||
def _assert_sc25b(self):
|
||||
"""SC-25B: batch KNOWN-node install (middle+ path, DB-resolved
|
||||
files URL) stays allowed at sl=normal local for this flag
|
||||
combination. Known-node dir pre-seeded — no real install."""
|
||||
_preseed_node_dir(KNOWN_GIT_DIRNAME)
|
||||
try:
|
||||
item = _batch_install_item(
|
||||
f"sc25b-known-{self.flags_label}", [KNOWN_GIT_URL])
|
||||
resp = _post_batch({"install": [item]})
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
assert f"sc25b-known-{self.flags_label}" not in failed, (
|
||||
f"SC-25B({self.flags_label}): KNOWN-node batch install was "
|
||||
f"denied at sl=normal — middle+ path must be unaffected by "
|
||||
f"the flags. failed={failed!r}"
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(KNOWN_GIT_DIRNAME)
|
||||
|
||||
|
||||
class TestGuardsFlagsOn(_LegacyGuardRows):
|
||||
flags_label = "flags-on"
|
||||
|
||||
def test_sc25a_middle_ops_allowed_with_flags_on(self, comfyui_flags_f):
|
||||
self._assert_sc25a()
|
||||
|
||||
def test_sc25b_known_install_allowed_with_flags_on(self, comfyui_flags_f):
|
||||
self._assert_sc25b()
|
||||
|
||||
|
||||
class TestGuardsFlagsOff(_LegacyGuardRows):
|
||||
flags_label = "flags-off"
|
||||
|
||||
def test_sc25a_middle_ops_allowed_with_flags_off(self, comfyui_flags_g):
|
||||
self._assert_sc25a()
|
||||
|
||||
def test_sc25b_known_install_allowed_with_flags_off(self, comfyui_flags_g):
|
||||
self._assert_sc25b()
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CFG-H: git=T, pip=F, sl=normal, nm=personal_cloud
|
||||
# ===========================================================================
|
||||
|
||||
class TestPersonalCloudBatch:
|
||||
def test_sc30_batch_unknown_url_queued_under_personal_cloud(
|
||||
self, comfyui_flags_h):
|
||||
"""SC-30: git=true, sl=normal, nm=personal_cloud -> batch
|
||||
unknown-URL install queued: the S-C composite gate's middle+ arm
|
||||
passes (is_local_mode OR is_personal_cloud) AND the flag is true.
|
||||
See module docstring for the loopback-harness limitation; the
|
||||
public-listener personal_cloud arm is proven at predicate level."""
|
||||
_preseed_node_dir(BATCH_UNKNOWN_DIRNAME)
|
||||
try:
|
||||
item = _batch_install_item("sc30-unknown", [BATCH_UNKNOWN_GIT_URL])
|
||||
resp = _post_batch({"install": [item]})
|
||||
assert resp.status_code == 200, resp.text[:200]
|
||||
failed = resp.json().get("failed", [])
|
||||
assert "sc30-unknown" not in failed, (
|
||||
"SC-30: batch unknown-URL install denied under "
|
||||
f"network_mode=personal_cloud with flag=true. "
|
||||
f"failed={failed!r}"
|
||||
)
|
||||
finally:
|
||||
_remove_node_dir(BATCH_UNKNOWN_DIRNAME)
|
||||
@@ -0,0 +1,268 @@
|
||||
"""[goal265 step4 — RED] Dual-reader config contract for the dedicated
|
||||
install flags ``allow_git_url_install`` / ``allow_pip_install``.
|
||||
|
||||
TARGET CONTRACT (NOT YET IMPLEMENTED — goal265-spec.md §3, LOCKED):
|
||||
|
||||
- Keys: ``allow_git_url_install``, ``allow_pip_install`` in
|
||||
config.ini ``[default]``.
|
||||
- BOTH readers carry the keys (read dict + write_config list +
|
||||
exception-fallback dict — 3 anchors each):
|
||||
* ``comfyui_manager/glob/manager_core.py``
|
||||
* ``comfyui_manager/legacy/manager_core.py``
|
||||
(dual-reader rule — the gated endpoints resolve through the LEGACY
|
||||
reader; a glob-only registration would silently never reach the gates.)
|
||||
- Only case-insensitive string "true" is truthy on read; anything else
|
||||
(including "1", "yes") reads False.
|
||||
- Missing key reads False — the shared ``get_bool(key, default)`` quirk:
|
||||
the ``default`` param is IGNORED, missing key -> always False.
|
||||
- Missing file / missing [default] section -> exception-fallback dict
|
||||
returns False for both keys.
|
||||
- Activation is restart-only (module-level ``cached_config``).
|
||||
|
||||
These tests are EXPECTED TO FAIL against current code (the keys are not
|
||||
registered in either reader yet) — RED confirmation is goal265 Step 5.
|
||||
|
||||
SC rows covered (goal265-scenarios.md):
|
||||
SC-19 missing keys -> False (both readers)
|
||||
SC-20 malformed values -> only "true"/"TRUE"/"True" truthy (both readers)
|
||||
SC-21 write_config round-trip persistence (both readers)
|
||||
SC-22 cached_config staleness — restart-only activation (reader-level arm)
|
||||
SC-29 no config.ini / no [default] section -> fallback False (both readers)
|
||||
|
||||
Fixtures (spec §4 binding): tmp config.ini + monkeypatch
|
||||
``context.manager_config_path`` + per-reader ``cached_config`` reset.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _install_flags_testutil import import_context, import_reader # noqa: E402
|
||||
|
||||
FLAG_KEYS = ("allow_git_url_install", "allow_pip_install")
|
||||
READERS = ("glob", "legacy")
|
||||
|
||||
|
||||
def _reset_cache(core) -> None:
|
||||
"""Clear the reader's module-level cached_config (simulates restart)."""
|
||||
setattr(core, "cached_config", None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture(params=READERS)
|
||||
def reader_core(request):
|
||||
"""Yield (reader_name, manager_core module) for BOTH readers, with the
|
||||
module-level cached_config saved/cleared around each test so one test's
|
||||
cache never leaks into another (or into other test modules)."""
|
||||
core = import_reader(request.param)
|
||||
saved = getattr(core, "cached_config", None)
|
||||
_reset_cache(core)
|
||||
yield request.param, core
|
||||
setattr(core, "cached_config", saved)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def point_config(monkeypatch):
|
||||
"""Return a function that points context.manager_config_path at a path.
|
||||
|
||||
Both readers resolve ``context.manager_config_path`` at CALL time
|
||||
(``config.read(context.manager_config_path)``), so monkeypatching the
|
||||
context attribute redirects glob AND legacy alike."""
|
||||
ctx = import_context()
|
||||
|
||||
def _point(path):
|
||||
monkeypatch.setattr(ctx, "manager_config_path", str(path))
|
||||
|
||||
return _point
|
||||
|
||||
|
||||
def _write_ini(tmp_path, body: str):
|
||||
ini = tmp_path / "config.ini"
|
||||
ini.write_text(body, encoding="utf-8")
|
||||
return ini
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-19 — missing keys read False (get_bool quirk: missing key -> False)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc19_missing_keys_read_false(reader_core, point_config, tmp_path):
|
||||
"""SC-19: config.ini WITHOUT either new key -> read_config returns
|
||||
allow_git_url_install == False AND allow_pip_install == False.
|
||||
|
||||
This is the observable manifestation of the shared ``get_bool(key,
|
||||
default)`` quirk (spec §3): the ``default`` parameter is IGNORED —
|
||||
a missing key always reads False, never the passed default. The quirk
|
||||
is documented, NOT fixed, by this GOAL (spec §5 freeze item 9)."""
|
||||
reader, core = reader_core
|
||||
point_config(_write_ini(tmp_path, "[default]\nsecurity_level = normal\n"))
|
||||
|
||||
cfg = core.read_config()
|
||||
|
||||
for key in FLAG_KEYS:
|
||||
assert key in cfg, (
|
||||
f"RED: {reader} read_config() does not register {key!r} "
|
||||
"(goal265-spec.md §3 reader registration)"
|
||||
)
|
||||
assert cfg[key] is False, f"{reader}: missing {key} must read False"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-20 — malformed values: only case-insensitive "true" is truthy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raw_value", "expected"),
|
||||
[
|
||||
("true", True),
|
||||
("TRUE", True),
|
||||
("True", True),
|
||||
("1", False),
|
||||
("yes", False),
|
||||
("false", False),
|
||||
("garbage", False),
|
||||
],
|
||||
ids=["true", "TRUE", "True", "1", "yes", "false", "garbage"],
|
||||
)
|
||||
@pytest.mark.parametrize("flag_key", FLAG_KEYS)
|
||||
def test_sc20_only_case_insensitive_true_is_truthy(
|
||||
reader_core, point_config, tmp_path, flag_key, raw_value, expected):
|
||||
"""SC-20: key present with malformed values -> only case-insensitive
|
||||
"true" reads True; "1"/"yes"/"false"/"garbage" read False (both flags,
|
||||
both readers)."""
|
||||
reader, core = reader_core
|
||||
point_config(_write_ini(
|
||||
tmp_path,
|
||||
f"[default]\nsecurity_level = normal\n{flag_key} = {raw_value}\n",
|
||||
))
|
||||
|
||||
cfg = core.read_config()
|
||||
|
||||
assert flag_key in cfg, (
|
||||
f"RED: {reader} read_config() does not register {flag_key!r}"
|
||||
)
|
||||
assert cfg[flag_key] is expected, (
|
||||
f"{reader}: {flag_key} = {raw_value!r} must read {expected}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-21 — write_config round-trip (write list registration, both writers)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc21_write_config_round_trips_both_flags(
|
||||
reader_core, point_config, tmp_path):
|
||||
"""SC-21: both flags true in cached config -> write_config -> reset the
|
||||
reader's cached_config -> read_config -> both keys persist as True.
|
||||
|
||||
Failure mode this guards: keys registered in the read dict but NOT in
|
||||
the write_config persistence list — the value would silently drop on
|
||||
the next config save (spec §2 F3/F4 dual-anchor rule, residual risk R2)."""
|
||||
reader, core = reader_core
|
||||
ini = _write_ini(
|
||||
tmp_path,
|
||||
"[default]\nsecurity_level = normal\n"
|
||||
"allow_git_url_install = true\nallow_pip_install = true\n",
|
||||
)
|
||||
point_config(ini)
|
||||
|
||||
loaded = core.get_config()
|
||||
for key in FLAG_KEYS:
|
||||
assert loaded.get(key) is True, (
|
||||
f"RED: {reader} get_config() did not load {key!r}=True "
|
||||
"from staged config.ini"
|
||||
)
|
||||
|
||||
core.write_config()
|
||||
|
||||
raw = ini.read_text(encoding="utf-8")
|
||||
for key in FLAG_KEYS:
|
||||
assert key in raw, (
|
||||
f"{reader}: write_config() dropped {key!r} — key missing from "
|
||||
"the write_config persistence list (spec §3 reader registration)"
|
||||
)
|
||||
|
||||
_reset_cache(core) # per-reader cache reset REQUIRED before re-read
|
||||
cfg = core.read_config()
|
||||
for key in FLAG_KEYS:
|
||||
assert cfg.get(key) is True, (
|
||||
f"{reader}: {key} did not survive the write/read round-trip"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-22 — cached_config staleness: restart-only activation (reader arm)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("flag_key", FLAG_KEYS)
|
||||
def test_sc22_flag_edit_without_restart_stays_stale(
|
||||
reader_core, point_config, tmp_path, flag_key):
|
||||
"""SC-22: process running with cached flag=false -> config.ini edited to
|
||||
flag=true WITHOUT restart -> the reader still serves False (module-level
|
||||
cached_config; identical semantics to security_level today, MM §1.5).
|
||||
Clearing the cache (= restart) then serves True."""
|
||||
reader, core = reader_core
|
||||
ini = _write_ini(
|
||||
tmp_path,
|
||||
f"[default]\nsecurity_level = normal\n{flag_key} = false\n",
|
||||
)
|
||||
point_config(ini)
|
||||
|
||||
first = core.get_config()
|
||||
assert first.get(flag_key) is False, (
|
||||
f"RED: {reader} get_config() does not register {flag_key!r}"
|
||||
)
|
||||
|
||||
# Edit config.ini on disk — NO cache reset (no restart).
|
||||
_write_ini(
|
||||
tmp_path,
|
||||
f"[default]\nsecurity_level = normal\n{flag_key} = true\n",
|
||||
)
|
||||
|
||||
stale = core.get_config()
|
||||
assert stale.get(flag_key) is False, (
|
||||
f"{reader}: {flag_key} hot-reloaded — activation MUST be "
|
||||
"restart-only (cached_config, spec §3)"
|
||||
)
|
||||
|
||||
_reset_cache(core) # simulate restart
|
||||
fresh = core.get_config()
|
||||
assert fresh.get(flag_key) is True, (
|
||||
f"{reader}: {flag_key}=true not visible after cache reset (restart)"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-29 — exception-fallback path carries the new keys as False
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.parametrize("breakage", ["missing_file", "no_default_section"])
|
||||
def test_sc29_fallback_path_returns_false_for_both_flags(
|
||||
reader_core, point_config, tmp_path, breakage):
|
||||
"""SC-29: NO config.ini present / config.ini WITHOUT [default] section ->
|
||||
read_config's exception-fallback dict returns False for BOTH new keys
|
||||
(fallback dicts are the third registration anchor per reader,
|
||||
spec §2 F3/F4)."""
|
||||
reader, core = reader_core
|
||||
if breakage == "missing_file":
|
||||
point_config(tmp_path / "does-not-exist" / "config.ini")
|
||||
else:
|
||||
point_config(_write_ini(tmp_path, "[other]\nsomething = 1\n"))
|
||||
|
||||
cfg = core.read_config()
|
||||
|
||||
for key in FLAG_KEYS:
|
||||
assert key in cfg, (
|
||||
f"RED: {reader} exception-fallback dict does not carry {key!r} "
|
||||
"(goal265-spec.md §3 'Reader registration' — fallback anchor)"
|
||||
)
|
||||
assert cfg[key] is False, (
|
||||
f"{reader}: fallback {key} must be False (secure by default)"
|
||||
)
|
||||
@@ -0,0 +1,256 @@
|
||||
"""[goal265 step4] Out-of-scope GUARD tests + structural frontend-copy
|
||||
assertion for the dedicated install flags GOAL.
|
||||
|
||||
Two populations in this module (intentionally different RED/GREEN states):
|
||||
|
||||
1. GUARD rows (SC-25C, SC-26, SC-27, SC-28) — regression-intent '=' rows:
|
||||
they assert the out-of-scope FREEZE (goal265-spec.md §5) and are expected
|
||||
to PASS both TODAY and AFTER Step 6. A failure at any point means the
|
||||
implementation leaked outside the locked scope.
|
||||
|
||||
2. SC-23 frontend arm — Δ row, EXPECTED TO FAIL TODAY (RED): the
|
||||
js/common.js 403 error copy must name the responsible flag after Step 6
|
||||
(spec §1.4, Q2 in scope). Do NOT weaken it to pass against today's code.
|
||||
|
||||
Placement per spec §4 rows 4-5 ('existing glob suite extension or unit' /
|
||||
'structural — copy-string assertion'): authored as a NEW unit/structural
|
||||
module so the pre-existing suite stays untouched (Step-4 acceptance bar (d));
|
||||
layout follows the Step-2 hand-off recommendation
|
||||
(goal265-scenarios.md §6: test_install_flags_guards.py).
|
||||
|
||||
Functional rows import the GLOB copy of the gate matrix
|
||||
(comfyui_manager/glob/utils/security_utils.py) with the lightweight runtime
|
||||
stubs from tests/_install_flags_testutil.py; the legacy copy of the same
|
||||
matrix is exercised end-to-end by tests/e2e/test_e2e_secgate_legacy_flags.py
|
||||
(SC-25A/B) under a real server.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _install_flags_testutil import ( # noqa: E402
|
||||
REPO_ROOT,
|
||||
import_glob_security_utils,
|
||||
)
|
||||
|
||||
GLOB_SERVER_PATH = os.path.join(
|
||||
REPO_ROOT, "comfyui_manager", "glob", "manager_server.py")
|
||||
CM_CLI_PATH = os.path.join(REPO_ROOT, "cm_cli", "__main__.py")
|
||||
COMMON_JS_PATH = os.path.join(REPO_ROOT, "comfyui_manager", "js", "common.js")
|
||||
|
||||
FLAG_KEYS = ("allow_git_url_install", "allow_pip_install")
|
||||
|
||||
|
||||
def _read(path: str) -> str:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
def _extract_js_function(source: str, name: str) -> str:
|
||||
"""Slice an `export async function <name>` block out of common.js
|
||||
(up to the next exported/function declaration or EOF)."""
|
||||
m = re.search(
|
||||
rf"export\s+async\s+function\s+{re.escape(name)}\b.*?"
|
||||
r"(?=\nexport\s|\nfunction\s|\Z)",
|
||||
source,
|
||||
re.DOTALL,
|
||||
)
|
||||
assert m, f"could not locate function {name!r} in js/common.js"
|
||||
return m.group(0)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-27 — glob route-table absence (guard, '=')
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc27_glob_registers_no_git_url_or_pip_install_route():
|
||||
"""SC-27 (guard): glob/manager_server.py must register NO
|
||||
/v2/customnode/install/git_url or /v2/customnode/install/pip route —
|
||||
no new glob HTTP surface under this GOAL (Q4 settled; spec §5 item 5).
|
||||
Structural: route-decorator scan of the glob server source."""
|
||||
src = _read(GLOB_SERVER_PATH)
|
||||
for path in ("/v2/customnode/install/git_url",
|
||||
"/v2/customnode/install/pip"):
|
||||
pattern = rf"routes\.(post|get|put|delete)\(\s*[\"']{re.escape(path)}[\"']"
|
||||
assert not re.search(pattern, src), (
|
||||
f"SC-27: glob manager_server registers {path} — Q4 forbids a "
|
||||
"new glob surface for these capabilities under this GOAL"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-26 — cm-cli path stays ungated (guard, '=')
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc26_cm_cli_install_path_has_no_gate():
|
||||
"""SC-26 (guard): the cm-cli install path (install_node ->
|
||||
core.gitclone_install, cm_cli/__main__.py) is a local operator tool —
|
||||
it must consult NEITHER the security_level gate NOR the new flags
|
||||
(spec §5 item 4: cm-cli untouched). Static source assertion — no clone
|
||||
is executed."""
|
||||
src = _read(CM_CLI_PATH)
|
||||
assert "is_allowed_security_level" not in src, (
|
||||
"SC-26: cm_cli/__main__.py grew a security_level gate — cm-cli "
|
||||
"must stay ungated under this GOAL"
|
||||
)
|
||||
for key in FLAG_KEYS:
|
||||
assert key not in src, (
|
||||
f"SC-26: cm_cli/__main__.py consults {key!r} — the dedicated "
|
||||
"flags must not gate the local operator CLI"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-25C — glob comfyui_switch_version stays high+ security_level-coupled
|
||||
# (guard, '=') — structural arm
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_sc25c_glob_switch_version_keeps_high_plus_gate():
|
||||
"""SC-25C (guard, structural arm): the glob comfyui_switch_version
|
||||
handler keeps its is_allowed_security_level('high+') guard and does not
|
||||
consult the new flags (spec §5 item 1). The functional arm (high+ denies
|
||||
at sl=normal regardless of flags) is test_sc25c_sc28_* below."""
|
||||
src = _read(GLOB_SERVER_PATH)
|
||||
m = re.search(
|
||||
r"@routes\.post\(\s*[\"']/v2/comfyui_manager/comfyui_switch_version[\"']\s*\)"
|
||||
r".*?(?=\n@routes\.|\Z)",
|
||||
src,
|
||||
re.DOTALL,
|
||||
)
|
||||
assert m, "SC-25C: comfyui_switch_version route not found in glob server"
|
||||
handler = m.group(0)
|
||||
assert re.search(
|
||||
r"is_allowed_security_level\(\s*[\"']high\+[\"']", handler), (
|
||||
"SC-25C: comfyui_switch_version no longer carries the "
|
||||
"is_allowed_security_level('high+') guard — out-of-scope gate "
|
||||
"changed (spec §5 freeze item 1)"
|
||||
)
|
||||
for key in FLAG_KEYS:
|
||||
assert key not in handler, (
|
||||
f"SC-25C: comfyui_switch_version consults {key!r} — the new "
|
||||
"flags must not affect this surface"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-28 (+ SC-25C functional arm) — security_level matrix untouched
|
||||
# (guard, '=')
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Frozen truth table of is_allowed_security_level at loopback listen,
|
||||
# network_mode=public (verified against legacy/manager_server.py:97-122 and
|
||||
# glob/utils/security_utils.py:22-48 as of base 01799f8c). The new flags
|
||||
# must leave every cell unchanged.
|
||||
_EXPECTED_MATRIX = {
|
||||
# (level, security_level): allowed
|
||||
("block", "weak"): False,
|
||||
("block", "normal"): False,
|
||||
("high+", "weak"): True,
|
||||
("high+", "normal-"): True,
|
||||
("high+", "normal"): False,
|
||||
("high+", "strong"): False,
|
||||
("high", "weak"): True,
|
||||
("high", "normal-"): True,
|
||||
("high", "normal"): False,
|
||||
("high", "strong"): False,
|
||||
("middle+", "weak"): True,
|
||||
("middle+", "normal-"): True,
|
||||
("middle+", "normal"): True,
|
||||
("middle+", "strong"): False,
|
||||
("middle", "weak"): True,
|
||||
("middle", "normal-"): True,
|
||||
("middle", "normal"): True,
|
||||
("middle", "strong"): False,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("flags_value", [True, False],
|
||||
ids=["flags-on", "flags-off"])
|
||||
def test_sc25c_sc28_security_level_matrix_unchanged_by_flags(
|
||||
monkeypatch, flags_value):
|
||||
"""SC-28 (guard): the is_allowed_security_level truth table for the
|
||||
non-target levels (middle/middle+/high/high+/block) is byte-identical
|
||||
whether both new flags are true or false — security_level semantics
|
||||
untouched (spec §5 item 2; MM §3).
|
||||
|
||||
SC-25C (functional arm): the ('high+', 'normal') -> False cell is the
|
||||
switch_version deny — it must hold for BOTH flag combinations.
|
||||
|
||||
Exercises the GLOB copy of the matrix (importable without a ComfyUI
|
||||
runtime); config access is monkeypatched at the consumer module so no
|
||||
real config.ini is read."""
|
||||
su = import_glob_security_utils()
|
||||
|
||||
fake_config = {
|
||||
"network_mode": "public",
|
||||
"allow_git_url_install": flags_value,
|
||||
"allow_pip_install": flags_value,
|
||||
}
|
||||
current_level = {"value": "normal"}
|
||||
|
||||
def fake_get_config():
|
||||
return {**fake_config, "security_level": current_level["value"]}
|
||||
|
||||
monkeypatch.setattr(su.core, "get_config", fake_get_config)
|
||||
monkeypatch.setattr(su.args, "listen", "127.0.0.1")
|
||||
|
||||
observed = {}
|
||||
for (level, sl) in _EXPECTED_MATRIX:
|
||||
current_level["value"] = sl
|
||||
observed[(level, sl)] = su.is_allowed_security_level(level)
|
||||
|
||||
mismatches = {
|
||||
cell: (got, _EXPECTED_MATRIX[cell])
|
||||
for cell, got in observed.items()
|
||||
if got is not _EXPECTED_MATRIX[cell]
|
||||
}
|
||||
assert not mismatches, (
|
||||
f"SC-28(flags={flags_value}): security_level matrix drifted — "
|
||||
f"{{cell: (got, expected)}} = {mismatches!r}. The dedicated-flag "
|
||||
"GOAL must not alter is_allowed_security_level semantics."
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# SC-23 frontend arm — js/common.js error copy names the flag (Δ — RED today)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestSc23FrontendCopy:
|
||||
"""SC-23 (frontend arm, structural — spec §1.4 / Q2 in scope): on a
|
||||
non-200 install response, the user-visible copy must name the
|
||||
responsible flag, replacing the 'security level' framing.
|
||||
Copy-string assertion on js/common.js — not an executed-browser test
|
||||
(goal265-scenarios.md §6 note). EXPECTED TO FAIL TODAY (RED)."""
|
||||
|
||||
def test_sc23_install_pip_error_copy_names_pip_flag(self):
|
||||
"""install_pip error branch (~:213 call site) names
|
||||
allow_pip_install and drops the security-level framing."""
|
||||
block = _extract_js_function(_read(COMMON_JS_PATH), "install_pip")
|
||||
assert "allow_pip_install" in block, (
|
||||
"SC-23: js/common.js install_pip error copy does not name "
|
||||
"allow_pip_install (spec §1.4 frontend copy contract)"
|
||||
)
|
||||
assert "security level" not in block.lower(), (
|
||||
"SC-23: js/common.js install_pip still carries the misleading "
|
||||
"'security level' framing"
|
||||
)
|
||||
|
||||
def test_sc23_install_git_url_error_copy_names_git_flag(self):
|
||||
"""install_via_git_url error branch (~:248 call site) names
|
||||
allow_git_url_install and drops the security-level framing."""
|
||||
block = _extract_js_function(
|
||||
_read(COMMON_JS_PATH), "install_via_git_url")
|
||||
assert "allow_git_url_install" in block, (
|
||||
"SC-23: js/common.js install_via_git_url error copy does not "
|
||||
"name allow_git_url_install (spec §1.4 frontend copy contract)"
|
||||
)
|
||||
assert "security level" not in block.lower(), (
|
||||
"SC-23: js/common.js install_via_git_url still carries the "
|
||||
"misleading 'security level' framing"
|
||||
)
|
||||
@@ -0,0 +1,340 @@
|
||||
"""[GOAL #347] Non-e2e unit regression guards for the OTHER legacy
|
||||
``security_level`` gate paths in ``comfyui_manager/legacy/manager_server.py``.
|
||||
|
||||
CONTEXT (GOAL #346 audit): the dedicated-install-flag work (git_url / pip
|
||||
endpoints) gained unit + e2e coverage, but the OTHER legacy gate sites — the
|
||||
ones that still gate on ``is_allowed_security_level(level)`` — had NO non-e2e
|
||||
coverage; only ``tests/e2e/`` exercised them, and only behind a real-server
|
||||
harness (E2E_ROOT). These guards close that gap WITHOUT a real server and
|
||||
WITHOUT modifying production code.
|
||||
|
||||
PATHS GUARDED (the OTHER security_level gates):
|
||||
P1 middle+ batch-install entry gate — ``_install_custom_node`` (L1441)
|
||||
P2 unknown-pip 'block' unconditional — ``_install_custom_node`` risky
|
||||
deny branch (``is_allowed_security_level
|
||||
('block')`` is hard-False; Q1)
|
||||
P3 nodepack/snapshot high+/middle — ``comfyui_switch_version`` (high+),
|
||||
gates ``restore_snapshot`` (middle+)
|
||||
P4 model non-.safetensors high+ gate — ``_install_model`` (L1713)
|
||||
|
||||
NON-VACUITY (the bar from the GOAL MM): every assertion below is written so
|
||||
that WEAKENING the underlying gate would FLIP the test result —
|
||||
* the decision-table tests drive the REAL ``is_allowed_security_level`` and
|
||||
pin the exact allow/deny truth for the level each path consumes; changing
|
||||
that function's logic for any level flips a row;
|
||||
* the response-shape tests invoke the REAL async handlers and assert the
|
||||
observable status (403/404/200) — each pairs a DENY case with an opposite
|
||||
(allow / flip) case proving the gate, not a constant, is what decides.
|
||||
A guard that merely imports or asserts a constant is explicitly avoided.
|
||||
|
||||
ISOLATION: reuses the ``tests/_install_flags_testutil.py`` stub approach
|
||||
(``ensure_comfy_stubs`` + ``_purge_fake_comfyui_manager``) and adds the
|
||||
minimal extra stubs ``manager_server`` needs at import (``server.PromptServer``
|
||||
route table, ``nodes``, ``folder_paths.__file__`` / ``get_folder_paths``), plus
|
||||
an inert ``manager_util.get_data`` so the module-level cache-warmup thread does
|
||||
no network I/O. No production code is touched.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _install_flags_testutil import ( # noqa: E402
|
||||
ensure_comfy_stubs,
|
||||
_purge_fake_comfyui_manager,
|
||||
)
|
||||
|
||||
_LEGACY_SERVER = None
|
||||
|
||||
|
||||
def _import_legacy_server():
|
||||
"""Import ``comfyui_manager.legacy.manager_server`` under minimal stubs
|
||||
(idempotent; memoized). Returns the real module object so the tests drive
|
||||
the REAL gate function and REAL handlers — only the runtime *environment*
|
||||
is stubbed, never the gate logic."""
|
||||
global _LEGACY_SERVER
|
||||
if _LEGACY_SERVER is not None:
|
||||
return _LEGACY_SERVER
|
||||
|
||||
ensure_comfy_stubs()
|
||||
_purge_fake_comfyui_manager()
|
||||
|
||||
# folder_paths: manager_server reads ``__file__`` and ``get_folder_paths``
|
||||
# at import (context.comfy_path derivation + check_invalid_nodes).
|
||||
fp = sys.modules["folder_paths"]
|
||||
if not hasattr(fp, "__file__"):
|
||||
fp.__file__ = os.path.join(tempfile.mkdtemp(prefix="cm-fp-"), "folder_paths.py")
|
||||
if not hasattr(fp, "get_folder_paths"):
|
||||
fp.get_folder_paths = lambda *a, **k: []
|
||||
|
||||
# nodes: imported but only used inside handlers we don't reach.
|
||||
sys.modules.setdefault("nodes", types.ModuleType("nodes"))
|
||||
|
||||
# server.PromptServer.instance.routes: the @routes.post/.get decorators run
|
||||
# at import. An identity-decorator route table lets every handler register
|
||||
# as a plain module-level function without a real aiohttp app.
|
||||
if "server" not in sys.modules:
|
||||
server = types.ModuleType("server")
|
||||
|
||||
class _Routes:
|
||||
def _identity(self, *a, **k):
|
||||
def deco(fn):
|
||||
return fn
|
||||
return deco
|
||||
|
||||
post = get = put = delete = _identity
|
||||
|
||||
class _Inst:
|
||||
routes = _Routes()
|
||||
|
||||
class PromptServer:
|
||||
instance = _Inst()
|
||||
|
||||
server.PromptServer = PromptServer
|
||||
sys.modules["server"] = server
|
||||
|
||||
# Make the module-level cache-warmup thread (manager_server L~1990
|
||||
# ``threading.Thread(... default_cache_update ...)``) inert — no network.
|
||||
import importlib
|
||||
mu = importlib.import_module("comfyui_manager.common.manager_util")
|
||||
|
||||
async def _inert_get_data(uri, *a, **k):
|
||||
return {"custom_nodes": [], "models": [], "node_map": {}}
|
||||
|
||||
mu.get_data = _inert_get_data
|
||||
|
||||
_LEGACY_SERVER = importlib.import_module("comfyui_manager.legacy.manager_server")
|
||||
return _LEGACY_SERVER
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def ms():
|
||||
return _import_legacy_server()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gate(ms, monkeypatch):
|
||||
"""Return a function ``set_gate(security_level, network_mode='public',
|
||||
is_local_mode=True)`` that drives the gate inputs.
|
||||
|
||||
Builds a COMPLETE config dict off the reader's real fallback so any
|
||||
concurrent background reader (the warmup thread) never KeyErrors, then
|
||||
overrides only ``security_level`` / ``network_mode``. ``is_local_mode`` is
|
||||
the module global the gate consults. monkeypatch auto-restores."""
|
||||
base = dict(ms.core.read_config())
|
||||
|
||||
def set_gate(security_level, network_mode="public", is_local_mode=True):
|
||||
cfg = dict(base)
|
||||
cfg["security_level"] = security_level
|
||||
cfg["network_mode"] = network_mode
|
||||
monkeypatch.setattr(ms.core, "get_config", lambda: dict(cfg))
|
||||
monkeypatch.setattr(ms, "is_local_mode", is_local_mode)
|
||||
|
||||
return set_gate
|
||||
|
||||
|
||||
def _run(coro):
|
||||
return asyncio.run(coro)
|
||||
|
||||
|
||||
class FakeRequest:
|
||||
"""Minimal stand-in for an aiohttp request. ``content_type`` satisfies
|
||||
``reject_simple_form_post``; nothing else is read on the deny path (the
|
||||
gate returns before any body access)."""
|
||||
|
||||
def __init__(self, content_type="application/json"):
|
||||
self.content_type = content_type
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Decision-table guards — drive the REAL is_allowed_security_level for the
|
||||
# exact level each OTHER path consumes. Non-vacuous: weakening the function
|
||||
# for any level flips a row.
|
||||
# ===========================================================================
|
||||
|
||||
# (security_level, is_local_mode, network_mode, expected_allowed)
|
||||
_MIDDLE_PLUS = [ # P1 batch entry, _update_all, _install_model entry, restore_snapshot
|
||||
("weak", True, "public", True),
|
||||
("normal", True, "public", True),
|
||||
("normal-", True, "public", True),
|
||||
("strong", True, "public", False),
|
||||
("weak", False, "public", False), # public non-loopback -> denied
|
||||
("normal", False, "public", False),
|
||||
("weak", False, "personal_cloud", True), # personal_cloud arm of middle+
|
||||
("strong", False, "personal_cloud", False),
|
||||
]
|
||||
|
||||
_HIGH_PLUS = [ # P3 comfyui_switch_version, P4 model non-.safetensors sub-check
|
||||
("weak", True, "public", True),
|
||||
("normal-", True, "public", True),
|
||||
("normal", True, "public", False),
|
||||
("strong", True, "public", False),
|
||||
("weak", False, "public", False),
|
||||
("weak", False, "personal_cloud", True),
|
||||
("normal-", False, "personal_cloud", False), # personal_cloud high+ needs weak
|
||||
]
|
||||
|
||||
_MIDDLE = [ # P3 remove_snapshot and the other middle-gated legacy ops
|
||||
("weak", True, "public", True),
|
||||
("normal", True, "public", True),
|
||||
("normal-", True, "public", True),
|
||||
("strong", True, "public", False),
|
||||
("normal", False, "public", True), # 'middle' ignores locality
|
||||
("strong", False, "public", False),
|
||||
]
|
||||
|
||||
_BLOCK_LEVELS = ["weak", "normal", "normal-", "strong"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("sl", "local", "nm", "expected"), _MIDDLE_PLUS)
|
||||
def test_p1_middle_plus_decision(ms, gate, sl, local, nm, expected):
|
||||
"""P1: the middle+ batch-install entry gate decision. is_allowed_security_level
|
||||
('middle+') must allow iff sl in {weak,normal,normal-} AND
|
||||
(loopback OR personal_cloud)."""
|
||||
gate(sl, network_mode=nm, is_local_mode=local)
|
||||
assert ms.is_allowed_security_level("middle+") is expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize("sl", _BLOCK_LEVELS)
|
||||
@pytest.mark.parametrize("local", [True, False], ids=["loopback", "public"])
|
||||
@pytest.mark.parametrize("nm", ["public", "personal_cloud"])
|
||||
def test_p2_block_is_unconditional_deny(ms, gate, sl, local, nm):
|
||||
"""P2: unknown-pip 'block' is an UNCONDITIONAL deny — is_allowed_security_level
|
||||
('block') is False for EVERY security_level, locality, and network_mode
|
||||
(Q1: the dedicated flags do not open it). If 'block' ever became
|
||||
allowable, every row here flips."""
|
||||
gate(sl, network_mode=nm, is_local_mode=local)
|
||||
assert ms.is_allowed_security_level("block") is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("sl", "local", "nm", "expected"), _HIGH_PLUS)
|
||||
def test_p3_high_plus_decision(ms, gate, sl, local, nm, expected):
|
||||
"""P3 (high+ arm): comfyui_switch_version gate decision. Allow iff
|
||||
(loopback AND sl in {weak,normal-}) OR (personal_cloud AND sl == weak)."""
|
||||
gate(sl, network_mode=nm, is_local_mode=local)
|
||||
assert ms.is_allowed_security_level("high+") is expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("sl", "local", "nm", "expected"), _MIDDLE)
|
||||
def test_p3_middle_decision(ms, gate, sl, local, nm, expected):
|
||||
"""P3 (middle arm): snapshot remove / middle-gated legacy ops. Allow iff
|
||||
sl in {weak,normal,normal-} (locality-independent)."""
|
||||
gate(sl, network_mode=nm, is_local_mode=local)
|
||||
assert ms.is_allowed_security_level("middle") is expected
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Response-shape guards — invoke the REAL async handlers; assert the observable
|
||||
# status. Each pairs a DENY with an opposite case so the gate (not a constant)
|
||||
# is provably what decides.
|
||||
# ===========================================================================
|
||||
|
||||
def test_p1_install_custom_node_denies_at_strong_403(ms, gate):
|
||||
"""P1 deny: at sl=strong the middle+ entry gate (first statement of
|
||||
_install_custom_node) returns 403 before any further processing."""
|
||||
gate("strong", is_local_mode=True)
|
||||
resp = _run(ms._install_custom_node({}))
|
||||
assert resp.status == 403
|
||||
|
||||
|
||||
def test_p1_install_custom_node_passes_entry_gate_at_normal(ms, gate, monkeypatch):
|
||||
"""P1 allow-flip: at sl=normal the middle+ entry gate PASSES — with the
|
||||
downstream risky level forced to an allowed value the handler reaches 200,
|
||||
proving the 403 above came from the gate, not unconditionally."""
|
||||
gate("normal", is_local_mode=True)
|
||||
|
||||
async def _risky_middle_plus(files, pip):
|
||||
return "middle+"
|
||||
|
||||
monkeypatch.setattr(ms, "get_risky_level", _risky_middle_plus)
|
||||
json_data = {
|
||||
"version": "unknown",
|
||||
"files": ["https://github.com/example/unknown-node.git"],
|
||||
"pip": [],
|
||||
"id": "x",
|
||||
"ui_id": "u",
|
||||
"channel": "default",
|
||||
"mode": "cache",
|
||||
}
|
||||
resp = _run(ms._install_custom_node(dict(json_data)))
|
||||
assert resp.status == 200
|
||||
|
||||
|
||||
def test_p2_unknown_pip_block_denies_404_even_at_weak(ms, gate, monkeypatch):
|
||||
"""P2: with the risky level resolved to 'block' (unknown pip), the handler
|
||||
returns 404 even at sl=weak (the most permissive level) — proving 'block'
|
||||
is an unconditional deny, not a security_level-relative one."""
|
||||
gate("weak", is_local_mode=True)
|
||||
|
||||
async def _risky_block(files, pip):
|
||||
return "block"
|
||||
|
||||
monkeypatch.setattr(ms, "get_risky_level", _risky_block)
|
||||
json_data = {
|
||||
"version": "unknown",
|
||||
"files": ["https://github.com/example/unknown-node.git"],
|
||||
"pip": ["some-unknown-pkg"],
|
||||
"id": "x",
|
||||
"ui_id": "u",
|
||||
"channel": "default",
|
||||
"mode": "cache",
|
||||
}
|
||||
resp = _run(ms._install_custom_node(dict(json_data)))
|
||||
assert resp.status == 404
|
||||
|
||||
|
||||
def test_p3_comfyui_switch_version_denies_high_plus_403(ms, gate):
|
||||
"""P3 high+ handler: comfyui_switch_version denies with 403 at sl=normal
|
||||
(high+ requires normal- or lower on loopback)."""
|
||||
gate("normal", is_local_mode=True)
|
||||
resp = _run(ms.comfyui_switch_version(FakeRequest()))
|
||||
assert resp.status == 403
|
||||
|
||||
|
||||
def test_p3_restore_snapshot_denies_middle_plus_403(ms, gate):
|
||||
"""P3 middle+ handler: restore_snapshot denies with 403 at sl=strong
|
||||
(middle+ requires normal or lower)."""
|
||||
gate("strong", is_local_mode=True)
|
||||
resp = _run(ms.restore_snapshot(FakeRequest()))
|
||||
assert resp.status == 403
|
||||
|
||||
|
||||
def test_p4_install_model_non_safetensors_denies_high_plus_403(ms, gate, monkeypatch):
|
||||
"""P4 deny: a non-.safetensors model not on the whitelist requires high+;
|
||||
at sl=normal (middle+ passes, high+ fails) the handler returns 403."""
|
||||
gate("normal", is_local_mode=True)
|
||||
|
||||
async def _whitelist_ok(item):
|
||||
return True
|
||||
|
||||
async def _models_empty(*a, **k):
|
||||
return {"models": []}
|
||||
|
||||
monkeypatch.setattr(ms, "check_whitelist_for_model", _whitelist_ok)
|
||||
monkeypatch.setattr(ms.core, "get_data_by_mode", _models_empty)
|
||||
json_data = {"filename": "model.ckpt", "url": "http://example/model.ckpt", "ui_id": "u"}
|
||||
resp = _run(ms._install_model(dict(json_data)))
|
||||
assert resp.status == 403
|
||||
|
||||
|
||||
def test_p4_install_model_safetensors_skips_high_plus_gate_200(ms, gate, monkeypatch):
|
||||
"""P4 flip: a .safetensors model skips the non-.safetensors high+ sub-check
|
||||
entirely and reaches 200 at the SAME sl=normal — proving the 403 above is
|
||||
driven by the non-.safetensors + high+ compound, not the security level
|
||||
alone."""
|
||||
gate("normal", is_local_mode=True)
|
||||
|
||||
async def _whitelist_ok(item):
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(ms, "check_whitelist_for_model", _whitelist_ok)
|
||||
json_data = {"filename": "model.safetensors", "url": "http://example/m.safetensors", "ui_id": "u"}
|
||||
resp = _run(ms._install_model(dict(json_data)))
|
||||
assert resp.status == 200
|
||||
Reference in New Issue
Block a user