diff --git a/tests/pytest/conftest.py b/tests/pytest/conftest.py new file mode 100644 index 00000000..a7563ec8 --- /dev/null +++ b/tests/pytest/conftest.py @@ -0,0 +1,18 @@ +"""Suite-wide test setup. + +Point the runtime's persistent and ephemeral directories at a temp location +BEFORE any test imports ``webserver.config`` (directly, or transitively via +``webserver.vpp_license_debug`` / ``webserver.plcapp_management``). ``config`` +creates ``PERSISTENT_DATA_DIR`` / ``RUNTIME_DIR`` at import time; without this +override the import would try to create ``/var/lib/openplc-runtime`` on a CI box +and the whole module would be skipped instead of tested. + +``setdefault`` so a more specific conftest (e.g. restapi) can still choose its +own paths. +""" +import os +import tempfile + +_TMP = os.path.join(tempfile.gettempdir(), "openplc-runtime-tests") +os.environ.setdefault("OPENPLC_RUNTIME_DIR", os.path.join(_TMP, "run")) +os.environ.setdefault("OPENPLC_PERSISTENT_DATA_DIR", os.path.join(_TMP, "data")) diff --git a/tests/pytest/plugins/test_vpp_license_debug.py b/tests/pytest/plugins/test_vpp_license_debug.py index 4c5d1195..72e0c579 100644 --- a/tests/pytest/plugins/test_vpp_license_debug.py +++ b/tests/pytest/plugins/test_vpp_license_debug.py @@ -572,3 +572,56 @@ def test_resolve_license_path_returns_none_for_a_prefix_sibling(tmp_path): assert lic.resolve_license_path(escaping, str(root)) is None assert lic.resolve_license_path(str(root / "plugin.json"), str(root)) == str(root / "plugin.license") + + +def test_resolve_license_path_accepts_the_persistent_dir(tmp_path, monkeypatch): + """The widened guard accepts config.VPP_DATA_DIR as a SECOND root -- the + location apply_vpp_plugin_conf relocates configs to so a license survives a + runtime update -- while still refusing anything outside both roots.""" + persist = tmp_path / "data" / "vpp" + persist.mkdir(parents=True) + monkeypatch.setattr(lic.config, "VPP_DATA_DIR", persist) + root = tmp_path / "runtime" + root.mkdir() + + # A config_path under the persistent dir resolves, even though it is NOT + # under the runtime root passed in. + assert lic.resolve_license_path(str(persist / "rpi.json"), str(root)) == str(persist / "rpi.license") + # Still fails closed for a path outside BOTH roots. + assert lic.resolve_license_path(str(tmp_path / "elsewhere" / "x.json"), str(root)) is None + + +def test_write_then_read_roundtrip_in_persistent_dir(tmp_path, monkeypatch): + """0x49/0x4A round-trip when config_path points at the persistent dir (as it + does after apply relocates it). The .license lands OUTSIDE the runtime cwd, + which is exactly what lets it survive a build/ wipe on the next update.""" + persist = tmp_path / "data" / "vpp" + persist.mkdir(parents=True) + monkeypatch.setattr(lic.config, "VPP_DATA_DIR", persist) + + cwd = tmp_path / "runtime" + cwd.mkdir() + monkeypatch.chdir(cwd) # cwd != persist on purpose + (cwd / "vpp_plugins.conf").write_text("dummy\n") + config_path = str(persist / "rpi_gpio.json") + + class _P: + name = "rpi_gpio" + + def __init__(self, cp): + self.config_path = cp + + class _Conf: + plugins = [_P(config_path)] + + monkeypatch.setattr(lic.PluginsConfiguration, "from_file", classmethod(lambda cls, _p: _Conf())) + + blob = _golden_blob() + assert lic.handle_license_command(_hex(bytes([0x49, 0x00, 0x62]) + blob)) == "49 7E" + # Landed in the persistent dir, not under cwd/build/vpp. + assert os.path.exists(persist / "rpi_gpio.license") + assert not os.path.exists(cwd / "build" / "vpp" / "rpi_gpio.license") + + read = lic.handle_license_command("4A") + assert read.startswith("4A 7E 00 62") + assert bytes(int(p, 16) for p in read.split()[4:]) == blob diff --git a/tests/pytest/plugins/test_vpp_license_delivery.py b/tests/pytest/plugins/test_vpp_license_delivery.py index a3ec7a0d..6799c10b 100644 --- a/tests/pytest/plugins/test_vpp_license_delivery.py +++ b/tests/pytest/plugins/test_vpp_license_delivery.py @@ -64,53 +64,148 @@ def test_delivery_path_matches_plugin_derivation(config_path): assert _runtime_license_dest(config_path) == _plugin_license_path(config_path) -def test_apply_vpp_plugin_conf_delivers_license(tmp_path, monkeypatch): - """Integration: a conf/.license in the upload is copied to the sibling - of the plugin's config_path; absence leaves no license (device -> demo).""" +def test_apply_vpp_plugin_conf_relocates_to_persistent_dir(tmp_path, monkeypatch): + """Integration: apply relocates config+license into PERSISTENT_DATA_DIR/vpp + (NOT build/vpp) and rewrites config_path in vpp_plugins.conf to that absolute + path, so a runtime update -- which wipes build/ -- cannot delete the license. + The .so path stays under build/vpp (it is code, rebuilt each upload).""" mgmt = pytest.importorskip( "webserver.plcapp_management", reason="runtime webserver package not importable (no venv)", ) - # Fake a single native plugin whose config_path lives under the temp cwd. + # Persistent dir lives OUTSIDE the runtime cwd on purpose -- that is the whole + # point of the change. Point the module's VPP_DATA_DIR at a temp location so + # the test never touches /var/lib. + persist = tmp_path / "data" / "vpp" + persist.mkdir(parents=True) + monkeypatch.setattr(mgmt, "VPP_DATA_DIR", persist) + cwd = tmp_path / "runtime" - (cwd).mkdir() + cwd.mkdir() monkeypatch.chdir(cwd) - config_path = str(cwd / "build" / "vpp" / "rpi_gpio.json") + monkeypatch.setattr(mgmt.build_state, "log", lambda *_a, **_k: None, raising=False) - # `path` is not decoration: apply_vpp_plugin_conf now runs the uploaded conf - # through validate_vpp_plugins_conf first, which requires every VPP plugin's - # .so to resolve inside build/vpp/. A fake without it would only prove the - # fake is out of date. - class _P: - name = "rpi_gpio" + # A REAL uploaded conf (no from_file monkeypatch): the .so path is relative + # and inside build/vpp so validate_vpp_plugins_conf accepts it; config_path is + # what the editor emits (relative build/vpp) and what apply must rewrite. + gen = tmp_path / "generated" + (gen / "conf").mkdir(parents=True) + (gen / "vpp_plugins.conf").write_text( + "rpi_gpio,./build/vpp/librpi_gpio_plugin.so,1,1,build/vpp/rpi_gpio.json,\n" + ) + (gen / "conf" / "rpi_gpio.json").write_text("{}\n") + (gen / "conf" / "rpi_gpio.license").write_bytes(b"\x4f\x50\x4c\x43" + b"\x00" * 94) # 98-byte blob - def __init__(self, cp, so): - self.config_path = cp - self.path = so + mgmt.apply_vpp_plugin_conf(str(gen)) + + # Config and license landed in the persistent dir, not build/vpp. + assert os.path.exists(persist / "rpi_gpio.json") + assert os.path.exists(persist / "rpi_gpio.license") + assert os.path.getsize(persist / "rpi_gpio.license") == 98 + assert not os.path.exists(cwd / "build" / "vpp" / "rpi_gpio.license") + + # vpp_plugins.conf was rewritten: config_path -> persistent absolute; the .so + # path is untouched (stays under build/vpp). + rewritten = mgmt.PluginsConfiguration.from_file(str(cwd / "vpp_plugins.conf")) + plugin = rewritten.plugins[0] + assert plugin.config_path == str(persist / "rpi_gpio.json") + assert plugin.path == "./build/vpp/librpi_gpio_plugin.so" - class _Conf: - plugins = [_P(config_path, "./build/vpp/librpi_gpio_plugin.so")] - monkeypatch.setattr(mgmt.PluginsConfiguration, "from_file", classmethod(lambda cls, _p: _Conf())) +def test_persistent_license_survives_an_upload_without_a_license(tmp_path, monkeypatch): + """A re-upload that does not carry a .license must NOT wipe the license the + device already holds in the persistent dir -- that survival is the point.""" + mgmt = pytest.importorskip( + "webserver.plcapp_management", + reason="runtime webserver package not importable (no venv)", + ) + persist = tmp_path / "data" / "vpp" + persist.mkdir(parents=True) + monkeypatch.setattr(mgmt, "VPP_DATA_DIR", persist) + cwd = tmp_path / "runtime" + cwd.mkdir() + monkeypatch.chdir(cwd) monkeypatch.setattr(mgmt.build_state, "log", lambda *_a, **_k: None, raising=False) - # Build the uploaded generated_dir: vpp_plugins.conf + conf/{json,license}. gen = tmp_path / "generated" (gen / "conf").mkdir(parents=True) - (gen / "vpp_plugins.conf").write_text("dummy\n") + conf_line = "rpi_gpio,./build/vpp/librpi_gpio_plugin.so,1,1,build/vpp/rpi_gpio.json,\n" + (gen / "vpp_plugins.conf").write_text(conf_line) (gen / "conf" / "rpi_gpio.json").write_text("{}\n") - (gen / "conf" / "rpi_gpio.license").write_bytes(b"\x4f\x50\x4c\x43" + b"\x00" * 94) # 98-byte blob + (gen / "conf" / "rpi_gpio.license").write_bytes(b"\x4f\x50\x4c\x43" + b"\x00" * 94) mgmt.apply_vpp_plugin_conf(str(gen)) + assert os.path.exists(persist / "rpi_gpio.license") - expected = config_path[:-5] + ".license" - assert os.path.exists(expected), "license blob not delivered to the plugin's sibling path" - assert os.path.getsize(expected) == 98 - - # Second pass without a .license in the upload must not resurrect a stale one - # from the same source (delivery only copies what the upload carries). - os.remove(expected) + # Second upload of the same VPP, this time WITHOUT the license blob. (gen / "conf" / "rpi_gpio.license").unlink() mgmt.apply_vpp_plugin_conf(str(gen)) - assert not os.path.exists(expected) + + assert os.path.exists(persist / "rpi_gpio.license"), "persistent license must survive a license-less upload" + assert os.path.getsize(persist / "rpi_gpio.license") == 98 + + +def test_migration_rescues_a_pre_change_license_from_build_vpp(tmp_path, monkeypatch): + """A device licensed before this change has its blob at build/vpp/.license. + When the update did not wipe build/ (the wipe is conditional on CMakeCache.txt), + the next upload without a bundled license migrates that blob into the persistent + dir instead of leaving it orphaned.""" + mgmt = pytest.importorskip( + "webserver.plcapp_management", + reason="runtime webserver package not importable (no venv)", + ) + persist = tmp_path / "data" / "vpp" + persist.mkdir(parents=True) + monkeypatch.setattr(mgmt, "VPP_DATA_DIR", persist) + cwd = tmp_path / "runtime" + (cwd / "build" / "vpp").mkdir(parents=True) + monkeypatch.chdir(cwd) + monkeypatch.setattr(mgmt.build_state, "log", lambda *_a, **_k: None, raising=False) + + old = cwd / "build" / "vpp" / "rpi_gpio.license" # pre-change location + old.write_bytes(b"\x4f\x50\x4c\x43" + b"\x00" * 94) + + gen = tmp_path / "generated" + (gen / "conf").mkdir(parents=True) + (gen / "vpp_plugins.conf").write_text( + "rpi_gpio,./build/vpp/librpi_gpio_plugin.so,1,1,build/vpp/rpi_gpio.json,\n" + ) + (gen / "conf" / "rpi_gpio.json").write_text("{}\n") # no .license in the upload + + mgmt.apply_vpp_plugin_conf(str(gen)) + assert (persist / "rpi_gpio.license").read_bytes() == old.read_bytes() + + +def test_migration_ignores_a_license_a_forged_config_path_points_at(tmp_path, monkeypatch): + """Security: the migration source is the FIXED build/vpp/.license, never + config_path. validate_vpp_plugins_conf only confines config_path to the runtime + root, so a forged conf could name a .license elsewhere under the root; that file + must NOT be copied to where 0x4A would read it back.""" + mgmt = pytest.importorskip( + "webserver.plcapp_management", + reason="runtime webserver package not importable (no venv)", + ) + persist = tmp_path / "data" / "vpp" + persist.mkdir(parents=True) + monkeypatch.setattr(mgmt, "VPP_DATA_DIR", persist) + cwd = tmp_path / "runtime" + (cwd / "build" / "vpp").mkdir(parents=True) + monkeypatch.chdir(cwd) + monkeypatch.setattr(mgmt.build_state, "log", lambda *_a, **_k: None, raising=False) + + # A decoy .license elsewhere in the runtime root (passes the validator's + # runtime-root confinement) that a forged config_path tries to point at. + (cwd / "secrets").mkdir() + (cwd / "secrets" / "target.license").write_bytes(b"\x4f\x50\x4c\x43" + b"\x00" * 94) + + gen = tmp_path / "generated" + (gen / "conf").mkdir(parents=True) + (gen / "vpp_plugins.conf").write_text( + "rpi_gpio,./build/vpp/librpi_gpio_plugin.so,1,1,secrets/target.json,\n" + ) + (gen / "conf" / "rpi_gpio.json").write_text("{}\n") # no .license in the upload + + mgmt.apply_vpp_plugin_conf(str(gen)) + # migration looked at build/vpp/rpi_gpio.license (absent), NOT secrets/target.license + assert not (persist / "rpi_gpio.license").exists() diff --git a/webserver/config.py b/webserver/config.py index 0abdbc24..069bde1a 100644 --- a/webserver/config.py +++ b/webserver/config.py @@ -104,6 +104,19 @@ def get_persistent_data_dir(): PERSISTENT_DATA_DIR = get_persistent_data_dir() ENV_PATH = PERSISTENT_DATA_DIR / ".env" DB_PATH = PERSISTENT_DATA_DIR / "restapi.db" +# VPP plugin configs + license blobs live here, OUTSIDE $OPENPLC_DIR/build, so a +# runtime version update (install.sh does ``rm -rf $OPENPLC_DIR/build``) can never +# delete a purchased license. The closed .so still reads them because the runtime +# writes this absolute path into vpp_plugins.conf's config_path field (see +# webserver/plcapp_management.py::apply_vpp_plugin_conf); the C loader passes +# config_path to the plugin verbatim, so only the .so binary itself must stay +# under build/vpp. +# +# Created on demand by apply_vpp_plugin_conf / _write_license_atomically, NOT at +# import: a bare module-scope mkdir is import-time filesystem work that turns a +# permission failure into a hard import crash (the very thing tests/pytest/ +# conftest.py exists to work around). +VPP_DATA_DIR = PERSISTENT_DATA_DIR / "vpp" BASE_DIR = os.path.abspath(os.path.dirname(__file__)) diff --git a/webserver/plcapp_management.py b/webserver/plcapp_management.py index 549d4bb5..ac949192 100644 --- a/webserver/plcapp_management.py +++ b/webserver/plcapp_management.py @@ -9,6 +9,7 @@ import glob from typing import Final +from webserver.config import VPP_DATA_DIR from webserver.runtimemanager import RuntimeManager from webserver.logger import get_logger, LogParser from webserver.plugin_config_model import PluginsConfiguration, PluginConfig, PluginType @@ -344,15 +345,21 @@ def apply_vpp_plugin_conf(generated_dir: str = "core/generated") -> None: * **Upload includes vpp_plugins.conf** → copy it to the runtime root so the C-side plugin loader picks it up at the next PLC start. - Also copy each plugin's JSON config from ``conf/`` into the VPP - build output directory (``build/vpp/``) so the .so can read it - from the stable location listed in vpp_plugins.conf. + Also copy each plugin's JSON config (and its license sibling) from + ``conf/`` into ``config.VPP_DATA_DIR`` (under PERSISTENT_DATA_DIR), + and REWRITE each ``config_path`` in vpp_plugins.conf to that + persistent absolute path. build/ is wiped by install.sh on a runtime + version update, which would otherwise delete a purchased license; the + persistent dir survives. Only the ``.so`` binary stays under build/vpp + (it is code, rebuilt each upload) — the C loader passes config_path to + the plugin verbatim, so the .so still finds its config and license. * **Upload does not include vpp_plugins.conf** → delete any existing ``vpp_plugins.conf`` from the runtime root. This ensures a vanilla upload never inadvertently loads a VPP driver left over from a previous project, regardless of what .so files exist in - ``build/vpp/``. + ``build/vpp/``. The persistent config/license are left in place, so a + device keeps its license if the VPP is re-added later. """ VPP_CONF_DEST = "vpp_plugins.conf" VPP_BUILD_DIR = "build/vpp" @@ -376,13 +383,14 @@ def apply_vpp_plugin_conf(generated_dir: str = "core/generated") -> None: shutil.copy2(uploaded_conf, VPP_CONF_DEST) build_state.log(f"[INFO] VPP: installed vpp_plugins.conf from upload\n") - # Copy each VPP plugin's config file to the path declared in - # vpp_plugins.conf (the config_path field). That field is the - # single source of truth for where the .so will look for its - # config at runtime — use it directly rather than constructing - # a separate destination. + # Copy each VPP plugin's config file into the persistent dir and rewrite + # its config_path to point there (see the loop below). config_path is the + # single source of truth for where the .so looks for its config at + # runtime, so relocating it there is what carries config+license out of + # the wipe-on-update build/ tree. conf_dir = os.path.join(generated_dir, "conf") vpp_conf_plugins = PluginsConfiguration.from_file(VPP_CONF_DEST) + rewrote_paths = False for p in vpp_conf_plugins.plugins: if not p.config_path: continue @@ -390,32 +398,79 @@ def apply_vpp_plugin_conf(generated_dir: str = "core/generated") -> None: if not os.path.exists(src_config): build_state.log(f"[WARNING] VPP: conf/{p.name}.json not found in upload, skipping\n") continue - dest_config = os.path.normpath(p.config_path) - # Guard against path traversal in editor-generated vpp_plugins.conf. - # Shares one containment definition with the 0x49 write path (see - # is_inside_root): rejects a sibling that merely shares runtime_root - # as a string prefix, AND resolves symlinks, which a lexical - # abspath check does not -- a link out of the tree would otherwise - # let an innocent-looking relative path write outside the root. - if not is_inside_root(dest_config, runtime_root): - build_state.log(f"[WARNING] VPP: config_path '{p.config_path}' escapes runtime root, skipping\n") + + # Relocate the config (and its license sibling) OUT of build/vpp and + # into PERSISTENT_DATA_DIR/vpp: install.sh does `rm -rf $OPENPLC_DIR/ + # build` on a runtime version update, which used to delete the + # purchased license with it. The .so still finds them because we + # rewrite config_path in vpp_plugins.conf below to this persistent + # absolute path -- the C loader passes config_path to the plugin + # verbatim (plugin_config.c only contains `path`, the .so itself, + # which stays under build/vpp). + # + # The destination is built from the plugin NAME (a basename), NEVER + # from the editor-supplied config_path, so a forged conf cannot steer + # the write outside the persistent dir. A name that is not a plain + # filename is refused rather than trusted. + if not p.name or os.path.basename(p.name) != p.name: + build_state.log(f"[WARNING] VPP: suspicious plugin name '{p.name}', skipping\n") + continue + dest_config = os.path.join(str(VPP_DATA_DIR), f"{p.name}.json") + if not is_inside_root(dest_config, str(VPP_DATA_DIR)): + build_state.log(f"[WARNING] VPP: config dest '{dest_config}' escapes the persistent dir, skipping\n") continue + # The old build/vpp sibling of THIS plugin, so a device licensed + # before this change can be migrated below. Derived from the FIXED + # build/vpp location plus the (already basename-checked) plugin name + # -- NOT from config_path. config_path is only confined to the runtime + # root by validate_vpp_plugins_conf (not to build/vpp), so deriving + # the migration source from it would let a forged conf point the read + # at any .license under the root and have it copied where 0x4A reads + # it back. The old code always wrote the license next to a build/vpp + # config, so this is exactly where a pre-change license lives, and it + # cannot be steered anywhere else. + old_license = os.path.join(runtime_root, VPP_BUILD_DIR, f"{p.name}.license") + os.makedirs(os.path.dirname(dest_config), exist_ok=True) shutil.copy2(src_config, dest_config) build_state.log(f"[INFO] VPP: copied {p.name}.json to {dest_config}\n") - # Deliver the optional device license blob alongside the config, at - # the sibling path the licensed plugin derives from its config path - # (derive_license_path, shared with vpp_license_debug.py's 0x49 + # Point the .so at the persistent config (absolute). This one line is + # what moves the license out of harm's way: the license sibling the + # .so derives from config_path now lives in the persistent dir too. + p.config_path = dest_config + rewrote_paths = True + dest_license = derive_license_path(dest_config) + + # Deliver the optional device license blob to the sibling of the + # persistent config (derive_license_path, shared with the 0x49 # handler so both write the SAME file the .so reads). Present only # for a licensed VPP whose device was activated; absent for free - # VPPs or demo devices. dest_config already passed the traversal - # guard above, so no need to re-check its .license sibling. + # VPPs or demo devices. src_license = os.path.join(conf_dir, f"{p.name}.license") if os.path.exists(src_license): - dest_license = derive_license_path(dest_config) shutil.copy2(src_license, dest_license) build_state.log(f"[INFO] VPP: copied {p.name}.license to {dest_license}\n") + elif old_license and os.path.exists(old_license) and not os.path.exists(dest_license): + # One-time migration: a device licensed before this change has + # its blob next to the OLD build/vpp config. Move it to the + # persistent sibling when the upload did not carry one, so the + # license is not orphaned in a directory install.sh wipes. + # Best-effort: a failure here just means the device re-activates + # from its existing entitlement on the next connect, as it does + # today when 0x4A reads EMPTY. + try: + shutil.copy2(old_license, dest_license) + build_state.log(f"[INFO] VPP: migrated {p.name}.license {old_license} -> {dest_license}\n") + except OSError as exc: + build_state.log(f"[WARNING] VPP: could not migrate {p.name}.license: {exc}\n") + + # Persist the rewritten config_path values so the C loader AND the + # 0x49/0x4A handlers (via _license_path) read the persistent location, + # not the build/vpp one the editor emitted. + if rewrote_paths: + vpp_conf_plugins.to_file(VPP_CONF_DEST) + build_state.log("[INFO] VPP: rewrote vpp_plugins.conf config_path to the persistent dir\n") else: # No VPP in this upload — remove any stale vpp_plugins.conf so # the plugin loader does not attempt to load old VPP drivers. diff --git a/webserver/vpp_license_debug.py b/webserver/vpp_license_debug.py index fb52fa93..56447e2f 100644 --- a/webserver/vpp_license_debug.py +++ b/webserver/vpp_license_debug.py @@ -31,6 +31,7 @@ import zlib from typing import Optional +from webserver import config from webserver.logger import get_logger from webserver.plugin_config_model import PluginsConfiguration @@ -206,15 +207,30 @@ def is_inside_root(path: str, runtime_root: Optional[str] = None) -> bool: def resolve_license_path(config_path: str, runtime_root: Optional[str] = None) -> Optional[str]: """``derive_license_path()`` plus the anti-traversal guard: never resolve - to a path outside the runtime root, even if a forged ``vpp_plugins.conf`` + to a path outside a KNOWN root, even if a forged ``vpp_plugins.conf`` carries an escaping ``config_path``. Returns None when it escapes. + + TWO roots are accepted, on purpose: + + * the runtime root (the legacy ``build/vpp`` location), and + * ``config.VPP_DATA_DIR`` under PERSISTENT_DATA_DIR, where + ``apply_vpp_plugin_conf`` now relocates VPP configs+licenses so a runtime + version update (which wipes ``$OPENPLC_DIR/build``) can no longer delete a + purchased license. After that relocation the ``config_path`` in + vpp_plugins.conf is an absolute path under VPP_DATA_DIR, and a guard that + only knew the runtime root would refuse the device its own license. + + This widens the guard to a SECOND known root, NOT to "anywhere": ``..`` and + arbitrary absolute paths still resolve outside both roots and are refused. """ path = derive_license_path(config_path) # An empty derivation (empty config_path) would resolve to the cwd, which # IS inside the root -- refuse it rather than let it through as a target. - if not path or not is_inside_root(path, runtime_root): + if not path: return None - return path + if is_inside_root(path, runtime_root) or is_inside_root(path, str(config.VPP_DATA_DIR)): + return path + return None def _license_path() -> Optional[str]: