-
Notifications
You must be signed in to change notification settings - Fork 8.1k
feat(tools): add Creduent zero-trust agent verification tool (Closes #6773) #6780
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
895971d
dd080c8
d7da6af
5d6326d
b7dbfb6
495221d
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| # Creduent Verification Tool | ||
|
|
||
| Verify external agent identities and cryptographic attestations locally before delegating tasks in CrewAI workflows. | ||
|
|
||
| ## Installation | ||
|
|
||
| ```bash | ||
| pip install creduent crewai-tools | ||
| ``` | ||
|
|
||
| ## Usage | ||
|
|
||
| ```python | ||
| from crewai_tools import CreduentVerificationTool | ||
|
|
||
| tool = CreduentVerificationTool() | ||
| result = tool.run(agent_uri="agent://assistant.dev/agent") | ||
| print(result) | ||
| ``` | ||
|
|
||
| ## Protocol Specifications | ||
|
|
||
| Creduent performs local Ed25519 signature checks and canonical JSON (JCS RFC 8785) verification on agent URIs (`agent://<namespace>/<name>`) in under 5ms. |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| # Creduent Verification Tool |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| import logging | ||
|
|
||
| from crewai.tools import BaseTool | ||
| from pydantic import BaseModel, Field | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class CreduentVerificationSchema(BaseModel): | ||
| """Input schema for Creduent verification tool. | ||
|
|
||
| Attributes: | ||
| agent_uri: Target agent URI to verify, formatted as agent://<namespace>/<name>. | ||
| """ | ||
|
|
||
| agent_uri: str = Field( | ||
| ..., | ||
| description="Target agent URI to verify, formatted as agent://<namespace>/<name>", | ||
| ) | ||
|
|
||
|
|
||
| class CreduentVerificationTool(BaseTool): | ||
| """Tool for verifying agent identity using the Creduent protocol. | ||
|
|
||
| Performs local cryptographic signature verification (Ed25519) and canonical JCS RFC 8785 | ||
| attestation checks on target agent URIs. | ||
| """ | ||
|
|
||
| name: str = "Creduent Agent Identity Verification" | ||
| description: str = ( | ||
| "Verifies the cryptographic identity, signature, and attestations of a target AI agent " | ||
| "using the Creduent open protocol before delegating tasks." | ||
| ) | ||
| args_schema: type[BaseModel] = CreduentVerificationSchema | ||
| strict: bool = True | ||
|
|
||
| def _run(self, agent_uri: str) -> str: | ||
| """Execute verification of the target agent URI. | ||
|
|
||
| Args: | ||
| agent_uri: Target agent URI formatted as agent://<namespace>/<name>. | ||
|
|
||
| Returns: | ||
| String containing verification status message or error details. | ||
|
|
||
| Raises: | ||
| ValueError: If strict mode is enabled and verification fails. | ||
| """ | ||
| try: | ||
| from creduent.verify import verify | ||
| except ImportError: | ||
| return ( | ||
| "Error: creduent package is not installed. " | ||
| "Install it using: pip install creduent" | ||
| ) | ||
|
|
||
| logger.info(f"Verifying target agent identity: {agent_uri}") | ||
| try: | ||
| result = verify(agent_uri) | ||
| if result.valid: | ||
| return ( | ||
| f"Verification SUCCESS for {agent_uri}. " | ||
| "Agent identity and cryptographic attestations are trusted." | ||
| ) | ||
| error_msg = f"Verification FAILED for {agent_uri}: {result.error}" | ||
| logger.warning(error_msg) | ||
| if self.strict: | ||
| raise ValueError(error_msg) | ||
| return error_msg | ||
| except Exception as err: | ||
| error_msg = f"Verification failure for {agent_uri}: {str(err)}" | ||
| logger.error(error_msg) | ||
| if self.strict: | ||
| raise ValueError(error_msg) from err | ||
| return error_msg | ||
|
Comment on lines
+58
to
+75
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win Fix double-wrapped error message on strict-mode failed verification. The The final message becomes Separate the "call 🐛 Proposed fix to separate exception handling from failed-result handling- logger.info(f"Verifying target agent identity: {agent_uri}")
- try:
- result = verify(agent_uri)
- if result.valid:
- return (
- f"Verification SUCCESS for {agent_uri}. "
- "Agent identity and cryptographic attestations are trusted."
- )
- error_msg = f"Verification FAILED for {agent_uri}: {result.error}"
- logger.warning(error_msg)
- if self.strict:
- raise ValueError(error_msg)
- return error_msg
- except Exception as err:
- error_msg = f"Verification failure for {agent_uri}: {str(err)}"
- logger.error(error_msg)
- if self.strict:
- raise ValueError(error_msg) from err
- return error_msg
+ logger.info(f"Verifying target agent identity: {agent_uri}")
+ try:
+ result = verify(agent_uri)
+ except Exception as err:
+ error_msg = f"Verification failure for {agent_uri}: {str(err)}"
+ logger.error(error_msg)
+ if self.strict:
+ raise ValueError(error_msg) from err
+ return error_msg
+
+ if result.valid:
+ return (
+ f"Verification SUCCESS for {agent_uri}. "
+ "Agent identity and cryptographic attestations are trusted."
+ )
+ error_msg = f"Verification FAILED for {agent_uri}: {result.error}"
+ logger.warning(error_msg)
+ if self.strict:
+ raise ValueError(error_msg)
+ return error_msg🤖 Prompt for AI Agents |
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| from unittest.mock import MagicMock, patch | ||
|
|
||
| import pytest | ||
| from crewai_tools.tools.creduent_verification_tool.creduent_verification_tool import ( | ||
| CreduentVerificationSchema, | ||
| CreduentVerificationTool, | ||
| ) | ||
|
|
||
|
|
||
| def test_schema_validation() -> None: | ||
| """Test schema validation for agent URI parameter.""" | ||
| schema = CreduentVerificationSchema(agent_uri="agent://assistant.dev/planner") | ||
| assert schema.agent_uri == "agent://assistant.dev/planner" | ||
|
|
||
|
|
||
| def test_verification_tool_initialization() -> None: | ||
| """Test tool attributes and default strict configuration.""" | ||
| tool = CreduentVerificationTool() | ||
| assert tool.name == "Creduent Agent Identity Verification" | ||
| assert tool.strict is True | ||
|
|
||
|
|
||
| @patch("creduent.verify.verify") | ||
| def test_successful_verification(mock_verify: MagicMock) -> None: | ||
| """Test successful verification flow when protocol returns valid result.""" | ||
| mock_result = MagicMock() | ||
| mock_result.valid = True | ||
| mock_verify.return_value = mock_result | ||
|
|
||
| tool = CreduentVerificationTool(strict=False) | ||
| output = tool._run(agent_uri="agent://assistant.dev/planner") | ||
|
|
||
| assert "Verification SUCCESS" in output | ||
| mock_verify.assert_called_once_with("agent://assistant.dev/planner") | ||
|
|
||
|
|
||
| @patch("creduent.verify.verify") | ||
| def test_failed_verification_strict(mock_verify: MagicMock) -> None: | ||
| """Test verification failure when strict mode is active.""" | ||
| mock_result = MagicMock() | ||
| mock_result.valid = False | ||
| mock_result.error = "Invalid signature" | ||
| mock_verify.return_value = mock_result | ||
|
|
||
| tool = CreduentVerificationTool(strict=True) | ||
|
|
||
| with pytest.raises(ValueError) as exc_info: | ||
| tool._run(agent_uri="agent://untrusted.dev/hacker") | ||
|
|
||
| assert "Invalid signature" in str(exc_info.value) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
Repository: crewAIInc/crewAI
Length of output: 215
🏁 Script executed:
Repository: crewAIInc/crewAI
Length of output: 10931
🏁 Script executed:
Repository: crewAIInc/crewAI
Length of output: 10931
Wire
creduentas an optional dependency.CreduentVerificationToolimportscreduent.verify.verify, but it does not setpackage_dependencies: list[str] = ["creduent"], andlib/crewai-tools/pyproject.tomlonly declares extras forscrapfly-sdkandsqlalchemy. Add acreduentextra and reference it in the tool docs so installers can opt in.test_creduent_verification_tool.pypatchescreduent.verify.verify, so the test/dev environment must makecreduentimportable even if the installed package lacks the actual package. Add a test-only dependency/fixture boundary or avoid patching the missing upstream library directly.🤖 Prompt for AI Agents