AI-written Python plugin sandbox code exposed built-in type objects to untrusted exec code, allowing type-hierarchy traversal to recover host capabilities and execute commands.

         BlockEncoder,         ReedSolomonDecoder,         ReedSolomonEncoder,     )+     ERROR_CORRECTION_AVAILABLE = True except ImportError:     ERROR_CORRECTION_AVAILABLE = False@@ -116,6 +116,7 @@ def silent_warn(message, category=None, stacklevel=1, source=None):         QIMAnalyzer,         UniformQIM,     )+     QIM_ALGORITHM_AVAILABLE = True except ImportError:     QIM_ALGORITHM_AVAILABLE = False@@ -126,6 +127,7 @@ def silent_warn(message, category=None, stacklevel=1, source=None):         ClassicalSteganalysis,         InformationTheoreticSecurity,     )+     STEGANALYSIS_AVAILABLE = True except ImportError:     STEGANALYSIS_AVAILABLE = False@@ -11049,5 +11051,1030 @@ def test_no_security_warning_with_defaults(self):     print(f"⚠️  Error importing keystore integration tests: {e}")  +# =============================================================================+# PLUGIN SYSTEM TESTS+# =============================================================================+++class TestPluginSystem(unittest.TestCase):+    """Test cases for the secure plugin system."""++    def setUp(self):+        """Set up test environment."""+        self.test_dir = tempfile.mkdtemp()+        self.plugin_dir = os.path.join(self.test_dir, "plugins")+        self.config_dir = os.path.join(self.test_dir, "config")+        os.makedirs(self.plugin_dir, exist_ok=True)+        os.makedirs(self.config_dir, exist_ok=True)++    def tearDown(self):+        """Clean up test environment."""+        if os.path.exists(self.test_dir):+            shutil.rmtree(self.test_dir)++    def test_plugin_capability_enum(self):+        """Test PluginCapability enum values."""+        try:+            from ..modules.plugin_system import PluginCapability++            # Test all expected capabilities exist+            expected_capabilities = [+                "READ_FILES",+                "MODIFY_METADATA",+                "ACCESS_CONFIG",+                "WRITE_LOGS",+                "NETWORK_ACCESS",+                "EXECUTE_PROCESSES",+            ]++            for cap_name in expected_capabilities:+                self.assertTrue(hasattr(PluginCapability, cap_name))+                cap = getattr(PluginCapability, cap_name)+                self.assertIsInstance(cap.value, str)++        except ImportError:+            self.skipTest("Plugin system not available")++    def test_plugin_type_enum(self):+        """Test PluginType enum values."""+        try:+            from ..modules.plugin_system import PluginType++            expected_types = [+                "PRE_PROCESSOR",+                "POST_PROCESSOR",+                "METADATA_HANDLER",+                "FORMAT_CONVERTER",

+This module provides the foundational base classes for the secure plugin system.+All plugins must inherit from these classes to ensure security boundaries are+maintained and sensitive data is never exposed to third-party code.++Security Architecture:+- Zero-trust design: plugins never access passwords, keys, or plaintext+- Capability-based security model with explicit permissions+- Sandboxed execution environment with restricted operations+- Type-safe interfaces with validation and sanitization+"""++import abc+import enum+import logging+import time+from typing import Any, Dict, List, Optional, Set, Union++logger = logging.getLogger(__name__)+++class PluginCapability(enum.Enum):+    """+    Security capabilities that plugins can request.+    Each capability grants specific, limited permissions.+    """++    # File system access (read-only, specific paths)+    READ_FILES = "read_files"++    # Metadata manipulation (non-sensitive only)+    MODIFY_METADATA = "modify_metadata"++    # Configuration access (plugin-specific config only)+    ACCESS_CONFIG = "access_config"++    # Logging capabilities (filtered, no sensitive data)+    WRITE_LOGS = "write_logs"++    # Network access (disabled by default for security)+    NETWORK_ACCESS = "network_access"++    # Process execution (highly restricted)+    EXECUTE_PROCESSES = "execute_processes"+++class PluginType(enum.Enum):+    """+    Plugin types defining where in the pipeline plugins can hook.+    Each type has specific interfaces and security boundaries.+    """++    # Pre-processing plugins (work with input files before encryption)+    PRE_PROCESSOR = "pre_processor"++    # Post-processing plugins (work with output files after encryption)+    POST_PROCESSOR = "post_processor"++    # Metadata plugins (manipulate non-sensitive metadata)+    METADATA_HANDLER = "metadata_handler"++    # Format plugins (handle input/output format conversions)+    FORMAT_CONVERTER = "format_converter"++    # Analysis plugins (analyze encrypted files without decrypting)+    ANALYZER = "analyzer"++    # Utility plugins (provide helper functions)+    UTILITY = "utility"+++class PluginSecurityContext:+    """+    Security context passed to plugins containing safe, sanitized data.+    This is the ONLY data interface plugins can access.+    """++    def __init__(self, plugin_id: str, capabilities: Set[PluginCapability]):+        self.plugin_id = plugin_id+        self.capabilities = capabilities+        self.metadata = {}  # Only non-sensitive metadata+        self.file_paths = []  # Only paths to encrypted files or safe temp files+        self.config = {}  # Plugin-specific configuration only+        self.timestamp = time.time()++    def has_capability(self, capability: PluginCapability) -> bool:

 except ImportError:@@ -1674,7 +1674,11 @@ def show_progress(algorithm, current, total):                     "whirlpool",                 ]             )-            or (hash_config and hash_config.get("scrypt", {}).get("enabled", False) and hash_config.get("scrypt", {}).get("rounds", 0) > 0)+            or (+                hash_config+                and hash_config.get("scrypt", {}).get("enabled", False)+                and hash_config.get("scrypt", {}).get("rounds", 0) > 0+            )         )      if has_hash_iterations:@@ -1690,7 +1694,7 @@ def show_progress(algorithm, current, total):         # Even when no hash iterations are configured, we need to combine password with salt         # for consistency with the original key derivation behavior         password = password + salt-    +     # Check if Argon2 is available on the system     argon2_available = ARGON2_AVAILABLE @@ -1725,23 +1729,33 @@ def show_progress(algorithm, current, total):         is_decryption = hash_config and hash_config.get("_is_from_decryption_metadata", False)         if not is_decryption:             enabled_kdfs = []-            if use_argon2: enabled_kdfs.append("Argon2")-            if use_randomx: enabled_kdfs.append("RandomX") -            if use_balloon: enabled_kdfs.append("Balloon")-            if use_hkdf: enabled_kdfs.append("HKDF")-            if use_scrypt: enabled_kdfs.append("Scrypt")-            +            if use_argon2:+                enabled_kdfs.append("Argon2")+            if use_randomx:+                enabled_kdfs.append("RandomX")+            if use_balloon:+                enabled_kdfs.append("Balloon")+            if use_hkdf:+                enabled_kdfs.append("HKDF")+            if use_scrypt:+                enabled_kdfs.append("Scrypt")+             print(f"\n⚠️  WARNING: Security Risk Detected")-            print(f"KDFs ({', '.join(enabled_kdfs)}) will operate directly on your password without prior hashing.")+            print(+                f"KDFs ({', '.join(enabled_kdfs)}) will operate directly on your password without prior hashing."+            )             print(f"This may be insecure if your password is short or has low entropy.")-            print(f"Consider adding hash rounds (--sha256-rounds, --blake2b-rounds, etc.) for better security.")+            print(+                f"Consider adding hash rounds (--sha256-rounds, --blake2b-rounds, etc.) for better security."+            )             print(f"Continue anyway? [y/N]: ", end="", flush=True)-            +             # Get user confirmation             import sys+             try:                 response = input().strip().lower()-                if response not in ['y', 'yes']:+                if response not in ["y", "yes"]:                     print("Operation cancelled by user.")                     sys.exit(1)                 print()  # Add blank line after confirmation@@ -2185,10 +2199,10 @@ def show_progress(algorithm, current, total):             password_for_salt = bytes(password)         else:             password_for_salt = bytes(password)-        +         # Create unique salt for RandomX by combining original salt with current password state         # This prevents salt reuse while maintaining deterministic behavior-        randomx_salt_material = salt + password_for_salt[:16] + b'randomx_salt'+        randomx_salt_material = salt + password_for_salt[:16] + b"randomx_salt"         base_salt = hashlib.sha256(randomx_salt_material).digest()[:16]         if not quiet and not progress:             print("Using RandomX for key derivation", end=" ")@@ -2247,7 +2261,7 @@ def show_progress(algorithm, current, total): 

++    def test_globals_call_detected(self):+        """globals() can leak the module namespace and enable sandbox escape"""+        code = """+def test():+    g = globals()+    return g+"""+        is_safe, violations = analyze_plugin_code(code, "test.py", strict_mode=True)++        assert not is_safe+        assert any(v.violation_type == "dangerous_function" for v in violations)+        assert any("globals" in v.description for v in violations)++    def test_locals_call_detected(self):+        """locals() can leak local scope for introspection attacks"""+        code = """+def test():+    l = locals()+    return l+"""+        is_safe, violations = analyze_plugin_code(code, "test.py", strict_mode=True)++        assert not is_safe+        assert any(v.violation_type == "dangerous_function" for v in violations)+        assert any("locals" in v.description for v in violations)++    def test_vars_call_detected(self):+        """vars() exposes object __dict__ for attribute manipulation"""+        code = """+def test(obj):+    v = vars(obj)+    return v+"""+        is_safe, violations = analyze_plugin_code(code, "test.py", strict_mode=True)++        assert not is_safe+        assert any(v.violation_type == "dangerous_function" for v in violations)+        assert any("vars" in v.description for v in violations)++    def test_dir_call_detected(self):+        """dir() enables attribute discovery for sandbox escape"""+        code = """+def test(obj):+    attrs = dir(obj)+    return attrs+"""+        is_safe, violations = analyze_plugin_code(code, "test.py", strict_mode=True)++        assert not is_safe+        assert any(v.violation_type == "dangerous_function" for v in violations)++    def test_type_call_detected(self):+        """type() with 3 args creates new classes dynamically, enabling sandbox escape"""+        code = """+def test():+    MyClass = type("MyClass", (object,), {"x": 1})+    return MyClass+"""+        is_safe, violations = analyze_plugin_code(code, "test.py", strict_mode=True)++        assert not is_safe+        assert any(v.violation_type == "dangerous_function" for v in violations)++    def test_breakpoint_call_detected(self):+        """breakpoint() drops into debugger which can execute arbitrary code"""+        code = """+def test():+    breakpoint()+    return True+"""+        is_safe, violations = analyze_plugin_code(code, "test.py", strict_mode=True)++        assert not is_safe+        assert any(v.violation_type == "dangerous_function" for v in violations)++    def test_class_dunder_access_detected(self):+        """__class__ attribute enables type hierarchy traversal escape"""+        code = """+def test():+    x = "".__class__.__bases__[0].__subclasses__()+    return x

     Args:         plugin: Plugin instance to validate      Raises:-        SandboxViolationError: If plugin code fails security analysis+        SandboxViolationError: If plugin code fails security analysis or+            the source cannot be retrieved for validation.     """     import inspect     import textwrap -    try:-        source = inspect.getsource(type(plugin))-    except (OSError, TypeError):-        # Can't get source (e.g., built-in or C extension) - skip validation-        logger.debug(f"Could not get source for plugin {plugin.plugin_id}, skipping AST validation")-        return+    # Try to get the full module source first (catches module-level code).+    # Fall back to the class source only if the module source is unavailable.+    source = None+    plugin_class = type(plugin)+    module = inspect.getmodule(plugin_class)++    if module is not None:+        try:+            source = inspect.getsource(module)+        except (OSError, TypeError):+            pass++    if source is None:+        try:+            source = inspect.getsource(plugin_class)+        except (OSError, TypeError):+            # Fail closed: reject plugins whose source cannot be analyzed+            raise SandboxViolationError(+                f"Plugin '{plugin.plugin_id}' rejected: source code unavailable for "+                f"security analysis (dynamically-created or compiled plugins are not allowed)"+            )      # Dedent the source code so ast.parse() works even when the class is     # defined inside an indented block (e.g., inside a try/except).@@ -1028,10 +1048,42 @@ def execute_in_process(         def target_function(plugin_code, context_data, result_queue):             """Target function for isolated execution."""             try:-                # This is a simplified example-                # In production, you'd need proper serialization/deserialization+                # Validate plugin code via AST analysis before execution+                is_safe, violations = analyze_plugin_code(+                    plugin_code, "<isolated-exec>", strict_mode=True+                )+                if not is_safe:+                    critical = [v for v in violations if v.severity == "critical"]+                    msgs = "; ".join(f"L{v.line}: {v.description}" for v in critical[:5])+                    result_queue.put(+                        (+                            "error",+                            PluginResult.error_result(f"Code failed security analysis: {msgs}"),+                        )+                    )+                    return++                # Apply resource limits (Linux only)+                try:+                    import resource as _resource++                    if hasattr(_resource, "RLIMIT_CPU"):+                        _resource.setrlimit(_resource.RLIMIT_CPU, (60, 60))+                    if hasattr(_resource, "RLIMIT_FSIZE"):+                        _resource.setrlimit(+                            _resource.RLIMIT_FSIZE, (50 * 1024 * 1024, 50 * 1024 * 1024)+                        )+                    if hasattr(_resource, "RLIMIT_NOFILE"):+                        _resource.setrlimit(_resource.RLIMIT_NOFILE, (64, 64))+                except (ImportError, OSError, ValueError):+                    pass++                # Install import guard to block dangerous modules+                guard = PluginImportGuard(None)+                guard.hide_dangerous_modules()+                sys.meta_path.insert(0, guard) 

+        "type",+        "breakpoint",     }      # Dunder attributes that enable type hierarchy traversal / sandbox escape.@@ -71,6 +77,10 @@ class DangerousPatternVisitor(ast.NodeVisitor):         "__code__",         "__reduce__",         "__reduce_ex__",+        "__class__",+        "__import__",+        "__loader__",+        "__spec__",     }      # Use the shared blocked modules set to keep AST and runtime in sync@@ -84,7 +94,16 @@ class DangerousPatternVisitor(ast.NodeVisitor):         "popen",         "spawn",         "exec",+        "execl",+        "execle",+        "execlp",+        "execlpe",+        "execv",+        "execve",+        "execvp",+        "execvpe",         "fork",+        "forkpty",         "kill",         "killpg",         "spawnl",@@ -95,6 +114,26 @@ class DangerousPatternVisitor(ast.NodeVisitor):         "spawnve",         "spawnvp",         "spawnvpe",+        "popen2",+        "popen3",+        "popen4",+        "remove",+        "unlink",+        "rmdir",+        "removedirs",+        "rename",+        "renames",+        "symlink",+        "link",+        "chmod",+        "chown",+        "chroot",+        "lchmod",+        "lchown",+        "setuid",+        "setgid",+        "seteuid",+        "setegid",     }      def __init__(self, strict_mode: bool = True):@@ -249,6 +288,16 @@ def visit_Call(self, node: ast.Call) -> None:                                     f"This is a known sandbox bypass technique.",                                     "critical",                                 )+                else:+                    # Dynamic getattr with non-constant attribute name — potential bypass+                    self.add_violation(+                        node,+                        "dynamic_getattr",+                        "getattr() called with a dynamic (non-constant) attribute name. "+                        "This can be used to bypass sandbox restrictions by computing "+                        "dangerous attribute names at runtime.",+                        "high",+                    )          self.generic_visit(node) @@ -365,10 +414,12 @@ def analyze_plugin_code(         visitor = DangerousPatternVisitor(strict_mode=strict_mode)         visitor.visit(tree) -        # In strict mode, any critical violation fails validation