feat(licensing): Phase 2 hardening - close tamper + ENGRAPHIS_DEV bypass vectors

engraphis/cloud_license.py

@@ -582,6 +582,7 @@ def gate(lic, key_material: str, *, base_url: Optional[str] = None) -> Tuple[boo
     an authoritative denial fails closed immediately, while a transient network failure
     may use an existing unexpired lease as offline grace. Without such a lease, failure
     to register fails closed."""
+    _verify_gate_integrity()
     base = (base_url or "").strip().rstrip("/") or cloud_url()
     if not base:
         # Online-only: no server to verify against ⇒ no offline path to paid features.
@@ -660,15 +661,17 @@ def revalidate(lic, key_material: str, *, base_url: Optional[str] = None) -> str
 
 def _verify_module_integrity():
     """Detect if this module was replaced with editable source after a compiled
-    extension was already installed (same check as licensing.py)."""
+    extension was already installed.
+
+    No env-var escape hatch — dev installs never build .pyd/.so, so the check
+    passes naturally.
+    """
     mod = sys.modules.get(__name__)
     if mod is None:
         return
     f = getattr(mod, "__file__", "")
     if not f:
         return
-    if os.environ.get("ENGRAPHIS_DEV", "").strip() in ("1", "true", "yes"):
-        return
     if not f.endswith(".py"):
         return
     from importlib.machinery import EXTENSION_SUFFIXES
@@ -679,8 +682,27 @@ def _verify_module_integrity():
             raise RuntimeError(
                 "Engraphis cloud_license integrity check failed: a compiled native "
                 "extension exists but is not being loaded. Reinstall from the "
-                "official distribution. For development, set ENGRAPHIS_DEV=1."
+                "official distribution."
             )
 
 
+_lock_sentinel = object()
+_GATE_SNAPSHOT = gate
+
+
+def _verify_gate_integrity():
+    """Raise if ``gate`` has been monkeypatched since import."""
+    try:
+        from engraphis.licensing import _TEST_MODE_PUBKEY_OVERRIDE
+        if _TEST_MODE_PUBKEY_OVERRIDE:
+            return
+    except ImportError:
+        pass
+    if gate != _GATE_SNAPSHOT:
+        raise RuntimeError(
+            "Engraphis cloud_license gate has been tampered with at runtime. "
+            "Reinstall from the official distribution."
+        )
+
+
 _verify_module_integrity()

engraphis/licensing.py

@@ -680,6 +680,7 @@ def current_license(*, refresh: bool = False) -> License:
     Never raises: a bad, revoked, expired, or currently unverifiable key degrades to the
     free tier and the reason is kept in :func:`license_error`.
     """
+    _verify_no_tampering()
     global _cached, _cache_error, _cache_recheck_at, _cache_generation
     # Fast path under the lock: a valid, unexpired cache entry is returned atomically so a
     # concurrent invalidate_cache() can't null it out between the check and the return.
@@ -933,6 +934,7 @@ def activate(key: str) -> License:
 
 
 def has_feature(feature: str) -> bool:
+    _verify_no_tampering()
     return current_license().has(feature)
 
 
@@ -949,6 +951,7 @@ def require_feature(feature: str) -> None:
 
     This is THE gate helper — every paid surface (Inspector routes, report scripts)
     funnels through here, so upgrade messaging changes in exactly one place."""
+    _verify_no_tampering()
     if not has_feature(feature):
         desc = FEATURES.get(feature, feature)
         tier = required_plan(feature)
@@ -1024,20 +1027,18 @@ def _verify_module_integrity():
     to patch the source. If no compiled extension exists (pure-python install on
     a platform without wheels), the check passes.
 
-    ``ENGRAPHIS_DEV=1`` skips the check entirely for local development.
+    There is deliberately NO env-var escape hatch (Phase 2 hardening). Dev
+    installs (pip install -e .) never build .pyd/.so, so the check passes
+    naturally without any env-var bypass.
     """
     mod = sys.modules.get(__name__)
     if mod is None:
         return
     f = getattr(mod, "__file__", "")
     if not f:
         return
-    if os.environ.get("ENGRAPHIS_DEV", "").strip() in ("1", "true", "yes"):
-        return
     if not f.endswith(".py"):
-        return  # running as compiled extension — all good
-    # Running as .py — check if a compiled extension exists alongside.
-    # If it does, someone replaced the extension with editable source.
+        return
     from importlib.machinery import EXTENSION_SUFFIXES
     dirname = os.path.dirname(f)
     basename = os.path.splitext(os.path.basename(f))[0]
@@ -1048,11 +1049,49 @@ def _verify_module_integrity():
                 "extension (.pyd/.so) exists but is not being loaded — the module "
                 "may have been replaced with editable source. Reinstall Engraphis "
                 "from the official distribution (pip install --force-reinstall "
-                "engraphis). For development, set ENGRAPHIS_DEV=1."
+                "engraphis)."
+            )
+
+
+_lock_sentinel = object()
+_LOCK_SNAPSHOT = None
+
+
+def _verify_no_tampering():
+    """Verify critical gate callables haven't been monkeypatched since import.
+
+    Skipped in test mode (``_TEST_MODE_PUBKEY_OVERRIDE=True``) so the test
+    suite can mock these freely.
+    """
+    if _TEST_MODE_PUBKEY_OVERRIDE:
+        return
+    snap = _LOCK_SNAPSHOT
+    if snap is None:
+        return
+    g = globals()
+    for name, expected in snap.items():
+        current = g.get(name, _lock_sentinel)
+        if current is _lock_sentinel or current != expected:
+            raise LicenseError(
+                "Licensing integrity violation: '%s' has been tampered with at "
+                "runtime. Reinstall Engraphis from the official distribution." % name
             )
-    # No compiled extension found — pure-python install on a platform without
-    # compiled wheels. This is less secure (the source is plaintext and patchable),
-    # but it's the expected fallback for platforms we don't pre-compile for.
+
+
+def _snapshot_critical_globals():
+    global _LOCK_SNAPSHOT
+    _LOCK_SNAPSHOT = {
+        "has_feature": has_feature,
+        "require_feature": require_feature,
+        "current_license": current_license,
+        "parse_key": parse_key,
+        "ed25519_verify": ed25519_verify,
+        "vendor_public_key": vendor_public_key,
+        "vendor_public_keys": vendor_public_keys,
+    }
+
+
+_snapshot_critical_globals()
 
 
 _verify_module_integrity()

tests/conftest.py

@@ -1,9 +1,12 @@
 import os
 
-# The licensing + cloud_license integrity guards (added 2026-07) refuse to import from
-# editable .py source in production installs. The entire test suite is development mode —
-# set ENGRAPHIS_DEV=1 BEFORE any import so the guards allow the plain .py source.
-os.environ.setdefault("ENGRAPHIS_DEV", "1")
+# Opt the licensing module into honoring ENGRAPHIS_LICENSE_PUBKEY, which is otherwise
+# dead in a shipped process. Set at import time so it covers both collection and
+# execution. This is the ONLY place that flips the switch — production never imports
+# this conftest, so the vendor-key override stays non-overridable in the field.
+# TEST_MODE_PUBKEY_OVERRIDE also disables the tamper-detection checks in licensing
+# has_feature / require_feature / current_license and cloud_license gate, so existing
+# test mocking continues to work.
 
 import pytest
 from engraphis import cloud_license, licensing

tests/test_compiled_integrity.py

@@ -1,4 +1,4 @@
-"""Tests for the licensing + cloud_license compilation integrity guards."""
+"""Tests for the licensing + cloud_license compilation integrity guards and tamper detection."""
 
 import sys
 
@@ -7,45 +7,53 @@
 import engraphis.licensing as lic
 
 
-# ── integrity guard ──
-
-def test_guard_skipped_in_dev_mode():
-    """ENGRAPHIS_DEV=1 is set by conftest.py — the guard does nothing."""
-    lic._verify_module_integrity()
-
-
-def test_guard_passes_when_running_as_py_without_pyd():
-    """On our dev machine (no compiled .pyd/.so alongside licensing.py), the guard
-    passes — it's the expected fallback for pure-python installs."""
+def test_guard_passes_when_no_compiled_extension():
     lic._verify_module_integrity()
 
 
 def test_guard_raises_when_pyd_exists_alongside(monkeypatch, tmp_path):
-    """If a .pyd file exists next to licensing.py, the guard fires because the
-    compiled extension should have been loaded instead."""
     from importlib.machinery import EXTENSION_SUFFIXES
-    monkeypatch.delenv("ENGRAPHIS_DEV", raising=False)
     py_path = tmp_path / "licensing.py"
     py_path.write_text("# stub")
-    # Create a file matching any extension suffix (e.g. .pyd or .cp311-win_amd64.pyd)
     for suffix in EXTENSION_SUFFIXES:
         (tmp_path / ("licensing" + suffix)).write_text("")
-        break  # one is enough for the guard to fire
-
+        break
     monkeypatch.setattr(sys.modules["engraphis.licensing"], "__file__", str(py_path))
     with pytest.raises(lic.LicenseError, match="integrity check failed"):
         lic._verify_module_integrity()
 
 
 def test_guard_passes_when_compiled_extension(monkeypatch):
-    """When __file__ ends in .pyd, the guard returns immediately without checking
-    for a sibling .py."""
-    monkeypatch.delenv("ENGRAPHIS_DEV", raising=False)
     monkeypatch.setattr(lic, "__file__", "/path/engraphis/licensing.cp312-win_amd64.pyd")
     lic._verify_module_integrity()
 
 
-# ── production warnings ──
+# ── tamper detection ──
+
+def test_tamper_detection_skipped_in_test_mode():
+    try:
+        lic.require_feature("nonexistent")
+    except lic.LicenseError as exc:
+        assert "nonexistent" in str(exc)
+        assert "integrity" not in str(exc).lower()
+
+
+def test_tamper_detects_replaced_has_feature(monkeypatch):
+    monkeypatch.setattr(lic, "_TEST_MODE_PUBKEY_OVERRIDE", False)
+    lic._snapshot_critical_globals()
+    monkeypatch.setattr(lic, "has_feature", lambda _: True)
+    with pytest.raises(lic.LicenseError, match="has_feature"):
+        lic.require_feature("nonexistent")
+
+
+def test_tamper_detects_replaced_current_license(monkeypatch):
+    monkeypatch.setattr(lic, "_TEST_MODE_PUBKEY_OVERRIDE", False)
+    lic._snapshot_critical_globals()
+    monkeypatch.setattr(lic, "current_license", lambda **kw: lic.License(plan="team",
+        features=frozenset({"analytics", "export", "automation", "team", "sync"})))
+    with pytest.raises(lic.LicenseError, match="current_license"):
+        lic.require_feature("nonexistent")
+
 
 def test_production_warnings_returns_list():
     assert isinstance(lic.production_warnings(), list)