diff --git a/engraphis/licensing.py b/engraphis/licensing.py index 7fbcb8b..9126965 100644 --- a/engraphis/licensing.py +++ b/engraphis/licensing.py @@ -1055,19 +1055,36 @@ def _verify_module_integrity(): _lock_sentinel = object() _LOCK_SNAPSHOT = None +_LOCK_TAKEN = False 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. + The snapshot is taken lazily on the first gate call, so test-mode setup + (conftest sets ``_TEST_MODE_PUBKEY_OVERRIDE=True`` after import) is + captured correctly without false positives. """ - if _TEST_MODE_PUBKEY_OVERRIDE: - return + global _LOCK_SNAPSHOT, _LOCK_TAKEN + if not _LOCK_TAKEN: + _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, + } + _LOCK_TAKEN = True snap = _LOCK_SNAPSHOT - if snap is None: - return + if snap is None or not isinstance(snap, dict): + raise LicenseError( + "Licensing integrity violation: tamper-detection snapshot has been " + "corrupted. Reinstall Engraphis from the official distribution." + ) + if _TEST_MODE_PUBKEY_OVERRIDE: + return # test mode: skip callable checks g = globals() for name, expected in snap.items(): current = g.get(name, _lock_sentinel) @@ -1079,8 +1096,10 @@ def _verify_no_tampering(): def _snapshot_critical_globals(): - global _LOCK_SNAPSHOT + """Force a re-snapshot of critical globals (used by tests after mocking).""" + global _LOCK_SNAPSHOT, _LOCK_TAKEN _LOCK_SNAPSHOT = { + "_TEST_MODE_PUBKEY_OVERRIDE": _TEST_MODE_PUBKEY_OVERRIDE, "has_feature": has_feature, "require_feature": require_feature, "current_license": current_license, @@ -1089,9 +1108,7 @@ def _snapshot_critical_globals(): "vendor_public_key": vendor_public_key, "vendor_public_keys": vendor_public_keys, } - - -_snapshot_critical_globals() + _LOCK_TAKEN = True _verify_module_integrity() diff --git a/engraphis/routes/vault.py b/engraphis/routes/vault.py index 4c7143e..244feb4 100644 --- a/engraphis/routes/vault.py +++ b/engraphis/routes/vault.py @@ -605,6 +605,8 @@ async def context_preview(req: ContextPreviewReq): @router.get("/vaults/{namespace}/export") async def export_vault(namespace: str): """GET /memory/vaults/{namespace}/export — export all memories in a vault as JSON.""" + from engraphis.licensing import require_feature + require_feature("export") docs = mem_store.list_documents(namespace=namespace, limit=10000) export_data = { "namespace": namespace, diff --git a/scripts/consolidate.py b/scripts/consolidate.py index 267b754..fa6bae3 100644 --- a/scripts/consolidate.py +++ b/scripts/consolidate.py @@ -204,6 +204,17 @@ def main(argv=None) -> int: llm = None if args.llm or args.structured: + # LLM-powered consolidation (inference / structured merging) is a paid + # automation feature — gate here because we call engine.consolidate() + # directly, which has no license awareness. + from engraphis.licensing import require_feature, LicenseError + try: + require_feature("automation") + except LicenseError as exc: + print(f"error: LLM-powered consolidation ({exc})", file=sys.stderr) + print("tip: run without --llm or --structured for the free, deterministic pass", + file=sys.stderr) + return 2 try: from engraphis.llm.client import LLMClient llm = LLMClient()