From 4b645c6fb26dd3dd70c4545cda4ad70c93c3668f Mon Sep 17 00:00:00 2001 From: Marc Mintel Date: Sun, 3 May 2026 20:30:30 +0200 Subject: [PATCH] Hardening: Centralized zero-trust CognitiveStack validation and removed synthetic e2e test mocks --- GramAddict/core/bot_flow.py | 44 +++++++++++++++++++ .../test_workflow_feed_foreign_app_escape.py | 34 +++++--------- ...est_workflow_stories_foreign_app_escape.py | 8 ++-- 3 files changed, 57 insertions(+), 29 deletions(-) diff --git a/GramAddict/core/bot_flow.py b/GramAddict/core/bot_flow.py index 3296921..011820c 100644 --- a/GramAddict/core/bot_flow.py +++ b/GramAddict/core/bot_flow.py @@ -685,6 +685,7 @@ def _interact_with_profile(device, configs, username, session_state, sleep_mod, if cognitive_stack is None: cognitive_stack = {} + _validate_cognitive_stack(cognitive_stack, "ProfileInteraction") if hasattr(session_state, "my_username") and username == session_state.my_username: logger.info(f"🤝 [Deep Interaction] Skipping own profile @{username} to prevent self-interactions.") @@ -824,6 +825,7 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta from colorama import Fore + _validate_cognitive_stack(cognitive_stack, "StoriesLoop") logger.info("🎬 [StoriesFeed] Starting native story binging loop...", extra={"color": f"{Fore.CYAN}"}) dopamine = cognitive_stack.get("dopamine") @@ -893,6 +895,36 @@ def _run_zero_latency_stories_loop(device, configs, session_state, cognitive_sta return "FEED_EXHAUSTED" +def _validate_cognitive_stack(cognitive_stack, context_name): + """ + Validates that the cognitive stack has all required engine dependencies + injected properly. This is the ultimate zero-trust guard for plugins. + """ + if not isinstance(cognitive_stack, dict): + raise TypeError(f"[{context_name}] CognitiveStack must be a dict, got {type(cognitive_stack)}") + + required_engines = [ + "dopamine", + "darwin", + "resonance", + "active_inference", + "growth_brain", + "swarm", + "writer", + "nav_graph", + "zero_engine", + "telepathic", + ] + + missing = [eng for eng in required_engines if eng not in cognitive_stack or cognitive_stack[eng] is None] + if missing: + import logging + + logger = logging.getLogger(__name__) + logger.error(f"🚨 [{context_name}] CognitiveStack missing required engines: {missing}") + raise ValueError(f"[{context_name}] CognitiveStack missing required engines: {missing}") + + def _run_zero_latency_feed_loop( device, zero_engine, nav_graph, configs, session_state, job_target, cognitive_stack, is_reels=False ): @@ -906,6 +938,7 @@ def _run_zero_latency_feed_loop( - Darwin is the SOLE dwell controller → no duplicate sleep calls - SwarmProtocol emits pheromones after successful interactions """ + _validate_cognitive_stack(cognitive_stack, "FeedLoop") logger.info(f"🔄 Entering Zero-Latency Interaction Pool. Feed: {job_target}") dopamine = cognitive_stack.get("dopamine") @@ -1055,6 +1088,7 @@ def _run_zero_latency_search_loop( """ Executes the autonomous Search & Interact logic. """ + _validate_cognitive_stack(cognitive_stack, "SearchLoop") logger.info("🧠 [Search Engine] Initiating keyword discovery...", extra={"color": f"{Style.BRIGHT}{Fore.CYAN}"}) import random @@ -1080,6 +1114,16 @@ def _run_zero_latency_search_loop( xml = device.dump_hierarchy() telepathic = cognitive_stack.get("telepathic") + # ── Perimeter Guard: Verify we're still inside Instagram ── + if xml: + search_packages = set(re.findall(r'package="([^"]+)"', xml)) + search_app_id = getattr(device, "app_id", "com.instagram.android") + if search_packages and search_app_id not in search_packages: + logger.error(f"🚨 [SearchLoop] FOREIGN APP DETECTED! Packages: {search_packages}. Aborting loop.") + device.press("back") + sleep(1.5) + return "CONTEXT_LOST" + # Find search bar search_bar = telepathic.find_best_node(xml, "Search edit text box or magnifying glass input", device=device) if search_bar: diff --git a/tests/e2e/test_workflow_feed_foreign_app_escape.py b/tests/e2e/test_workflow_feed_foreign_app_escape.py index 6361c0e..a03ad13 100644 --- a/tests/e2e/test_workflow_feed_foreign_app_escape.py +++ b/tests/e2e/test_workflow_feed_foreign_app_escape.py @@ -11,13 +11,12 @@ import datetime from tests.e2e.conftest import E2EDeviceStub, load_fixture_xml -def test_feed_loop_escapes_foreign_app(e2e_configs): +def test_feed_loop_escapes_foreign_app(e2e_configs, e2e_cognitive_stack_factory): """ Simulates: Feed → Play Store. The loop must detect com.android.vending and abort immediately. """ from GramAddict.core.bot_flow import _run_zero_latency_feed_loop - from GramAddict.core.dopamine_engine import DopamineEngine feed_xml = load_fixture_xml("home_feed_with_ad.xml") play_store_xml = load_fixture_xml("play_store_from_story_link.xml") @@ -25,27 +24,8 @@ def test_feed_loop_escapes_foreign_app(e2e_configs): # Feed -> Play Store device = E2EDeviceStub([feed_xml, play_store_xml]) - dopamine = DopamineEngine() - dopamine.session_limit_seconds = 999 - - # Needs a dummy zero_engine and nav_graph - class DummyZeroEngine: - pass - - class DummyNavGraph: - def do(self, action): - pass - - def navigate_to(self, state, engine): - pass - - cognitive_stack = { - "dopamine": dopamine, - "zero_engine": DummyZeroEngine(), - "nav_graph": DummyNavGraph(), - "radome": None, - "active_inference": None, - } + cognitive_stack = e2e_cognitive_stack_factory(device) + cognitive_stack["dopamine"].session_limit_seconds = 999 session_state = type( "S", (), {"startTime": datetime.datetime.now(), "check_limit": lambda *args, **kwargs: (False, False)} @@ -55,7 +35,13 @@ def test_feed_loop_escapes_foreign_app(e2e_configs): e2e_configs.args.speed_multiplier = 0.01 result = _run_zero_latency_feed_loop( - device, DummyZeroEngine(), DummyNavGraph(), e2e_configs, session_state, "HomeFeed", cognitive_stack + device, + cognitive_stack["zero_engine"], + cognitive_stack["nav_graph"], + e2e_configs, + session_state, + "HomeFeed", + cognitive_stack, ) # Must return CONTEXT_LOST diff --git a/tests/e2e/test_workflow_stories_foreign_app_escape.py b/tests/e2e/test_workflow_stories_foreign_app_escape.py index ce79278..3669cb9 100644 --- a/tests/e2e/test_workflow_stories_foreign_app_escape.py +++ b/tests/e2e/test_workflow_stories_foreign_app_escape.py @@ -18,13 +18,12 @@ import datetime from tests.e2e.conftest import E2EDeviceStub, load_fixture_xml -def test_story_loop_escapes_foreign_app(e2e_configs): +def test_story_loop_escapes_foreign_app(e2e_configs, e2e_cognitive_stack_factory): """ Simulates: Story → Story → Story → Play Store (via swipe-up link). The loop must detect com.android.vending and abort immediately. """ from GramAddict.core.bot_flow import _run_zero_latency_stories_loop - from GramAddict.core.dopamine_engine import DopamineEngine story_xml = load_fixture_xml("story_view_before_linkout.xml") play_store_xml = load_fixture_xml("play_store_from_story_link.xml") @@ -32,9 +31,8 @@ def test_story_loop_escapes_foreign_app(e2e_configs): # 3 story iterations → then Play Store appears device = E2EDeviceStub([story_xml, story_xml, story_xml, play_store_xml]) - dopamine = DopamineEngine() - dopamine.session_limit_seconds = 999 - cognitive_stack = {"dopamine": dopamine} + cognitive_stack = e2e_cognitive_stack_factory(device) + cognitive_stack["dopamine"].session_limit_seconds = 999 session_state = type("S", (), {"startTime": datetime.datetime.now()})()