wangbo/ComfyUI-Manager
1
0
Fork 0
mirror of https://github.com/Comfy-Org/ComfyUI-Manager.git synced 2026-07-27 02:27:30 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

feat(security): dedicated install flags decouple git_url/pip from security_level

Browse Source 
Add two boolean config.ini [default] flags — allow_git_url_install and
allow_pip_install (both default false) — that fully REPLACE the
security_level term on the legacy install surfaces:

- POST /v2/customnode/install/git_url (S-A) and
  POST /v2/customnode/install/pip (S-B) are now gated solely by their
  dedicated flag AND the retained network-position invariant
  (loopback listener OR network_mode=personal_cloud). security_level
  no longer affects these two surfaces in either direction.
- The batch unknown-URL branch (S-C) routes through the same predicate;
  the unknown-pip branch stays unconditionally blocked; the general
  middle+ batch entry gate is unchanged.
- New pure predicate is_dedicated_install_allowed() in
  common/manager_security.py (config-import-free; callers pass values
  from their own reader). Both config readers (glob + legacy) register
  the keys in read/write/fallback paths.
- Denial logs and frontend copy name the responsible flag instead of
  the misleading security_level guidance. Public listeners remain
  denied regardless of the flags (no exposure widening).
- README security policy updated: config keys documented, git-url/pip
  removed from the security_level risky table, and a dedicated-flags
  subsection (REPLACE semantics, network rule, batch behavior,
  restart-only activation, weak/normal- opt-in migration note).
- Migration: existing weak/normal- users must opt in via the new flags
  (CHANGELOG note; deliberate no auto-seed).

Includes the unit/config/guard test suites (88 tests): predicate truth
table, dual-reader config contract (missing/malformed keys read false,
round-trip, cache staleness), security_level-matrix freeze guards, and
suite-order-independent test stubs.
This commit is contained in:
Dr.Lt.Data
2026-06-08 02:12:14 +09:00
parent 01799f8cac
commit 16ba874566
12 changed files with 1076 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/_install_flags_testutil.py
+177
View File
@@ -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")
tests/common/test_install_flag_predicate.py
+267
View File
@@ -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
tests/test_install_flags_config.py
+268
View File
@@ -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)"
)
tests/test_install_flags_guards.py
+256
View File
@@ -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"
)