fix: address 7 HIGH security findings from deep audit

Identity (agent_id.py):
- V01: Enforce MAX_DELEGATION_DEPTH=10 to prevent Sybil attacks
- V02: Block wildcard '*' capability propagation via delegation

Marketplace (installer.py):
- V03: Path traversal guard — resolve+verify dest stays in plugins_dir
- V29: TOCTOU fix — re-verify manifest signature after dep resolution

Middleware:
- V16: Change http_middleware permissive_mode default to False
- V17: Add AGENTMESH_TRUSTED_PROXIES setting for header spoofing defense

Capability (capability.py):
- V30: Tighten prefix matching to require colon boundary

All 1853 tests pass, 0 failures.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: imran-siddique

packages/agent-mesh/src/agentmesh/identity/agent_id.py

@@ -8,7 +8,7 @@
 """
 
 from datetime import datetime
-from typing import Optional, Literal
+from typing import ClassVar, Optional, Literal
 from pydantic import BaseModel, Field, field_validator
 from cryptography.hazmat.primitives.asymmetric import ed25519
 from cryptography.hazmat.primitives import serialization
@@ -214,6 +214,9 @@ def verify_signature(self, data: bytes, signature: str) -> bool:
             logger.debug("Signature verification failed", exc_info=True)
             return False
 
+    # Maximum delegation depth to prevent Sybil attacks via infinite chains.
+    MAX_DELEGATION_DEPTH: ClassVar[int] = 10
+
     def delegate(
         self,
         name: str,
@@ -239,7 +242,25 @@ def delegate(
             description: Optional description.
             max_initial_trust_score: Upper bound on the child's initial trust
                 score. Typically set to the parent's current trust score.
+
+        Raises:
+            ValueError: If capabilities are not a subset, delegation depth
+                exceeds MAX_DELEGATION_DEPTH, or wildcard is propagated.
         """
+        # V01: Enforce maximum delegation depth
+        if self.delegation_depth >= self.MAX_DELEGATION_DEPTH:
+            raise ValueError(
+                f"Maximum delegation depth ({self.MAX_DELEGATION_DEPTH}) exceeded. "
+                f"Current depth: {self.delegation_depth}"
+            )
+
+        # V02: Block wildcard capability propagation
+        if "*" in capabilities:
+            raise ValueError(
+                "Cannot delegate wildcard capability '*'. "
+                "Explicitly list the capabilities to delegate."
+            )
+
         # Validate capabilities are a subset
         for cap in capabilities:
             if cap not in self.capabilities:

packages/agent-mesh/src/agentmesh/integrations/django_middleware/middleware.py

@@ -70,6 +70,16 @@ def _signature_header() -> str:
     def _exempt_paths() -> List[str]:
         return list(_get_setting("AGENTMESH_EXEMPT_PATHS", []))
 
+    @staticmethod
+    def _trusted_proxies() -> List[str]:
+        """Return list of trusted proxy IPs/CIDRs.
+
+        When set, DID headers are only trusted from these source IPs.
+        Empty list (default) means trust headers from any source — set
+        this in production to prevent header spoofing.
+        """
+        return list(_get_setting("AGENTMESH_TRUSTED_PROXIES", []))
+
     # ------------------------------------------------------------------
     # request processing
     # ------------------------------------------------------------------
@@ -80,6 +90,20 @@ def __call__(self, request: HttpRequest) -> HttpResponse:
             if request.path.startswith(prefix):
                 return self.get_response(request)
 
+        # V17: Validate request comes from a trusted proxy when configured
+        trusted_proxies = self._trusted_proxies()
+        if trusted_proxies:
+            remote_addr = request.META.get("REMOTE_ADDR", "")
+            if remote_addr not in trusted_proxies:
+                logger.warning(
+                    "Rejecting agent DID header from untrusted source: %s",
+                    remote_addr,
+                )
+                return JsonResponse(
+                    {"error": "Untrusted proxy", "detail": "Request source is not in AGENTMESH_TRUSTED_PROXIES."},
+                    status=403,
+                )
+
         did_header = self._did_header()
         sig_header = self._signature_header()
 

packages/agent-mesh/src/agentmesh/integrations/http_middleware.py

@@ -31,7 +31,8 @@ class TrustConfig:
 
     required_trust_score: float = 0.5
     required_capabilities: List[str] = field(default_factory=list)
-    permissive_mode: bool = True  # allow requests without trust headers
+    # V16: Default to strict mode — require explicit opt-in for permissive
+    permissive_mode: bool = False
 
 
 @dataclass

packages/agent-mesh/src/agentmesh/marketplace/installer.py

@@ -95,7 +95,7 @@ def install(
         """
         manifest = self._registry.get_plugin(name, version)
 
-        # Signature verification
+        # Signature verification (first check)
         if verify and manifest.signature and manifest.author in self._trusted_keys:
             public_key = self._trusted_keys[manifest.author]
             verify_signature(manifest, public_key)
@@ -106,8 +106,19 @@ def install(
             _seen = set()
         self._resolve_dependencies(manifest, _seen=_seen)
 
-        # Install to plugins directory
-        dest = self._plugins_dir / name
+        # V29: Re-verify signature after dependency resolution (TOCTOU guard)
+        if verify and manifest.signature and manifest.author in self._trusted_keys:
+            public_key = self._trusted_keys[manifest.author]
+            verify_signature(manifest, public_key)
+
+        # V03: Path traversal guard — ensure dest stays within plugins_dir
+        dest = (self._plugins_dir / name).resolve()
+        plugins_root = self._plugins_dir.resolve()
+        if not str(dest).startswith(str(plugins_root)):
+            raise MarketplaceError(
+                f"Plugin name '{name}' resolves outside plugins directory "
+                f"(path traversal blocked)"
+            )
         dest.mkdir(parents=True, exist_ok=True)
         manifest_file = dest / MANIFEST_FILENAME
         import yaml

packages/agent-mesh/src/agentmesh/trust/capability.py

@@ -113,8 +113,13 @@ def matches(self, requested: str, resource_id: Optional[str] = None) -> bool:
             prefix = self.capability[:-1]  # e.g. "read:" from "read:*"
             if not requested.startswith(prefix):
                 return False
-        elif requested.startswith(self.capability):
-            pass  # granted is a prefix of requested
+        elif ":" in self.capability and requested.startswith(self.capability + ":"):
+            # V30: Only allow prefix match at colon boundaries to prevent
+            # "read" matching "readwrite:secret". Require the granted
+            # capability to be a colon-delimited prefix of the requested one.
+            pass
+        elif self.capability == requested:
+            pass
         else:
             # Fall back to component matching
             req_action, req_resource, req_qualifier = self.parse_capability(requested)

packages/agent-mesh/tests/test_coverage_boost.py

@@ -598,11 +598,19 @@ class TestTrustMiddleware:
     """Tests for TrustMiddleware."""
 
     def test_verify_request_permissive_no_headers(self):
-        mw = TrustMiddleware()
+        cfg = TrustConfig(permissive_mode=True)
+        mw = TrustMiddleware(config=cfg)
         result, err = mw.verify_request({})
         assert result.verified is True
         assert err is None
 
+    def test_verify_request_default_strict_no_headers(self):
+        """V16: Default config is now strict (permissive_mode=False)."""
+        mw = TrustMiddleware()
+        result, err = mw.verify_request({})
+        assert result.verified is False
+        assert err is not None
+
     def test_verify_request_strict_no_headers(self):
         cfg = TrustConfig(permissive_mode=False)
         mw = TrustMiddleware(config=cfg)

packages/agent-mesh/tests/test_negative_security.py

@@ -64,30 +64,30 @@ def _setup_handshake():
 class TestDelegationDepthAbuse:
     """Verify delegation depth limits are respected."""
 
-    def test_deep_delegation_chain(self):
-        """An agent can create arbitrarily deep delegation chains.
-        This is a potential Sybil attack vector — each level spawns
-        a new identity with a fresh keypair.
-        """
+    def test_deep_delegation_chain_blocked(self):
+        """Delegation beyond MAX_DELEGATION_DEPTH is rejected."""
         root = _create_agent("root", ["read:data", "write:data"])
         current = root
-        # Create a 20-level deep delegation chain
-        for i in range(20):
+        # Create chain up to the max (10)
+        for i in range(AgentIdentity.MAX_DELEGATION_DEPTH):
             current = current.delegate(
                 name=f"child-{i}",
                 capabilities=["read:data"],
             )
-        assert current.delegation_depth == 20
-        # The child at depth 20 still works — no enforcement
-        assert current.is_active()
+        assert current.delegation_depth == AgentIdentity.MAX_DELEGATION_DEPTH
+        # The next delegation should be rejected
+        with pytest.raises(ValueError, match="Maximum delegation depth"):
+            current.delegate(name="too-deep", capabilities=["read:data"])
 
-    def test_delegated_agent_retains_capabilities(self):
-        """Even at extreme depth, capabilities are preserved."""
+    def test_delegation_at_max_depth_still_works(self):
+        """Delegation at depth MAX-1 succeeds (boundary check)."""
         root = _create_agent("root", ["read:data"])
-        child = root
-        for _ in range(10):
-            child = child.delegate(name="deep", capabilities=["read:data"])
-        assert child.has_capability("read:data")
+        current = root
+        for i in range(AgentIdentity.MAX_DELEGATION_DEPTH - 1):
+            current = current.delegate(name=f"child-{i}", capabilities=["read:data"])
+        # One more should still work (depth == MAX-1 → child depth == MAX)
+        child = current.delegate(name="last", capabilities=["read:data"])
+        assert child.delegation_depth == AgentIdentity.MAX_DELEGATION_DEPTH
 
 
 # ===========================================================================
@@ -110,13 +110,13 @@ def test_wildcard_parent_cannot_delegate_arbitrary_capabilities(self):
                 capabilities=["admin:delete-all", "nuclear:launch"],
             )
 
-    def test_wildcard_parent_can_delegate_wildcard(self):
-        """Parent with '*' can delegate '*' itself — this is the only
-        way to pass through the wildcard.
+    def test_wildcard_delegation_blocked(self):
+        """V02: Parent with '*' CANNOT delegate '*' itself —
+        wildcard propagation is explicitly blocked.
         """
         root = _create_agent("root", ["*"])
-        child = root.delegate(name="child", capabilities=["*"])
-        assert child.has_capability("anything:at:all")
+        with pytest.raises(ValueError, match="Cannot delegate wildcard"):
+            root.delegate(name="child", capabilities=["*"])
 
     def test_star_capability_matches_everything(self):
         agent = _create_agent("super", ["*"])